2u/optimizer test happy path

cms/djangoapps/contentstore/tasks.py

@@ -1255,15 +1255,21 @@ def _filter_by_status(results):
 
     return filtered_results, retry_list
 
-@shared_task(base=CourseLinkCheckTask, bind=True)
-def check_broken_links(self, user_id, course_key_string, language):
-    """
-    Checks for broken links in a course. Store the results in a file.
-    """
-    user = _validate_user(self, user_id, language)
+def _save_broken_links_file(artifact, file_to_save):
+    artifact.file.save(name=os.path.basename(file_to_save.name), content=File(file_to_save))
+    artifact.save()
+    return True
 
-    self.status.set_state('Scanning')
+def _write_broken_links_to_file(broken_or_locked_urls, broken_links_file):
+    with open(broken_links_file.name, 'w') as file:
+        json.dump(broken_or_locked_urls, file, indent=4)
+
+def _check_broken_links(task_instance, user_id, course_key_string, language):
+    user = _validate_user(task_instance, user_id, language)
+
+    task_instance.status.set_state('Scanning')
     course_key = CourseKey.from_string(course_key_string)
+
     url_list = _scan_course_for_links(course_key)
     validated_url_list = asyncio.run(_validate_urls_access_in_batches(url_list, course_key, batch_size=100))
     broken_or_locked_urls, retry_list = _filter_by_status(validated_url_list)
@@ -1273,7 +1279,7 @@ def check_broken_links(self, user_id, course_key_string, language):
         broken_or_locked_urls.extend(retry_results)
 
     try:
-        self.status.increment_completed_steps()
+        task_instance.status.increment_completed_steps()
 
         file_name = str(course_key)
         broken_links_file = NamedTemporaryFile(prefix=file_name + '.', suffix='.json')
@@ -1282,13 +1288,22 @@ def check_broken_links(self, user_id, course_key_string, language):
         with open(broken_links_file.name, 'w') as file:
             json.dump(broken_or_locked_urls, file, indent=4)
 
-        artifact = UserTaskArtifact(status=self.status, name='BrokenLinks')
-        artifact.file.save(name=os.path.basename(broken_links_file.name), content=File(broken_links_file))
-        artifact.save()
+        _write_broken_links_to_file(broken_or_locked_urls, broken_links_file)
+
+        artifact = UserTaskArtifact(status=task_instance.status, name='BrokenLinks')
+        _save_broken_links_file(artifact, broken_links_file)
 
     # catch all exceptions so we can record useful error messages
     except Exception as e:  # pylint: disable=broad-except
         LOGGER.exception('Error checking links for course %s', course_key, exc_info=True)
-        if self.status.state != UserTaskStatus.FAILED:
-            self.status.fail({'raw_error_msg': str(e)})
+        if task_instance.status.state != UserTaskStatus.FAILED:
+            task_instance.status.fail({'raw_error_msg': str(e)})
         return
+
+
+@shared_task(base=CourseLinkCheckTask, bind=True)
+def check_broken_links(self, user_id, course_key_string, language):
+    """
+    Checks for broken links in a course. Store the results in a file.
+    """
+    return _check_broken_links(self, user_id, course_key_string, language)

cms/djangoapps/contentstore/tests/test_tasks.py

@@ -5,13 +5,16 @@
 
 import copy
 import json
+import os
+from tempfile import NamedTemporaryFile
 from unittest import mock, TestCase
 from uuid import uuid4
 
 import pytest as pytest
 from django.conf import settings
 from django.contrib.auth.models import User  # lint-amnesty, pylint: disable=imported-auth-user
 from django.test.utils import override_settings
+from django.core.files import File
 from edx_toggles.toggles.testutils import override_waffle_flag
 from opaque_keys.edx.locator import CourseLocator
 from organizations.models import OrganizationCourse
@@ -22,7 +25,8 @@
     export_olx,
     update_special_exams_and_publish,
     rerun_course,
-    _convert_to_standard_url
+    _convert_to_standard_url,
+    _check_broken_links
 )
 from cms.djangoapps.contentstore.tests.test_libraries import LibraryTestCase
 from cms.djangoapps.contentstore.tests.utils import CourseTestCase
