feat: sync with exam service on course publish

cms/djangoapps/contentstore/exams.py

@@ -0,0 +1,136 @@
+"""
+Code related to working with the exam service
+"""
+
+import json
+import logging
+
+import requests
+from django.conf import settings
+from django.contrib.auth import get_user_model
+from edx_rest_api_client.auth import SuppliedJwtAuth
+
+from openedx.core.djangoapps.course_apps.toggles import exams_ida_enabled
+from openedx.core.djangoapps.oauth_dispatch.jwt import create_jwt_for_user
+from xmodule.modulestore.django import modulestore
+from xmodule.modulestore.exceptions import ItemNotFoundError
+
+from .views.helpers import is_item_in_course_tree
+
+log = logging.getLogger(__name__)
+User = get_user_model()
+
+
+def register_exams(course_key):
+    """
+    This is typically called on a course published signal. The course is examined for sequences
+    that are marked as timed exams. Then these are registered with the exams service.
+    Likewise, if formerly registered exams are not included in the payload they will
+    be marked inactive by the exam service.
+    """
+    if not settings.FEATURES.get('ENABLE_SPECIAL_EXAMS') or not exams_ida_enabled(course_key):
+        # if feature is not enabled then do a quick exit
+        return
+
+    course = modulestore().get_course(course_key)
+    if course is None:
+        raise ItemNotFoundError("Course {} does not exist", str(course_key))  # lint-amnesty, pylint: disable=raising-format-tuple
+
+    # get all sequences, since they can be marked as timed/proctored exams
+    _timed_exams = modulestore().get_items(
+        course_key,
+        qualifiers={
+            'category': 'sequential',
+        },
+        settings={
+            'is_time_limited': True,
+        }
+    )
+
+    # filter out any potential dangling sequences
+    timed_exams = [
+        timed_exam
+        for timed_exam in _timed_exams
+        if is_item_in_course_tree(timed_exam)
+    ]
+
+    exams_list = []
+    locations = []
+    for timed_exam in timed_exams:
+        location = str(timed_exam.location)
+        msg = (
+            'Found {location} as an exam in course structure.'.format(
+                location=location
+            )
+        )
+        log.info(msg)
+        locations.append(location)
+
+        exam_type = get_exam_type(
+            timed_exam.is_proctored_exam,
+            timed_exam.is_practice_exam,
+            timed_exam.is_onboarding_exam
+        )
+        exams_list.append({
+            'course_id': str(course_key),
+            'content_id': str(timed_exam.location),
+            'exam_name': timed_exam.display_name,
+            'time_limit_mins': timed_exam.default_time_limit_minutes,
+            'due_date': timed_exam.due if not course.self_paced else None,
+            'exam_type': exam_type,
+            'is_active': True,
+            'hide_after_due': timed_exam.hide_after_due,
+            # backend is only required for continued edx-proctoring support
+            'backend': course.proctoring_provider,
+        })
+
+    try:
+        _patch_course_exams(exams_list, str(course_key))
+        log.info(f'Successfully registered {locations} with exam service')
+    # pylint: disable=broad-except
+    except Exception as ex:
+        log.exception('Failed to register exams with exam API', exc_info=True)
+        raise ex
+
+
+def get_exam_type(is_proctored, is_practice, is_onboarding):
+    """
+    Get the exam type string based on the proctored, practice and onboarding
+    attributes.
+    """
+    if is_proctored:
+        if is_onboarding:
+            exam_type = 'onboarding'
+        elif is_practice:
+            exam_type = 'practice_proctored'
+        else:
+            exam_type = 'proctored'
+    else:
+        exam_type = 'timed'
+
+    return exam_type
+
+
+def _get_exams_api_client():
+    """
+    Returns an API client which can be used to make Exams API requests.
+    """
+    user = User.objects.get(username=settings.EXAMS_SERVICE_USERNAME)
+    jwt = create_jwt_for_user(user)
+    client = requests.Session()
+    client.auth = SuppliedJwtAuth(jwt)
+
+    return client
+
+
+def _patch_course_exams(exams_list, course_id):
+    """
+    Make a PATCH request to update course exams
+    """
+    url = f'{settings.EXAMS_SERVICE_URL}/exams/course_id/{course_id}/'
+    api_client = _get_exams_api_client()
+
+    response = api_client.patch(url, data=json.dumps(exams_list))
+    response.raise_for_status()
+    response = response.json()
+    return response

