Andal.LND/edx-platform
6
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: move rejected exam event handlers

Browse Source
This commit is contained in:
Varsha Menon
2024-01-08 15:10:35 -05:00
parent 43747ee8f0
commit 0bf676c830
6 changed files with 104 additions and 120 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
lms/djangoapps/certificates/api.py
View File

@@ -929,9 +929,9 @@ def invalidate_certificate(user_id, course_key_or_id, source):
Invalidate the user certificate in a given course if it exists and the user is not on the allowlist for this
course run.
This function is called in services.py and handlers.py within the certificates folder. As of now,
This function is called in services.py and signals.py within the certificates folder. As of now,
The call in services.py occurs when an exam attempt is rejected in the legacy exams backend, edx-proctoring.
The call in handlers.py is occurs when an exam attempt is rejected in the newer exams backend, edx-exams.
The call in signals.py is occurs when an exam attempt is rejected in the newer exams backend, edx-exams.
"""
course_key = _get_key(course_key_or_id, CourseKey)
if _is_on_certificate_allowlist(user_id, course_key):

1
lms/djangoapps/certificates/apps.py
View File

@@ -23,7 +23,6 @@ class CertificatesConfig(AppConfig):
# Can't import models at module level in AppConfigs, and models get
# included from the signal handlers
from lms.djangoapps.certificates import signals # pylint: disable=unused-import
from lms.djangoapps.certificates import handlers # pylint: disable=unused-import
if settings.FEATURES.get('ENABLE_SPECIAL_EXAMS'):
from lms.djangoapps.certificates.services import CertificateService
set_runtime_service('certificates', CertificateService())

28
lms/djangoapps/certificates/handlers.py
View File

@@ -1,28 +0,0 @@
"""
Handlers for credits
"""
import logging
from django.contrib.auth import get_user_model
from django.dispatch import receiver
from openedx_events.learning.signals import EXAM_ATTEMPT_REJECTED
from lms.djangoapps.certificates.api import invalidate_certificate
User = get_user_model()
log = logging.getLogger(__name__)
@receiver(EXAM_ATTEMPT_REJECTED)
def handle_exam_attempt_rejected_event(sender, signal, **kwargs):
"""
Consume `EXAM_ATTEMPT_REJECTED` events from the event bus.
Pass the received data to invalidate_certificate in the services.py file in this folder.
"""
event_data = kwargs.get('exam_attempt')
user_data = event_data.student_user
course_key = event_data.course_key
# Note that the course_key is the same as the course_key_or_id, and is being passed in as the course_key param
invalidate_certificate(user_data.id, course_key, source='exam_event')

23
lms/djangoapps/certificates/signals.py
View File

@@ -4,6 +4,7 @@ Signal handler for enabling/disabling self-generated certificates based on the c
import logging
from django.contrib.auth import get_user_model
from django.db.models.signals import post_save
from django.dispatch import receiver
@@ -22,7 +23,10 @@ from lms.djangoapps.certificates.models import (
CertificateStatuses,
GeneratedCertificate
)
from lms.djangoapps.certificates.api import auto_certificate_generation_enabled
from lms.djangoapps.certificates.api import (
auto_certificate_generation_enabled,
invalidate_certificate
)
from lms.djangoapps.verify_student.services import IDVerificationService
from openedx.core.djangoapps.content.course_overviews.signals import COURSE_PACING_CHANGED
from openedx.core.djangoapps.signals.signals import (
@@ -30,6 +34,9 @@ from openedx.core.djangoapps.signals.signals import (
COURSE_GRADE_NOW_PASSED,
LEARNER_NOW_VERIFIED
)
from openedx_events.learning.signals import EXAM_ATTEMPT_REJECTED
User = get_user_model()
log = logging.getLogger(__name__)
@@ -156,3 +163,17 @@ def _listen_for_enrollment_mode_change(sender, user, course_key, mode, **kwargs)
course_key,
)
return False
@receiver(EXAM_ATTEMPT_REJECTED)
def handle_exam_attempt_rejected_event(sender, signal, **kwargs):
"""
Consume `EXAM_ATTEMPT_REJECTED` events from the event bus.
Pass the received data to invalidate_certificate in the services.py file in this folder.
"""
event_data = kwargs.get('exam_attempt')
user_data = event_data.student_user
course_key = event_data.course_key
# Note that the course_key is the same as the course_key_or_id, and is being passed in as the course_key param
invalidate_certificate(user_data.id, course_key, source='exam_event')

87
lms/djangoapps/certificates/tests/test_handlers.py
View File

@@ -1,87 +0,0 @@
"""
Unit tests for certificates signals
"""
from datetime import datetime, timezone
from unittest import mock
from uuid import uuid4
from django.test import TestCase
from opaque_keys.edx.keys import CourseKey, UsageKey
from openedx_events.data import EventsMetadata
from openedx_events.learning.data import ExamAttemptData, UserData, UserPersonalData
from openedx_events.learning.signals import EXAM_ATTEMPT_REJECTED
from common.djangoapps.student.tests.factories import UserFactory
from lms.djangoapps.certificates.handlers import handle_exam_attempt_rejected_event
class ExamCompletionEventBusTests(TestCase):
"""
Tests completion events from the event bus.
"""
@classmethod
def setUpClass(cls):
super().setUpClass()
cls.course_key = CourseKey.from_string('course-v1:edX+TestX+Test_Course')
cls.subsection_id = 'block-v1:edX+TestX+Test_Course+type@sequential+block@subsection'
cls.usage_key = UsageKey.from_string(cls.subsection_id)
cls.student_user = UserFactory(
username='student_user',
)
@staticmethod
def _get_exam_event_data(student_user, course_key, usage_key, exam_type, requesting_user=None):
""" create ExamAttemptData object for exam based event """
if requesting_user:
requesting_user_data = UserData(
id=requesting_user.id,
is_active=True,
pii=None
)
else:
requesting_user_data = None
return ExamAttemptData(
student_user=UserData(
id=student_user.id,
is_active=True,
pii=UserPersonalData(
username=student_user.username,
email=student_user.email,
),
),
course_key=course_key,
usage_key=usage_key,
requesting_user=requesting_user_data,
exam_type=exam_type,
)
@staticmethod
def _get_exam_event_metadata(event_signal):
""" create metadata object for event """
return EventsMetadata(
event_type=event_signal.event_type,
id=uuid4(),
minorversion=0,
source='openedx/lms/web',
sourcehost='lms.test',
time=datetime.now(timezone.utc)
)
@mock.patch('lms.djangoapps.certificates.handlers.invalidate_certificate')
def test_exam_attempt_rejected_event(self, mock_api_function):
"""
Assert that CertificateService api's invalidate_certificate is called upon consuming the event
"""
exam_event_data = self._get_exam_event_data(self.student_user,
self.course_key,
self.usage_key,
exam_type='proctored')
event_metadata = self._get_exam_event_metadata(EXAM_ATTEMPT_REJECTED)
event_kwargs = {
'exam_attempt': exam_event_data,
'metadata': event_metadata
}
handle_exam_attempt_rejected_event(None, EXAM_ATTEMPT_REJECTED, **event_kwargs)
mock_api_function.assert_called_once_with(self.student_user.id, self.course_key, source='exam_event')

81
lms/djangoapps/certificates/tests/test_signals.py
View File

@@ -2,11 +2,17 @@
Unit tests for enabling self-generated certificates for self-paced courses
and disabling for instructor-paced courses.
"""
from datetime import datetime, timezone
from unittest import mock
from uuid import uuid4
import ddt
from django.test import TestCase
from edx_toggles.toggles.testutils import override_waffle_flag, override_waffle_switch
from opaque_keys.edx.keys import CourseKey, UsageKey
from openedx_events.data import EventsMetadata
from openedx_events.learning.data import ExamAttemptData, UserData, UserPersonalData
from openedx_events.learning.signals import EXAM_ATTEMPT_REJECTED
from xmodule.modulestore.tests.django_utils import ModuleStoreTestCase
from xmodule.modulestore.tests.factories import CourseFactory
@@ -18,6 +24,7 @@ from lms.djangoapps.certificates.models import (
CertificateGenerationConfiguration,
GeneratedCertificate
)
from lms.djangoapps.certificates.signals import handle_exam_attempt_rejected_event
from lms.djangoapps.certificates.tests.factories import CertificateAllowlistFactory, GeneratedCertificateFactory
from lms.djangoapps.grades.course_grade_factory import CourseGradeFactory
from lms.djangoapps.grades.tests.utils import mock_passing_grade
@@ -433,3 +440,75 @@ class EnrollmentModeChangeCertsTest(ModuleStoreTestCase):
) as mock_allowlist_task:
self.verified_enrollment.change_mode('audit')
mock_allowlist_task.assert_not_called()
class ExamCompletionEventBusTests(TestCase):
"""
Tests completion events from the event bus.
"""
@classmethod
def setUpClass(cls):
super().setUpClass()
cls.course_key = CourseKey.from_string('course-v1:edX+TestX+Test_Course')
cls.subsection_id = 'block-v1:edX+TestX+Test_Course+type@sequential+block@subsection'
cls.usage_key = UsageKey.from_string(cls.subsection_id)
cls.student_user = UserFactory(
username='student_user',
)
@staticmethod
def _get_exam_event_data(student_user, course_key, usage_key, exam_type, requesting_user=None):
""" create ExamAttemptData object for exam based event """
if requesting_user:
requesting_user_data = UserData(
id=requesting_user.id,
is_active=True,
pii=None
)
else:
requesting_user_data = None
return ExamAttemptData(
student_user=UserData(
id=student_user.id,
is_active=True,
pii=UserPersonalData(
username=student_user.username,
email=student_user.email,
),
),
course_key=course_key,
usage_key=usage_key,
requesting_user=requesting_user_data,
exam_type=exam_type,
)
@staticmethod
def _get_exam_event_metadata(event_signal):
""" create metadata object for event """
return EventsMetadata(
event_type=event_signal.event_type,
id=uuid4(),
minorversion=0,
source='openedx/lms/web',
sourcehost='lms.test',
time=datetime.now(timezone.utc)
)
@mock.patch('lms.djangoapps.certificates.signals.invalidate_certificate')
def test_exam_attempt_rejected_event(self, mock_api_function):
"""
Assert that CertificateService api's invalidate_certificate is called upon consuming the event
"""
exam_event_data = self._get_exam_event_data(self.student_user,
self.course_key,
self.usage_key,
exam_type='proctored')
event_metadata = self._get_exam_event_metadata(EXAM_ATTEMPT_REJECTED)
event_kwargs = {
'exam_attempt': exam_event_data,
'metadata': event_metadata
}
handle_exam_attempt_rejected_event(None, EXAM_ATTEMPT_REJECTED, **event_kwargs)
mock_api_function.assert_called_once_with(self.student_user.id, self.course_key, source='exam_event')