fixup! feat: add course authoring migration and rollback scripts

Author: dwong2708

openedx_authz/engine/utils.py

@@ -29,6 +29,15 @@
 GROUPING_POLICY_PTYPES = ["g", "g2", "g3", "g4", "g5", "g6"]
 
 
+# Map new roles back to legacy roles
+role_to_legacy_role = {
+    COURSE_ADMIN.external_key: "instructor",
+    COURSE_STAFF.external_key: "staff",
+    COURSE_LIMITED_STAFF.external_key: "limited_staff",
+    COURSE_DATA_RESEARCHER.external_key: "data_researcher",
+}
+
+
 def migrate_policy_between_enforcers(
     source_enforcer: Enforcer,
     target_enforcer: Enforcer,
@@ -268,14 +277,6 @@ def migrate_authz_to_legacy_course_roles(CourseAccessRole, UserSubject, delete_a
                 if not scope.startswith("course-v1:"):
                     continue
 
-                # Map new roles back to legacy roles
-                role_to_legacy_role = {
-                    COURSE_ADMIN.external_key: "instructor",
-                    COURSE_STAFF.external_key: "staff",
-                    COURSE_LIMITED_STAFF.external_key: "limited_staff",
-                    COURSE_DATA_RESEARCHER.external_key: "data_researcher",
-                }
-
                 legacy_role = role_to_legacy_role.get(role.external_key)
                 if legacy_role is None:
                     logger.error(f"Unknown role: {role} for User: {user_external_key}")

openedx_authz/migrations/0008_migrate_course_roles_to_authz.py

@@ -590,6 +590,99 @@ def test_migrate_legacy_course_roles_to_authz_and_rollback_with_deletion(self):
         # After rollback, we should have 0 UserSubjects related to the course permissions
         self.assertEqual(len(state_after_migration_user_subjects), 0)
 
+    @patch("openedx_authz.api.data.CourseOverview", CourseOverview)
+    def test_migrate_legacy_course_roles_to_authz_and_rollback_with_no_new_role_equivalent(self):
+        """Test the migration of legacy course roles to the new Casbin-based model
+        and the rollback when there is no equivalent new role.
+        """
+
+        # Migrate from legacy CourseAccessRole to new Casbin-based model
+        permissions_with_errors = migrate_legacy_course_roles_to_authz(CourseAccessRole, delete_after_migration=True)
+        AuthzEnforcer.get_enforcer().load_policy()
+
+        # Now let's rollback
+
+        # Capture the state of permissions before rollback to verify that rollback restores the original state
+        original_state_user_subjects = list(
+            UserSubject.objects.filter(casbin_rules__scope__coursescope__course_overview__isnull=False)
+            .distinct()
+            .order_by("id")
+            .values("id", "user_id")
+        )
+        original_state_access_roles = list(
+            CourseAccessRole.objects.all().order_by("id").values("id", "user_id", "org", "course_id", "role")
+        )
+
+        # Mock the role_to_legacy_role mapping to only include a mapping
+        # for COURSE_ADMIN to simulate the scenario where the staff, limited_staff
+        # and data_researcher roles do not have a legacy role equivalent and
+        # therefore cannot be migrated back to legacy roles during the rollback.
+        with patch(
+            "openedx_authz.engine.utils.role_to_legacy_role",
+            {COURSE_ADMIN.external_key: "instructor"},
+        ):
+            permissions_with_errors = migrate_authz_to_legacy_course_roles(
+                CourseAccessRole, UserSubject, delete_after_migration=True
+            )
+
+        # Check that each user has the expected legacy role after rollback
+        # and that errors are logged for any permissions that could not be rolled back
+        for user in self.admin_users:
+            assignments = get_user_role_assignments_in_scope(
+                user_external_key=user.username, scope_external_key=self.course_id
+            )
+            self.assertEqual(len(assignments), 0)
+        for user in self.staff_users:
+            assignments = get_user_role_assignments_in_scope(
+                user_external_key=user.username, scope_external_key=self.course_id
+            )
+            # Since we are mocking the role_to_legacy_role mapping to only include a mapping for COURSE_ADMIN,
+            # the staff role will not have a legacy role equivalent and therefore should not be migrated back
+            self.assertEqual(len(assignments), 1)
+        for user in self.limited_staff:
+            assignments = get_user_role_assignments_in_scope(
+                user_external_key=user.username, scope_external_key=self.course_id
+            )
+            # Since we are mocking the role_to_legacy_role mapping to only include a mapping for COURSE_ADMIN,
+            # the limited_staff role will not have a legacy role equivalent and therefore should not be migrated back
+            self.assertEqual(len(assignments), 1)
+        for user in self.data_researcher:
+            assignments = get_user_role_assignments_in_scope(
+                user_external_key=user.username, scope_external_key=self.course_id
+            )
+            # Since we are mocking the role_to_legacy_role mapping to only include a mapping for COURSE_ADMIN,
+            # the data_researcher role will not have a legacy role equivalent and therefore should not be migrated back
+            self.assertEqual(len(assignments), 1)
+
+        # 3 staff + 3 limited_staff + 3 data_researcher = 9 entries with no legacy role equivalent
+        self.assertEqual(len(permissions_with_errors), 9)
+
+        state_after_migration_user_subjects = list(
+            UserSubject.objects.filter(casbin_rules__scope__coursescope__course_overview__isnull=False)
+            .distinct()
+            .order_by("id")
+            .values("id", "user_id")
+        )
+        after_migrate_state_access_roles = list(
+            CourseAccessRole.objects.all().order_by("id").values("id", "user_id", "org", "course_id", "role")
+        )
+
+        # Before the rollback, we should only have the 1 invalid role entry
+        # since we set delete_after_migration to True in the migration.
+        self.assertEqual(len(original_state_access_roles), 1)
+
+        # All original entries (1) + 3 users * 1 roles = 4
+        self.assertEqual(len(after_migrate_state_access_roles), 1 + 3)
+
+        # Before the rollback, we should have the 12 UserSubjects related to the course permissions
+        # since we had 3 users with 4 roles each in the original state.
+        self.assertEqual(len(original_state_user_subjects), 12)
+
+        # After rollback, we should have 9 UserSubjects related to the course permissions
+        # since the users with staff, limited_staff and data_researcher roles will not be
+        # migrated back to legacy roles due to our mocked role_to_legacy_role mapping.
+        self.assertEqual(len(state_after_migration_user_subjects), 9)
+
     @patch("openedx_authz.management.commands.authz_migrate_course_authoring.CourseAccessRole", CourseAccessRole)
     @patch("openedx_authz.management.commands.authz_migrate_course_authoring.migrate_legacy_course_roles_to_authz")
     def test_authz_migrate_course_authoring_command(self, mock_migrate):