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

Merge pull request #28640 from eduNEXT/MJG/2nd_batch_openedx_events

Browse Source 
[BD-32] feat: add 2nd batch of Open edX Events
This commit is contained in:
Felipe Montoya
2021-09-22 14:12:06 -05:00
committed by GitHub
parent 0504d97ea2 1c447e6176
commit dbb4658819
8 changed files with 691 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files

52
common/djangoapps/student/models.py
View File

@@ -61,7 +61,11 @@ from openedx_events.learning.data import (
UserData,
UserPersonalData,
)
from openedx_events.learning.signals import COURSE_ENROLLMENT_CREATED
from openedx_events.learning.signals import (
COURSE_ENROLLMENT_CHANGED,
COURSE_ENROLLMENT_CREATED,
COURSE_UNENROLLMENT_COMPLETED,
)
import openedx.core.djangoapps.django_comment_common.comment_client as cc
from common.djangoapps.course_modes.models import CourseMode, get_cosmetic_verified_display_price
from common.djangoapps.student.emails import send_proctoring_requirements_email
@@ -1417,6 +1421,16 @@ class CourseEnrollment(models.Model):
self.mode = mode
mode_changed = True
try:
course_data = CourseData(
course_key=self.course_id,
display_name=self.course.display_name,
)
except CourseOverview.DoesNotExist:
course_data = CourseData(
course_key=self.course_id,
)
if activation_changed or mode_changed:
self.save()
self._update_enrollment_in_request_cache(
@@ -1425,6 +1439,24 @@ class CourseEnrollment(models.Model):
CourseEnrollmentState(self.mode, self.is_active),
)
COURSE_ENROLLMENT_CHANGED.send_event(
enrollment=CourseEnrollmentData(
user=UserData(
pii=UserPersonalData(
username=self.user.username,
email=self.user.email,
name=self.user.profile.name,
),
id=self.user.id,
is_active=self.user.is_active,
),
course=course_data,
mode=self.mode,
is_active=self.is_active,
creation_date=self.created,
)
)
if activation_changed:
if self.is_active:
self.emit_event(EVENT_NAME_ENROLLMENT_ACTIVATED, enterprise_uuid=enterprise_uuid)
@@ -1433,6 +1465,24 @@ class CourseEnrollment(models.Model):
self.emit_event(EVENT_NAME_ENROLLMENT_DEACTIVATED)
self.send_signal(EnrollStatusChange.unenroll)
COURSE_UNENROLLMENT_COMPLETED.send_event(
enrollment=CourseEnrollmentData(
user=UserData(
pii=UserPersonalData(
username=self.user.username,
email=self.user.email,
name=self.user.profile.name,
),
id=self.user.id,
is_active=self.user.is_active,
),
course=course_data,
mode=self.mode,
is_active=self.is_active,
creation_date=self.created,
)
)
if mode_changed:
# If mode changed to one that requires proctoring, send proctoring requirements email
if should_send_proctoring_requirements_email(self.user.username, self.course_id):

100
common/djangoapps/student/tests/test_events.py
View File