cms/djangoapps/contentstore/proctoring.py

@@ -1,5 +1,8 @@
 """
 Code related to the handling of Proctored Exams in Studio
+using the edx-proctoring plugin.
+
+Courses using the exam IDA are handled by ./exams.py
 """
 
 

cms/djangoapps/contentstore/tasks.py

@@ -51,6 +51,7 @@
 from common.djangoapps.student.auth import has_course_author_access
 from common.djangoapps.util.monitoring import monitor_import_failure
 from openedx.core.djangoapps.content.learning_sequences.api import key_supports_outlines
+from openedx.core.djangoapps.course_apps.toggles import exams_ida_enabled
 from openedx.core.djangoapps.discussions.tasks import update_unit_discussion_state_from_discussion_blocks
 from openedx.core.djangoapps.embargo.models import CountryAccessRule, RestrictedCourse
 from openedx.core.lib.extract_tar import safetar_extractall
@@ -242,13 +243,17 @@ def update_special_exams_and_publish(course_key_str):
     on_course_publish expects that the edx-proctoring subsystem has been refreshed
     before being executed, so both functions are called here synchronously.
     """
-    from cms.djangoapps.contentstore.proctoring import register_special_exams
+    from cms.djangoapps.contentstore.exams import register_exams
+    from cms.djangoapps.contentstore.proctoring import register_special_exams as register_exams_legacy
     from openedx.core.djangoapps.credit.signals import on_course_publish
 
     course_key = CourseKey.from_string(course_key_str)
     LOGGER.info('Attempting to register exams for course %s', course_key_str)
+
+    # Call the appropriate handler for either the exams IDA or the edx-proctoring plugin
+    register_exams_handler = register_exams if exams_ida_enabled(course_key) else register_exams_legacy
     try:
-        register_special_exams(course_key)
+        register_exams_handler(course_key)
         LOGGER.info('Successfully registered exams for course %s', course_key_str)
     # pylint: disable=broad-except
     except Exception as exception:

cms/djangoapps/contentstore/tests/test_exams.py

@@ -0,0 +1,173 @@
+"""
+Test the exams service integration into Studio
+"""
+from datetime import datetime, timedelta
+from unittest.mock import patch
+
+import ddt
+from django.conf import settings
+from edx_toggles.toggles.testutils import override_waffle_flag
+from pytz import UTC
+
+from cms.djangoapps.contentstore.signals.handlers import listen_for_course_publish
+from openedx.core.djangoapps.course_apps.toggles import EXAMS_IDA
+from xmodule.modulestore.tests.django_utils import TEST_DATA_MONGO_AMNESTY_MODULESTORE, ModuleStoreTestCase
+from xmodule.modulestore.tests.factories import CourseFactory, ItemFactory
+
+
+@ddt.ddt
+@override_waffle_flag(EXAMS_IDA, active=True)
+@patch.dict('django.conf.settings.FEATURES', {'ENABLE_SPECIAL_EXAMS': True})
+@patch('cms.djangoapps.contentstore.exams._patch_course_exams')
+class TestExamService(ModuleStoreTestCase):
+    """
+    Test for syncing exams to the exam service
+    """
+    MODULESTORE = TEST_DATA_MONGO_AMNESTY_MODULESTORE
+
+    def setUp(self):
+        """
+        Initial data setup
+        """
+        super().setUp()
+
+        self.course = CourseFactory.create(
+            org='edX',
+            course='900',
+            run='test_run',
+            enable_proctored_exams=True,
+            proctoring_provider='null',
+        )
+        self.chapter = ItemFactory.create(parent=self.course, category='chapter', display_name='Test Section')
+        self.course_key = str(self.course.id)
+
+        # create one non-exam sequence
+        chapter2 = ItemFactory.create(parent=self.course, category='chapter', display_name='Test Homework')
+        ItemFactory.create(
+            parent=chapter2,
+            category='sequential',
+            display_name='Homework 1',
+            graded=True,
+            is_time_limited=False,
+            due=datetime.now(UTC) + timedelta(minutes=60),
+        )
+
+    def _get_exams_url(self, course_id):
+        return f'{settings.EXAMS_SERVICE_URL}/exams/course_id/{course_id}/'
+
+    @ddt.data(
+        (False, False, False, 'timed'),
+        (True, False, False, 'proctored'),
+        (True, True, False, 'practice_proctored'),
+        (True, True, True, 'onboarding'),
+    )
+    @ddt.unpack
+    def test_publishing_exam(self, is_proctored_exam, is_practice_exam,
+                             is_onboarding_exam, expected_type, mock_patch_course_exams):
+        """
+        When a course is published it will register all exams sections with the exams service
+        """
+        default_time_limit_minutes = 10
+
+        sequence = ItemFactory.create(
+            parent=self.chapter,
+            category='sequential',
+            display_name='Test Proctored Exam',
+            graded=True,
+            is_time_limited=True,
+            default_time_limit_minutes=default_time_limit_minutes,
+            is_proctored_exam=is_proctored_exam,
+            is_practice_exam=is_practice_exam,
+            due=datetime.now(UTC) + timedelta(minutes=default_time_limit_minutes + 1),
+            hide_after_due=True,
+            is_onboarding_exam=is_onboarding_exam,
+        )
+
+        expected_exams = [{
+            'course_id': self.course_key,
+            'content_id': str(sequence.location),
+            'exam_name': sequence.display_name,
+            'time_limit_mins': sequence.default_time_limit_minutes,
+            'due_date': sequence.due,
+            'exam_type': expected_type,
+            'is_active': True,
+            'hide_after_due': True,
+            # backend is only required for edx-proctoring support edx-exams will maintain LTI backends
+            'backend': 'null',
+        }]
+        listen_for_course_publish(self, self.course.id)
+        mock_patch_course_exams.assert_called_once_with(expected_exams, self.course_key)
+
+    def test_publish_no_exam(self, mock_patch_course_exams):
+        """
+        Exam service is called with an empty list if there are no exam sections.
+        This will deactivate any currently active exams
+        """
+        listen_for_course_publish(self, self.course.id)
+        mock_patch_course_exams.assert_called_once_with([], self.course_key)
+
+    def test_dangling_exam(self, mock_patch_course_exams):
+        """
+        Make sure we filter out all dangling items
+        """
+        ItemFactory.create(
+            parent=self.chapter,
+            category='sequential',
+            display_name='Test Proctored Exam',
+            graded=True,
+            is_time_limited=True,
+            default_time_limit_minutes=10,
+            is_proctored_exam=True,
+            hide_after_due=False,
+        )
+        self.store.delete_item(self.chapter.location, self.user.id)
+
+        listen_for_course_publish(self, self.course.id)
+        mock_patch_course_exams.assert_called_once_with([], self.course_key)
+
+    @patch.dict('django.conf.settings.FEATURES', {'ENABLE_SPECIAL_EXAMS': False})
+    def test_feature_flag_off(self, mock_patch_course_exams):
+        """
+        Make sure the feature flag is honored
+        """
+        ItemFactory.create(
+            parent=self.chapter,
+            category='sequential',
+            display_name='Test Proctored Exam',
+            graded=True,
+            is_time_limited=True,
+            default_time_limit_minutes=10,
+            is_proctored_exam=True,
+            hide_after_due=False,
+        )
+
+        listen_for_course_publish(self, self.course.id)
+        mock_patch_course_exams.assert_not_called()
+
+    def test_self_paced_no_due_dates(self, mock_patch_course_exams):
+        self.course.self_paced = True
+        self.course = self.update_course(self.course, 1)
+        ItemFactory.create(
+            parent=self.chapter,
+            category='sequential',
+            display_name='Test Proctored Exam',
+            graded=True,
+            is_time_limited=True,
+            default_time_limit_minutes=60,
+            is_proctored_exam=False,
+            is_practice_exam=False,
+            due=datetime.now(UTC) + timedelta(minutes=60),
+            hide_after_due=True,
+            is_onboarding_exam=False,
+        )
+        listen_for_course_publish(self, self.course.id)
+        called_exams, called_course = mock_patch_course_exams.call_args[0]
+        assert called_exams[0]['due_date'] is None
+
+        # now switch to instructor paced
+        # the exam will be updated with a due date
+        self.course.self_paced = False
+        self.course = self.update_course(self.course, 1)
+        listen_for_course_publish(self, self.course.id)
+        called_exams, called_course = mock_patch_course_exams.call_args[0]
+        assert called_exams[0]['due_date'] is not None