feat!: add mango (maple) support for course creation assignment receiver

openedx_demo_plugin/receivers.py

@@ -4,16 +4,21 @@
 For a detailed description on events receivers definitions please refer to the
 hooks official documentation.
 """
+import logging
+
 from django.conf import settings
 from django.contrib.auth import get_user_model
 from openedx_events.learning.data import UserData
 
 try:
-    from common.djangoapps.student.api import get_access_role_by_role_name
+    from cms.djangoapps.course_creators.models import CourseCreator
+    from organizations.api import get_organization_by_short_name
 except ImportError:
-    get_access_role_by_role_name = object
+    get_organization_by_short_name = object
+    CourseCreator = object
 
 User = get_user_model()
+log = logging.getLogger(__name__)
 
 
 def assign_org_course_access_to_user(user: UserData, **kwargs):
@@ -24,10 +29,31 @@ def assign_org_course_access_to_user(user: UserData, **kwargs):
     OPEN_EDX_VISITOR_ORG setting if exists. If doesn't exist, then acts
     like a noop.
     """
-    visitor_org = getattr(settings, "OPEN_EDX_VISITOR_ORG", None)
-    if not visitor_org:
+    visitor_org_short_name = getattr(settings, "OPEN_EDX_VISITOR_ORG", None)
+    if not visitor_org_short_name:
+        log.info("No OPEN_EDX_VISITOR_ORG provided, terminating course creation assignment.")
+        return
+
+    course_creator_admin_id = getattr(settings, "COURSE_CREATOR_ADMIN_ID", None)
+    if not course_creator_admin_id:
+        log.info("No COURSE_CREATOR_ADMIN_ID provided, terminating course creation assignment.")
         return
 
     registered_user = User.objects.get(username=user.pii.username)
-    org_content_creator_role = get_access_role_by_role_name("org_course_creator_group")
-    org_content_creator_role(org=visitor_org).add_users(registered_user)
+    course_creator = CourseCreator(
+        user=registered_user,
+        state=CourseCreator.GRANTED,
+        all_organizations=False,
+    )
+
+    # In order to add course creator permissions programmatically, we must attach
+    # to the user just registered. So the post_add signals receivers like
+    # `course_creator_organizations_changed_callback` can run checks over instance.admin.
+    visitor_org = get_organization_by_short_name(visitor_org_short_name)
+    try:
+        course_creator.admin = User.objects.get(username=course_creator_admin_id)
+    except User.DoestNotExist:
+        log.exception("User with username specified in COURSE_CREATOR_ADMIN_ID does not exist.")
+        return
+    course_creator.save()
+    course_creator.organizations.add(visitor_org.get("id"))

openedx_demo_plugin/settings/common.py

@@ -24,3 +24,5 @@ def plugin_settings(settings):
             ]
         }
     }
+    if "cms.djangoapps.course_creators" not in settings.INSTALLED_APPS:
+        settings.INSTALLED_APPS = settings.INSTALLED_APPS + ["cms.djangoapps.course_creators"]

openedx_demo_plugin/tests/test_receivers.py

@@ -5,6 +5,7 @@
 """
 from unittest.mock import patch
 
+from django.conf import settings
 from django.contrib.auth import get_user_model
 from django.test import TestCase, override_settings
 from openedx_events.data import EventsMetadata
@@ -16,6 +17,9 @@
 User = get_user_model()
 
 
+@override_settings(
+    OPEN_EDX_VISITOR_ORG="Public", COURSE_CREATOR_ADMIN_ID="dummy-staff-user",
+)
 class RegistrationCompletedReceiverTest(TestCase):
     """
     Tests the registration receiver assigns the correct permissions.
@@ -40,25 +44,77 @@ def setUp(self):
             minorversion=0,
         )
         self.registered_user = User.objects.create(username=self.user.pii.username)
+        self.staff = User.objects.create(username=settings.COURSE_CREATOR_ADMIN_ID, is_staff=True)
 