@@ -20,7 +20,11 @@ from openedx_events.learning.data import (
UserData,
UserPersonalData,
)
from openedx_events.learning.signals import COURSE_ENROLLMENT_CREATED
from openedx_events.learning.signals import (
COURSE_ENROLLMENT_CHANGED,
COURSE_ENROLLMENT_CREATED,
COURSE_UNENROLLMENT_COMPLETED,
)
from openedx_events.tests.utils import OpenEdxEventsTestMixin
from openedx.core.djangolib.testing.utils import skip_unless_lms
@@ -203,9 +207,15 @@ class EnrollmentEventsTest(SharedModuleStoreTestCase, OpenEdxEventsTestMixin):
the exact Data Attributes as the event definition stated:
- COURSE_ENROLLMENT_CREATED: sent after the user's enrollment.
- COURSE_ENROLLMENT_CHANGED: sent after the enrollment update.
- COURSE_UNENROLLMENT_COMPLETED: sent after the user's unenrollment.
"""
ENABLED_OPENEDX_EVENTS = ["org.openedx.learning.course.enrollment.created.v1"]
ENABLED_OPENEDX_EVENTS = [
"org.openedx.learning.course.enrollment.created.v1",
"org.openedx.learning.course.enrollment.changed.v1",
"org.openedx.learning.course.unenrollment.completed.v1",
]
@classmethod
def setUpClass(cls):
@@ -276,3 +286,89 @@ class EnrollmentEventsTest(SharedModuleStoreTestCase, OpenEdxEventsTestMixin):
},
event_receiver.call_args.kwargs
)
def test_enrollment_changed_event_emitted(self):
"""
Test whether the student enrollment changed event is sent after the enrollment
update process ends.
Expected result:
- COURSE_ENROLLMENT_CHANGED is sent and received by the mocked receiver.
- The arguments that the receiver gets are the arguments sent by the event
except the metadata generated on the fly.
"""
enrollment = CourseEnrollment.enroll(self.user, self.course.id)
event_receiver = mock.Mock(side_effect=self._event_receiver_side_effect)
COURSE_ENROLLMENT_CHANGED.connect(event_receiver)
enrollment.update_enrollment(mode="verified")
self.assertTrue(self.receiver_called)
self.assertDictContainsSubset(
{
"signal": COURSE_ENROLLMENT_CHANGED,
"sender": None,
"enrollment": CourseEnrollmentData(
user=UserData(
pii=UserPersonalData(
username=self.user.username,
email=self.user.email,
name=self.user.profile.name,
),
id=self.user.id,
is_active=self.user.is_active,
),
course=CourseData(
course_key=self.course.id,
display_name=self.course.display_name,
),
mode=enrollment.mode,
is_active=enrollment.is_active,
creation_date=enrollment.created,
),
},
event_receiver.call_args.kwargs
)
def test_unenrollment_completed_event_emitted(self):
"""
Test whether the student un-enrollment completed event is sent after the
user's unenrollment process.
Expected result:
- COURSE_UNENROLLMENT_COMPLETED is sent and received by the mocked receiver.
- The arguments that the receiver gets are the arguments sent by the event
except the metadata generated on the fly.
"""
enrollment = CourseEnrollment.enroll(self.user, self.course.id)
event_receiver = mock.Mock(side_effect=self._event_receiver_side_effect)
COURSE_UNENROLLMENT_COMPLETED.connect(event_receiver)
CourseEnrollment.unenroll(self.user, self.course.id)
self.assertTrue(self.receiver_called)
self.assertDictContainsSubset(
{
"signal": COURSE_UNENROLLMENT_COMPLETED,
"sender": None,
"enrollment": CourseEnrollmentData(
user=UserData(
pii=UserPersonalData(
username=self.user.username,
email=self.user.email,
name=self.user.profile.name,
),
id=self.user.id,
is_active=self.user.is_active,
),
course=CourseData(
course_key=self.course.id,
display_name=self.course.display_name,
),
mode=enrollment.mode,
is_active=False,
creation_date=enrollment.created,
),
},
event_receiver.call_args.kwargs
)

70
lms/djangoapps/certificates/models.py
View File

@@ -35,6 +35,9 @@ from lms.djangoapps.instructor_task.models import InstructorTask
from openedx.core.djangoapps.signals.signals import COURSE_CERT_AWARDED, COURSE_CERT_CHANGED, COURSE_CERT_REVOKED
from openedx.core.djangoapps.xmodule_django.models import NoneToEmptyManager
from openedx_events.learning.data import CourseData, UserData, UserPersonalData, CertificateData
from openedx_events.learning.signals import CERTIFICATE_CHANGED, CERTIFICATE_CREATED, CERTIFICATE_REVOKED
log = logging.getLogger(__name__)
User = get_user_model()
@@ -391,6 +394,28 @@ class GeneratedCertificate(models.Model):
status=self.status,
)
CERTIFICATE_REVOKED.send_event(
certificate=CertificateData(
user=UserData(
pii=UserPersonalData(
username=self.user.username,
email=self.user.email,
name=self.user.profile.name,
),
id=self.user.id,
is_active=self.user.is_active,
),
course=CourseData(
course_key=self.course_id,
),
mode=self.mode,
grade=self.grade,
current_status=self.status,
download_url=self.download_url,
name=self.name,
)
)
if previous_certificate_status == CertificateStatuses.downloadable:
# imported here to avoid a circular import issue
from lms.djangoapps.certificates.utils import emit_certificate_event
@@ -446,6 +471,29 @@ class GeneratedCertificate(models.Model):
mode=self.mode,
status=self.status,
)
CERTIFICATE_CHANGED.send_event(
certificate=CertificateData(
user=UserData(
pii=UserPersonalData(
username=self.user.username,
email=self.user.email,
name=self.user.profile.name,
),
id=self.user.id,
is_active=self.user.is_active,
),
course=CourseData(
course_key=self.course_id,
),
mode=self.mode,
grade=self.grade,
current_status=self.status,
download_url=self.download_url,
name=self.name,
)
)
if CertificateStatuses.is_passing_status(self.status):
COURSE_CERT_AWARDED.send_robust(
sender=self.__class__,
@@ -455,6 +503,28 @@ class GeneratedCertificate(models.Model):
status=self.status,
)
CERTIFICATE_CREATED.send_event(
certificate=CertificateData(
user=UserData(
pii=UserPersonalData(
username=self.user.username,
email=self.user.email,
name=self.user.profile.name,
),
id=self.user.id,
is_active=self.user.is_active,
),
course=CourseData(
course_key=self.course_id,
),
mode=self.mode,
grade=self.grade,
current_status=self.status,
download_url=self.download_url,
name=self.name,
)
)
class CertificateGenerationHistory(TimeStampedModel):
"""

