diff --git a/common/djangoapps/student/tests/test_handlers.py b/common/djangoapps/student/tests/test_handlers.py deleted file mode 100644 index bd8d6a9baf47..000000000000 --- a/common/djangoapps/student/tests/test_handlers.py +++ /dev/null @@ -1,67 +0,0 @@ -""" -Unit tests for event bus tests for course unenrollments -""" - -import unittest -from datetime import datetime, timezone -from unittest import mock -from uuid import uuid4 - -from django.test.utils import override_settings -from common.djangoapps.student.handlers import course_unenrollment_receiver -from common.djangoapps.student.tests.factories import ( - UserFactory, - CourseEnrollmentFactory, -) - -from openedx_events.data import EventsMetadata -from openedx_events.learning.signals import COURSE_UNENROLLMENT_COMPLETED -from pytest import mark - - -@mark.django_db -class UnenrollmentEventBusTests(unittest.TestCase): - """ - Tests for unenrollment events that interact with the event bus. - """ - @override_settings(ENABLE_SEND_ENROLLMENT_EVENTS_OVER_BUS=False) - @mock.patch('common.djangoapps.student.handlers.get_producer', autospec=True) - def test_event_disabled(self, mock_producer): - """ - Test to verify that we do not push `CERTIFICATE_CREATED` events to the event bus if the - `SEND_CERTIFICATE_CREATED_SIGNAL` setting is disabled. - """ - course_unenrollment_receiver(None, None) - mock_producer.assert_not_called() - - @override_settings(FEATURES={'ENABLE_SEND_ENROLLMENT_EVENTS_OVER_BUS': True}) - @mock.patch('common.djangoapps.student.handlers.get_producer', autospec=True) - def test_event_enabled(self, mock_producer): - """ - Test to verify that we push `COURSE_UNENROLLMENT_COMPLETED` events to the event bus. - """ - user = UserFactory() - enrollment = CourseEnrollmentFactory(user=user) - - event_metadata = EventsMetadata( - event_type=COURSE_UNENROLLMENT_COMPLETED.event_type, - id=uuid4(), - minorversion=0, - source='openedx/lms/web', - sourcehost='lms.test', - time=datetime.now(timezone.utc) - ) - - event_kwargs = { - 'enrollment': enrollment, - 'metadata': event_metadata - } - course_unenrollment_receiver(None, COURSE_UNENROLLMENT_COMPLETED, **event_kwargs) - - # verify that the data sent to the event bus matches what we expect - print(mock_producer.return_value) - print(mock_producer.return_value.send.call_args) - data = mock_producer.return_value.send.call_args.kwargs - assert data['event_data']['enrollment'] == enrollment - assert data['topic'] == 'course-unenrollment-lifecycle' - assert data['event_key_field'] == 'enrollment.course.course_key' diff --git a/lms/djangoapps/certificates/tests/test_signals.py b/lms/djangoapps/certificates/tests/test_signals.py index 4eb8cf27f48e..6f4c888c6832 100644 --- a/lms/djangoapps/certificates/tests/test_signals.py +++ b/lms/djangoapps/certificates/tests/test_signals.py @@ -21,10 +21,6 @@ CertificateGenerationConfiguration, GeneratedCertificate ) -from lms.djangoapps.certificates.signals import ( - listen_for_certificate_created_event, - listen_for_certificate_revoked_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 @@ -443,109 +439,3 @@ def test_verified_to_audit(self): ) as mock_allowlist_task: self.verified_enrollment.change_mode('audit') mock_allowlist_task.assert_not_called() - - -class CertificateEventBusTests(ModuleStoreTestCase): - """ - Tests for Certificate events that interact with the event bus. - """ - def setUp(self): - super().setUp() - self.user = UserFactory.create() - self.name = f'{self.user.first_name} {self.user.last_name}' - self.course = CourseFactory.create(self_paced=True) - self.enrollment = CourseEnrollmentFactory( - user=self.user, - course_id=self.course.id, - is_active=True, - mode='verified', - ) - - def _create_event_data(self, event_type, certificate_status): - """ - Utility function to create test data for unit tests. - """ - expected_course_data = CourseData(course_key=self.course.id) - expected_user_data = UserData( - pii=UserPersonalData( - username=self.user.username, - email=self.user.email, - name=self.name, - ), - id=self.user.id, - is_active=self.user.is_active - ) - expected_certificate_data = CertificateData( - user=expected_user_data, - course=expected_course_data, - mode='verified', - grade='', - current_status=certificate_status, - download_url='', - name='', - ) - expected_event_metadata = EventsMetadata( - event_type=event_type.event_type, - id=uuid4(), - minorversion=0, - source='openedx/lms/web', - sourcehost='lms.test', - time=datetime.now(timezone.utc) - ) - - return { - 'certificate': expected_certificate_data, - 'metadata': expected_event_metadata, - } - - @override_settings(SEND_CERTIFICATE_CREATED_SIGNAL=False) - @mock.patch('lms.djangoapps.certificates.signals.get_producer', autospec=True) - def test_certificate_created_event_disabled(self, mock_producer): - """ - Test to verify that we do not publish `CERTIFICATE_CREATED` events to the event bus if the - `SEND_CERTIFICATE_CREATED_SIGNAL` setting is disabled. - """ - listen_for_certificate_created_event(None, CERTIFICATE_CREATED) - mock_producer.assert_not_called() - - @override_settings(SEND_CERTIFICATE_REVOKED_SIGNAL=False) - @mock.patch('lms.djangoapps.certificates.signals.get_producer', autospec=True) - def test_certificate_revoked_event_disabled(self, mock_producer): - """ - Test to verify that we do not publish `CERTIFICATE_REVOKED` events to the event bus if the - `SEND_CERTIFICATE_REVOKED_SIGNAL` setting is disabled. - """ - listen_for_certificate_created_event(None, CERTIFICATE_REVOKED) - mock_producer.assert_not_called() - - @override_settings(SEND_CERTIFICATE_CREATED_SIGNAL=True) - @mock.patch('lms.djangoapps.certificates.signals.get_producer', autospec=True) - def test_certificate_created_event_enabled(self, mock_producer): - """ - Test to verify that we push `CERTIFICATE_CREATED` events to the event bus if the - `SEND_CERTIFICATE_CREATED_SIGNAL` setting is enabled. - """ - event_data = self._create_event_data(CERTIFICATE_CREATED, CertificateStatuses.downloadable) - listen_for_certificate_created_event(None, CERTIFICATE_CREATED, **event_data) - # verify that the data sent to the event bus matches what we expect - data = mock_producer.return_value.send.call_args.kwargs - assert data['signal'].event_type == CERTIFICATE_CREATED.event_type - assert data['event_data']['certificate'] == event_data['certificate'] - assert data['topic'] == 'learning-certificate-lifecycle' - assert data['event_key_field'] == 'certificate.course.course_key' - - @override_settings(SEND_CERTIFICATE_REVOKED_SIGNAL=True) - @mock.patch('lms.djangoapps.certificates.signals.get_producer', autospec=True) - def test_certificate_revoked_event_enabled(self, mock_producer): - """ - Test to verify that we push `CERTIFICATE_REVOKED` events to the event bus if the - `SEND_CERTIFICATE_REVOKED_SIGNAL` setting is enabled. - """ - event_data = self._create_event_data(CERTIFICATE_REVOKED, CertificateStatuses.notpassing) - listen_for_certificate_revoked_event(None, CERTIFICATE_REVOKED, **event_data) - # verify that the data sent to the event bus matches what we expect - data = mock_producer.return_value.send.call_args.kwargs - assert data['signal'].event_type == CERTIFICATE_REVOKED.event_type - assert data['event_data']['certificate'] == event_data['certificate'] - assert data['topic'] == 'learning-certificate-lifecycle' - assert data['event_key_field'] == 'certificate.course.course_key'