-    @patch("openedx_demo_plugin.receivers.get_access_role_by_role_name")
-    def test_receiver_called_after_event(self, get_access_role_by_role_name):
+    @patch("openedx_demo_plugin.receivers.CourseCreator")
+    @patch("openedx_demo_plugin.receivers.get_organization_by_short_name")
+    def test_receiver_called_after_event(self, get_organization_by_short_name, course_creator):
         """
         Test that assign_org_course_access_to_user is called the correct information after sending
         STUDENT_REGISTRATION_COMPLETED event.
         """
-        org_content_creator_role = get_access_role_by_role_name("org_course_creator_group")
+        organization_id = 1
+        get_organization_by_short_name.return_value = {
+            "id": organization_id,
+        }
         STUDENT_REGISTRATION_COMPLETED.connect(assign_org_course_access_to_user)
 
         STUDENT_REGISTRATION_COMPLETED.send_event(
             user=self.user,
         )
 
-        org_content_creator_role(org="Public").add_users.assert_called_with(self.registered_user)
+        get_organization_by_short_name.assert_called_with(settings.OPEN_EDX_VISITOR_ORG)
+        course_creator.assert_called_with(
+            user=self.registered_user,
+            state=course_creator.GRANTED,
+            all_organizations=False,
+        )
+        course_creator.return_value.organizations.add.assert_called_with(organization_id)
+
+    @override_settings(COURSE_CREATOR_ADMIN_ID="non-existent-user")
+    @patch("openedx_demo_plugin.receivers.CourseCreator")
+    @patch("openedx_demo_plugin.receivers.get_organization_by_short_name")
+    def test_unexistent_course_creator_staff(self, get_organization_by_short_name, course_creator):
+        """
+        Test that stops when the user associated with COURSE_CREATOR_ADMIN_ID does not exist
+        after sending STUDENT_REGISTRATION_COMPLETED event.
+        """
+        STUDENT_REGISTRATION_COMPLETED.connect(assign_org_course_access_to_user)
+
+        STUDENT_REGISTRATION_COMPLETED.send_event(
+            user=self.user,
+        )
+
+        get_organization_by_short_name.assert_called_with(settings.OPEN_EDX_VISITOR_ORG)
+        course_creator.assert_called_with(
+            user=self.registered_user,
+            state=course_creator.GRANTED,
+            all_organizations=False,
+        )
+        course_creator.return_value.organizations.add.assert_not_called()
+
+    @override_settings(COURSE_CREATOR_ADMIN_ID=None)
+    @patch("openedx_demo_plugin.receivers.CourseCreator")
+    @patch("openedx_demo_plugin.receivers.get_organization_by_short_name")
+    def test_not_specified_course_creator_id(self, get_organization_by_short_name, course_creator):
+        """
+        Test that stops when COURSE_CREATOR_ADMIN_ID is not specified before sending STUDENT_REGISTRATION_COMPLETED
+        event.
+        """
+        STUDENT_REGISTRATION_COMPLETED.connect(assign_org_course_access_to_user)
+
+        STUDENT_REGISTRATION_COMPLETED.send_event(
+            user=self.user,
+        )
+
+        get_organization_by_short_name.assert_not_called()
+        course_creator.assert_not_called()
+        course_creator.return_value.organizations.add.assert_not_called()
 
     @override_settings(OPEN_EDX_VISITOR_ORG=None)
-    @patch("openedx_demo_plugin.receivers.get_access_role_by_role_name")
-    def test_receiver_noop(self, get_access_role_by_role_name):
+    @patch("openedx_demo_plugin.receivers.CourseCreator")
+    @patch("openedx_demo_plugin.receivers.get_organization_by_short_name")
+    def test_receiver_noop(self, get_organization_by_short_name, course_creator):
         """
         Test that when OPEN_EDX_VISITOR_ORG is not defined then the receiver acts as
         a noop.
@@ -69,4 +125,6 @@ def test_receiver_noop(self, get_access_role_by_role_name):
             user=self.user,
         )
 
-        get_access_role_by_role_name.return_value.assert_not_called()
+        get_organization_by_short_name.return_value.assert_not_called()
+        course_creator.assert_not_called()
+        course_creator.return_value.organizations.add.assert_not_called()