227
lms/djangoapps/certificates/tests/test_events.py Normal file
View File

@@ -0,0 +1,227 @@
"""
Test classes for the events sent in the certification process.
Classes:
CertificateEventTest: Test event sent after creating, changing or deleting
certificates.
"""
from unittest.mock import Mock
from openedx_events.learning.data import CertificateData, CourseData, UserData, UserPersonalData
from openedx_events.learning.signals import CERTIFICATE_CHANGED, CERTIFICATE_CREATED, CERTIFICATE_REVOKED
from openedx_events.tests.utils import OpenEdxEventsTestMixin
from common.djangoapps.student.tests.factories import UserFactory
from lms.djangoapps.certificates.tests.factories import GeneratedCertificateFactory
from lms.djangoapps.certificates.models import GeneratedCertificate, CertificateStatuses
from openedx.core.djangoapps.content.course_overviews.tests.factories import CourseOverviewFactory
from openedx.core.djangolib.testing.utils import skip_unless_lms
from xmodule.modulestore.tests.django_utils import SharedModuleStoreTestCase
@skip_unless_lms
class CertificateEventTest(SharedModuleStoreTestCase, OpenEdxEventsTestMixin):
"""
Tests for the Open edX Events associated with the student's certification
process.
This class guarantees that the following events are sent during the user's
certification process, with the exact Data Attributes as the event definition stated:
- CERTIFICATE_CREATED: after the user's certificate generation has been
completed.
- CERTIFICATE_CHANGED: after the certificate update has been completed.
- CERTIFICATE_REVOKED: after the certificate revocation has been completed.
"""
ENABLED_OPENEDX_EVENTS = [
"org.openedx.learning.certificate.created.v1",
"org.openedx.learning.certificate.changed.v1",
"org.openedx.learning.certificate.revoked.v1",
]
@classmethod
def setUpClass(cls):
"""
Set up class method for the Test class.
This method starts manually events isolation. Explanation here:
openedx/core/djangoapps/user_authn/views/tests/test_events.py#L44
"""
super().setUpClass()
cls.start_events_isolation()
def setUp(self): # pylint: disable=arguments-differ
super().setUp()
self.course = CourseOverviewFactory()
self.user = UserFactory.create(
username="somestudent",
first_name="Student",
last_name="Person",
email="robot@robot.org",
is_active=True
)
self.receiver_called = False
def _event_receiver_side_effect(self, **kwargs): # pylint: disable=unused-argument
"""
Used show that the Open edX Event was called by the Django signal handler.
"""
self.receiver_called = True
def test_send_certificate_created_event(self):
"""
Test whether the certificate created event is sent at the end of the
certificate creation process.
Expected result:
- CERTIFICATE_CREATED is sent and received by the mocked receiver.
- The arguments that the receiver gets are the arguments sent by the event
except the metadata generated on the fly.
"""
event_receiver = Mock(side_effect=self._event_receiver_side_effect)
CERTIFICATE_CREATED.connect(event_receiver)
certificate = GeneratedCertificateFactory.create(
status=CertificateStatuses.downloadable,
user=self.user,
course_id=self.course.id,
mode=GeneratedCertificate.MODES.honor,
name="Certificate",
grade="100",
download_url="https://certificate.pdf"
)
self.assertTrue(self.receiver_called)
self.assertDictContainsSubset(
{
"signal": CERTIFICATE_CREATED,
"sender": None,
"certificate": CertificateData(
user=UserData(
pii=UserPersonalData(
username=certificate.user.username,
email=certificate.user.email,
name=certificate.user.profile.name,
),
id=certificate.user.id,
is_active=certificate.user.is_active,
),
course=CourseData(
course_key=certificate.course_id,
),
mode=certificate.mode,
grade=certificate.grade,
current_status=certificate.status,
download_url=certificate.download_url,
name=certificate.name,
),
},
event_receiver.call_args.kwargs
)
def test_send_certificate_changed_event(self):
"""
Test whether the certificate changed event is sent at the end of the
certificate update process.
Expected result:
- CERTIFICATE_CHANGED is sent and received by the mocked receiver.
- The arguments that the receiver gets are the arguments sent by the event
except the metadata generated on the fly.
"""
event_receiver = Mock(side_effect=self._event_receiver_side_effect)
CERTIFICATE_CHANGED.connect(event_receiver)
certificate = GeneratedCertificateFactory.create(
status=CertificateStatuses.downloadable,
user=self.user,
course_id=self.course.id,
mode=GeneratedCertificate.MODES.honor,
name="Certificate",
grade="100",
download_url="https://certificate.pdf"
)
certificate.grade = "50"
certificate.save()
self.assertTrue(self.receiver_called)
self.assertDictContainsSubset(
{
"signal": CERTIFICATE_CHANGED,
"sender": None,
"certificate": CertificateData(
user=UserData(
pii=UserPersonalData(
username=certificate.user.username,
email=certificate.user.email,
name=certificate.user.profile.name,
),
id=certificate.user.id,
is_active=certificate.user.is_active,
),
course=CourseData(
course_key=certificate.course_id,
),
mode=certificate.mode,
grade=certificate.grade,
current_status=certificate.status,
download_url=certificate.download_url,
name=certificate.name,
),
},
event_receiver.call_args.kwargs
)
def test_send_certificate_revoked_event(self):
"""
Test whether the certificate revoked event is sent at the end of the
user certificate's revoking process.
Expected result:
- CERTIFICATE_REVOKED is sent and received by the mocked receiver.
- The arguments that the receiver gets are the arguments sent by the event
except the metadata generated on the fly.
"""
event_receiver = Mock(side_effect=self._event_receiver_side_effect)
CERTIFICATE_REVOKED.connect(event_receiver)
certificate = GeneratedCertificateFactory.create(
status=CertificateStatuses.downloadable,
user=self.user,
course_id=self.course.id,
mode=GeneratedCertificate.MODES.honor,
name="Certificate",
grade="100",
download_url="https://certificate.pdf"
)
certificate.invalidate()
self.assertTrue(self.receiver_called)
self.assertDictContainsSubset(
{
"signal": CERTIFICATE_REVOKED,
"sender": None,
"certificate": CertificateData(
user=UserData(
pii=UserPersonalData(
username=certificate.user.username,
email=certificate.user.email,
name=certificate.user.profile.name,
),
id=certificate.user.id,
is_active=certificate.user.is_active,
),
course=CourseData(
course_key=certificate.course_id,
),
mode=certificate.mode,
grade=certificate.grade,
current_status=certificate.status,
download_url=certificate.download_url,
name=certificate.name,
),
},
event_receiver.call_args.kwargs
)