@@ -31,7 +35,8 @@
 from openedx.core.djangoapps.course_apps.toggles import EXAMS_IDA
 from openedx.core.djangoapps.embargo.models import Country, CountryAccessRule, RestrictedCourse
 from xmodule.modulestore.django import modulestore  # lint-amnesty, pylint: disable=wrong-import-order
-from xmodule.modulestore.tests.django_utils import TEST_DATA_SPLIT_MODULESTORE
+from xmodule.modulestore.tests.django_utils import TEST_DATA_SPLIT_MODULESTORE, ModuleStoreTestCase
+from celery import Task
 
 TEST_DATA_CONTENTSTORE = copy.deepcopy(settings.CONTENTSTORE)
 TEST_DATA_CONTENTSTORE['DOC_STORE_CONFIG']['db'] = 'test_xcontent_%s' % uuid4().hex
@@ -207,7 +212,61 @@ def test_register_exams_failure(self, _mock_register_exams_proctoring, _mock_reg
             course_publish.assert_called()
 
 
-class CourseOptimizerTestCase(TestCase):
+class MockCourseLinkCheckTask(Task):
+    def __init__(self):
+        self.status = mock.Mock()
+
+
+class CheckBrokenLinksTaskTest(ModuleStoreTestCase):
+    def setUp(self):
+        super().setUp()
+        self.mock_urls = [
+            ["block-v1:edX+DemoX+Demo_Course+type@vertical+block@1", "http://example.com/valid"],
+            ["block-v1:edX+DemoX+Demo_Course+type@vertical+block@2", "http://example.com/invalid"]
+        ]
+        self.expected_file_contents = [
+            ['block-v1:edX+DemoX+Demo_Course+type@vertical+block@1', 'http://example.com/valid', False],
+            ['block-v1:edX+DemoX+Demo_Course+type@vertical+block@2', 'http://example.com/invalid', False]
+        ]
+
+    @mock.patch('cms.djangoapps.contentstore.tasks.UserTaskArtifact', autospec=True)
+    @mock.patch('cms.djangoapps.contentstore.tasks.UserTaskStatus', autospec=True)
+    @mock.patch('cms.djangoapps.contentstore.tasks._scan_course_for_links')
+    @mock.patch('cms.djangoapps.contentstore.tasks._save_broken_links_file', autospec=True)
+    @mock.patch('cms.djangoapps.contentstore.tasks._write_broken_links_to_file', autospec=True)
+    def test_check_broken_links_stores_broken_and_locked_urls(
+        self,
+        mock_write_broken_links_to_file,
+        mock_save_broken_links_file,
+        mock_scan_course_for_links,
+        _mock_user_task_status,
+        mock_user_task_artifact
+    ):
+        '''
+        The test should verify that the check_broken_links task correctly
+        identifies and stores broken or locked URLs in the course.
+        The expected behavior is that the task scans the course,
+        validates the URLs, filters the results, and stores them in a
+        JSON file.
+        '''
+        # Arrange
+        mock_user = UserFactory.create(username='student', password='password')
+        mock_course_key_string = "course-v1:edX+DemoX+Demo_Course"
+        mock_task = MockCourseLinkCheckTask()
+        mock_scan_course_for_links.return_value = self.mock_urls
+
+        # Act
+        _check_broken_links(mock_task, mock_user.id, mock_course_key_string, 'en')  # pylint: disable=no-value-for-parameter
+
+        # Assert
+        ### Check that UserTaskArtifact was called with the correct arguments
+        mock_user_task_artifact.assert_called_once_with(status=mock.ANY, name='BrokenLinks')
+
+        ### Check that the correct links are written to the file
+        mock_write_broken_links_to_file.assert_called_once_with(self.expected_file_contents, mock.ANY)
+
+        ### Check that _save_broken_links_file was called with the correct arguments
+        mock_save_broken_links_file.assert_called_once_with(mock_user_task_artifact.return_value, mock.ANY)
 
 
     def test_user_does_not_exist_raises_exception(self):
@@ -265,4 +324,4 @@ def test_max_number_of_retries_is_respected(self):
         raise NotImplementedError
 
     def test_scan_generates_file_named_by_course_key(self):
-        raise NotImplementedErro
+        raise NotImplementedError