108
lms/djangoapps/certificates/tests/test_models.py
View File

@@ -17,6 +17,7 @@ from edx_name_affirmation.statuses import VerifiedNameStatus
from edx_name_affirmation.toggles import VERIFIED_NAME_FLAG
from edx_toggles.toggles.testutils import override_waffle_flag
from opaque_keys.edx.locator import CourseKey, CourseLocator
from openedx_events.tests.utils import OpenEdxEventsTestMixin
from path import Path as path
from common.djangoapps.course_modes.models import CourseMode
@@ -55,7 +56,7 @@ PLATFORM_ROOT = TEST_DIR.parent.parent.parent.parent
TEST_DATA_ROOT = PLATFORM_ROOT / TEST_DATA_DIR
class ExampleCertificateTest(TestCase):
class ExampleCertificateTest(TestCase, OpenEdxEventsTestMixin):
"""Tests for the ExampleCertificate model. """
COURSE_KEY = CourseLocator(org='test', course='test', run='test')
@@ -65,6 +66,19 @@ class ExampleCertificateTest(TestCase):
DOWNLOAD_URL = 'https://www.example.com'
ERROR_REASON = 'Kaboom!'
ENABLED_OPENEDX_EVENTS = []
@classmethod
def setUpClass(cls):
"""
Set up class method for the Test class.
This method starts manually events isolation. Explanation here:
openedx/core/djangoapps/user_authn/views/tests/test_events.py#L44
"""
super().setUpClass()
cls.start_events_isolation()
def setUp(self):
super().setUp()
self.cert_set = ExampleCertificateSet.objects.create(course_key=self.COURSE_KEY)
@@ -112,10 +126,24 @@ class ExampleCertificateTest(TestCase):
assert result is None
class CertificateHtmlViewConfigurationTest(TestCase):
class CertificateHtmlViewConfigurationTest(TestCase, OpenEdxEventsTestMixin):
"""
Test the CertificateHtmlViewConfiguration model.
"""
ENABLED_OPENEDX_EVENTS = []
@classmethod
def setUpClass(cls):
"""
Set up class method for the Test class.
This method starts manually events isolation. Explanation here:
openedx/core/djangoapps/user_authn/views/tests/test_events.py#L44
"""
super().setUpClass()
cls.start_events_isolation()
def setUp(self):
super().setUp()
self.configuration_string = """{
@@ -205,12 +233,25 @@ class CertificateTemplateAssetTest(TestCase):
assert certificate_template_asset.asset == 'certificate_template_assets/1/picture2.jpg'
class EligibleCertificateManagerTest(SharedModuleStoreTestCase):
class EligibleCertificateManagerTest(SharedModuleStoreTestCase, OpenEdxEventsTestMixin):
"""
Test the GeneratedCertificate model's object manager for filtering
out ineligible certs.
"""
ENABLED_OPENEDX_EVENTS = []
@classmethod
def setUpClass(cls):
"""
Set up class method for the Test class.
This method starts manually events isolation. Explanation here:
openedx/core/djangoapps/user_authn/views/tests/test_events.py#L44
"""
super().setUpClass()
cls.start_events_isolation()
def setUp(self):
super().setUp()
self.user = UserFactory()
@@ -250,10 +291,24 @@ class EligibleCertificateManagerTest(SharedModuleStoreTestCase):
@ddt.ddt
class TestCertificateGenerationHistory(TestCase):
class TestCertificateGenerationHistory(TestCase, OpenEdxEventsTestMixin):
"""
Test the CertificateGenerationHistory model's methods
"""
ENABLED_OPENEDX_EVENTS = []
@classmethod
def setUpClass(cls):
"""
Set up class method for the Test class.
This method starts manually events isolation. Explanation here:
openedx/core/djangoapps/user_authn/views/tests/test_events.py#L44
"""
super().setUpClass()
cls.start_events_isolation()
@ddt.data(
({"student_set": "allowlisted_not_generated"}, "For exceptions", True),
({"student_set": "allowlisted_not_generated"}, "For exceptions", False),
@@ -308,11 +363,24 @@ class TestCertificateGenerationHistory(TestCase):
assert certificate_generation_history.get_task_name() == expected
class CertificateInvalidationTest(SharedModuleStoreTestCase):
class CertificateInvalidationTest(SharedModuleStoreTestCase, OpenEdxEventsTestMixin):
"""
Test for the Certificate Invalidation model.
"""
ENABLED_OPENEDX_EVENTS = []
@classmethod
def setUpClass(cls):
"""
Set up class method for the Test class.
This method starts manually events isolation. Explanation here:
openedx/core/djangoapps/user_authn/views/tests/test_events.py#L44
"""
super().setUpClass()
cls.start_events_isolation()
def setUp(self):
super().setUp()
self.course = CourseFactory()
@@ -365,11 +433,24 @@ class CertificateInvalidationTest(SharedModuleStoreTestCase):
@ddt.ddt
class GeneratedCertificateTest(SharedModuleStoreTestCase):
class GeneratedCertificateTest(SharedModuleStoreTestCase, OpenEdxEventsTestMixin):
"""
Test GeneratedCertificates
"""
ENABLED_OPENEDX_EVENTS = []
@classmethod
def setUpClass(cls):
"""
Set up class method for the Test class.
This method starts manually events isolation. Explanation here:
openedx/core/djangoapps/user_authn/views/tests/test_events.py#L44
"""
super().setUpClass()
cls.start_events_isolation()
def setUp(self):
super().setUp()
self.user = UserFactory()
@@ -606,11 +687,24 @@ class GeneratedCertificateTest(SharedModuleStoreTestCase):
self._assert_event_data(mock_emit_certificate_event, expected_event_data)
class CertificateAllowlistTest(SharedModuleStoreTestCase):
class CertificateAllowlistTest(SharedModuleStoreTestCase, OpenEdxEventsTestMixin):
"""
Tests for the CertificateAllowlist model.
"""
ENABLED_OPENEDX_EVENTS = []
@classmethod
def setUpClass(cls):
"""
Set up class method for the Test class.
This method starts manually events isolation. Explanation here:
openedx/core/djangoapps/user_authn/views/tests/test_events.py#L44
"""
super().setUpClass()
cls.start_events_isolation()
def setUp(self):
super().setUp()
self.username = 'fun_username'

21
openedx/core/djangoapps/course_groups/models.py
View File

@@ -16,6 +16,9 @@ from opaque_keys.edx.django.models import CourseKeyField
from openedx.core.djangolib.model_mixins import DeletableByUserValue
from openedx_events.learning.data import CohortData, CourseData, UserData, UserPersonalData
from openedx_events.learning.signals import COHORT_MEMBERSHIP_CHANGED
log = logging.getLogger(__name__)
@@ -129,6 +132,24 @@ class CohortMembership(models.Model):
def save(self, force_insert=False, force_update=False, using=None, update_fields=None):
self.full_clean(validate_unique=False)
COHORT_MEMBERSHIP_CHANGED.send_event(
cohort=CohortData(
user=UserData(
pii=UserPersonalData(
username=self.user.username,
email=self.user.email,
name=self.user.profile.name,
),
id=self.user.id,
is_active=self.user.is_active,
),
course=CourseData(
course_key=self.course_id,
),
name=self.course_user_group.name,
)
)
log.info("Saving CohortMembership for user '%s' in '%s'", self.user.id, self.course_id)
return super().save(
force_insert=force_insert,

16
openedx/core/djangoapps/course_groups/tests/test_cohorts.py
View File

@@ -12,6 +12,7 @@ from django.http import Http404
from django.test import TestCase
from opaque_keys.edx.keys import CourseKey
from opaque_keys.edx.locator import CourseLocator
from openedx_events.tests.utils import OpenEdxEventsTestMixin
from common.djangoapps.student.models import CourseEnrollment
from common.djangoapps.student.tests.factories import UserFactory
@@ -25,11 +26,24 @@ from ..tests.helpers import CohortFactory, CourseCohortFactory, config_course_co
@patch("openedx.core.djangoapps.course_groups.cohorts.tracker", autospec=True)
class TestCohortSignals(TestCase):
class TestCohortSignals(TestCase, OpenEdxEventsTestMixin):
"""
Test cases to validate event emissions for various cohort-related workflows
"""
ENABLED_OPENEDX_EVENTS = []
@classmethod
def setUpClass(cls):
"""
Set up class method for the Test class.
This method starts manually events isolation. Explanation here:
openedx/core/djangoapps/user_authn/views/tests/test_events.py#L44
"""
super().setUpClass()
cls.start_events_isolation()
def setUp(self):
super().setUp()
self.course_key = CourseLocator("dummy", "dummy", "dummy")

108
openedx/core/djangoapps/course_groups/tests/test_events.py Normal file
View File

@@ -0,0 +1,108 @@
"""
Test classes for the events sent in the cohort assignment process.
Classes:
CohortEventTest: Test event sent after cohort membership changes.
"""
from openedx.core.djangoapps.course_groups.models import CohortMembership
from unittest.mock import Mock
from openedx_events.learning.data import CohortData, CourseData, UserData, UserPersonalData
from openedx_events.learning.signals import COHORT_MEMBERSHIP_CHANGED
from openedx_events.tests.utils import OpenEdxEventsTestMixin
from common.djangoapps.student.tests.factories import UserFactory
from openedx.core.djangoapps.content.course_overviews.tests.factories import CourseOverviewFactory
from openedx.core.djangolib.testing.utils import skip_unless_lms
from openedx.core.djangoapps.course_groups.tests.helpers import CohortFactory
from xmodule.modulestore.tests.django_utils import SharedModuleStoreTestCase
@skip_unless_lms
class CohortEventTest(SharedModuleStoreTestCase, OpenEdxEventsTestMixin):
"""
Tests for the Open edX Events associated with the cohort update process.
This class guarantees that the following events are sent during the user's
certification process, with the exact Data Attributes as the event definition stated:
- COHORT_MEMBERSHIP_CHANGED: when a cohort membership update ends.
"""
ENABLED_OPENEDX_EVENTS = [
"org.openedx.learning.cohort_membership.changed.v1",
]
@classmethod
def setUpClass(cls):
"""
Set up class method for the Test class.
This method starts manually events isolation. Explanation here:
openedx/core/djangoapps/user_authn/views/tests/test_events.py#L44
"""
super().setUpClass()
cls.start_events_isolation()
def setUp(self): # pylint: disable=arguments-differ
super().setUp()
self.course = CourseOverviewFactory()
self.user = UserFactory.create(
username="somestudent",
first_name="Student",
last_name="Person",
email="robot@robot.org",
is_active=True
)
self.cohort = CohortFactory(course_id=self.course.id, name="FirstCohort")
self.receiver_called = False
def _event_receiver_side_effect(self, **kwargs): # pylint: disable=unused-argument
"""
Used show that the Open edX Event was called by the Django signal handler.
"""
self.receiver_called = True
def test_send_cohort_membership_changed_event(self):
"""
Test whether the COHORT_MEMBERSHIP_CHANGED event is sent when a cohort
membership update ends.
Expected result:
- COHORT_MEMBERSHIP_CHANGED is sent and received by the mocked receiver.
- The arguments that the receiver gets are the arguments sent by the event
except the metadata generated on the fly.
"""
event_receiver = Mock(side_effect=self._event_receiver_side_effect)
COHORT_MEMBERSHIP_CHANGED.connect(event_receiver)
cohort_membership, _ = CohortMembership.assign(
cohort=self.cohort,
user=self.user,
)
self.assertTrue(self.receiver_called)
self.assertDictContainsSubset(
{
"signal": COHORT_MEMBERSHIP_CHANGED,
"sender": None,
"cohort": CohortData(
user=UserData(
pii=UserPersonalData(
username=cohort_membership.user.username,
email=cohort_membership.user.email,
name=cohort_membership.user.profile.name,
),
id=cohort_membership.user.id,
is_active=cohort_membership.user.is_active,
),
course=CourseData(
course_key=cohort_membership.course_id,
),
name=cohort_membership.course_user_group.name,
),
},
event_receiver.call_args.kwargs
)