fixup! feat: add course authoring migration and rollback scripts

Author: dwong2708

openedx_authz/admin.py

@@ -5,6 +5,8 @@
 from django.contrib import admin
 
 from openedx_authz.models import ExtendedCasbinRule
+from openedx_authz.models.scopes import CourseScope
+from openedx_authz.models.subjects import UserSubject
 
 
 class CasbinRuleForm(forms.ModelForm):
@@ -48,3 +50,18 @@ class CasbinRuleAdmin(admin.ModelAdmin):
     # TODO: In a future, possibly we should only show an inline for the rules that
     # have an extended rule, and show the subject and scope information in detail.
     inlines = [ExtendedCasbinRuleInline]
+
+
+@admin.register(ExtendedCasbinRule)
+class ExtendedCasbinRuleAdmin(admin.ModelAdmin):
+    pass
+
+
+@admin.register(UserSubject)
+class UserSubjectAdmin(admin.ModelAdmin):
+    pass
+
+
+@admin.register(CourseScope)
+class CourseScopeAdmin(admin.ModelAdmin):
+    list_display = ("id", "course_overview")

openedx_authz/engine/utils.py

@@ -8,6 +8,7 @@
 
 from casbin import Enforcer
 
+from openedx_authz.api.data import CourseOverviewData
 from openedx_authz.api.users import (
     assign_role_to_user_in_scope,
     batch_assign_role_to_users_in_scope,
@@ -193,7 +194,9 @@ def migrate_legacy_course_roles_to_authz(CourseAccessRole, delete_after_migratio
     param CourseAccessRole: The CourseAccessRole model to use.
     """
 
-    legacy_permissions = CourseAccessRole.objects.select_related("user").all()
+    legacy_permissions = (
+        CourseAccessRole.objects.filter(course_id__startswith="course-v1:").select_related("user").all()
+    )
 
     # List to keep track of any permissions that could not be migrated
     permissions_with_errors = []
@@ -224,11 +227,20 @@ def migrate_legacy_course_roles_to_authz(CourseAccessRole, delete_after_migratio
             f"to Role: {role.external_key} in Scope: {permission.course_id}"
         )
 
-        assign_role_to_user_in_scope(
+        is_user_added = assign_role_to_user_in_scope(
             user_external_key=permission.user.username,
             role_external_key=role.external_key,
             scope_external_key=str(permission.course_id),
         )
+
+        if not is_user_added:
+            logger.error(
+                f"Failed to migrate permission for User: {permission.user.username} "
+                f"to Role: {role.external_key} in Scope: {permission.course_id}"
+            )
+            permissions_with_errors.append(permission)
+            continue
+
         permissions_with_no_errors.append(permission)
 
     if delete_after_migration:
@@ -264,6 +276,10 @@ def migrate_authz_to_legacy_course_roles(CourseAccessRole, UserSubject, delete_a
         role_assignments = get_user_role_assignments(user_external_key=user_external_key)
 
         for assignment in role_assignments:
+            if not isinstance(assignment.scope, CourseOverviewData):
+                logger.error(f"Skipping role assignment for User: {user_external_key} due to missing course scope.")
+                continue
+
             scope = assignment.scope.external_key
 
             course_overview = assignment.scope.get_object()

openedx_authz/management/commands/authz_migrate_course_authoring.py

@@ -34,31 +34,26 @@ def handle(self, *args, **options):
         self.stdout.write(self.style.WARNING("Starting legacy → Authz migration..."))
 
         try:
+            if delete_after_migration:
+                confirm = input(
+                    "Are you sure you want to delete successfully migrated legacy roles? Type 'yes' to continue: "
+                )
+
+                if confirm != "yes":
+                    self.stdout.write(self.style.WARNING("Deletion aborted."))
+                    return
             with transaction.atomic():
                 errors = migrate_legacy_course_roles_to_authz(
                     CourseAccessRole=CourseAccessRole,
-                    delete_after_migration=False,  # control deletion here instead
+                    delete_after_migration=delete_after_migration,
                 )
 
                 if errors:
                     self.stdout.write(self.style.ERROR(f"Migration completed with {len(errors)} errors."))
                 else:
                     self.stdout.write(self.style.SUCCESS("Migration completed successfully with no errors."))
 
-                # Handle deletion separately for safety
                 if delete_after_migration:
-                    confirm = input(
-                        "Are you sure you want to delete successfully migrated legacy roles? Type 'yes' to continue: "
-                    )
-
-                    if confirm != "yes":
-                        self.stdout.write(self.style.WARNING("Deletion aborted."))
-                        return
-
-                    migrated_ids = [p.id for p in CourseAccessRole.objects.all() if p not in errors]
-
-                    CourseAccessRole.objects.filter(id__in=migrated_ids).delete()
-
                     self.stdout.write(self.style.SUCCESS("Legacy roles deleted successfully."))
 
         except Exception as exc:

openedx_authz/management/commands/authz_rollback_course_authoring.py

@@ -36,36 +36,28 @@ def handle(self, *args, **options):
         self.stdout.write(self.style.WARNING("Starting Authz → Legacy rollback migration..."))
 
         try:
+            if delete_after_migration:
+                confirm = input(
+                    "Are you sure you want to remove the new Authz role "
+                    "assignments after rollback? Type 'yes' to continue: "
+                )
+
+                if confirm != "yes":
+                    self.stdout.write(self.style.WARNING("Deletion aborted."))
+                    return
             with transaction.atomic():
                 errors = migrate_authz_to_legacy_course_roles(
                     CourseAccessRole=CourseAccessRole,
                     UserSubject=UserSubject,
-                    delete_after_migration=False,  # control deletion here
+                    delete_after_migration=delete_after_migration,  # control deletion here
                 )
 
                 if errors:
                     self.stdout.write(self.style.ERROR(f"Rollback completed with {len(errors)} errors."))
                 else:
                     self.stdout.write(self.style.SUCCESS("Rollback completed successfully with no errors."))
 
-                # Handle deletion separately for safety
                 if delete_after_migration:
-                    confirm = input(
-                        "Are you sure you want to remove the new Authz role "
-                        "assignments after rollback? Type 'yes' to continue: "
-                    )
-
-                    if confirm != "yes":
-                        self.stdout.write(self.style.WARNING("Deletion aborted."))
-                        return
-
-                    # Re-run with deletion enabled
-                    migrate_authz_to_legacy_course_roles(
-                        CourseAccessRole=CourseAccessRole,
-                        UserSubject=UserSubject,
-                        delete_after_migration=True,
-                    )
-
                     self.stdout.write(self.style.SUCCESS("Authz role assignments removed successfully."))
 
         except Exception as exc:

openedx_authz/models/core.py

@@ -114,6 +114,9 @@ class Scope(BaseRegistryModel):
     class Meta:
         abstract = False
 
+    def __str__(self):
+        return f"Scope (namespace={self.NAMESPACE})"
+
 
 class Subject(BaseRegistryModel):
     """Model representing a subject in the authorization system.
@@ -132,6 +135,9 @@ class Subject(BaseRegistryModel):
     class Meta:
         abstract = False
 
+    def __str__(self):
+        return f"Subject (namespace={self.NAMESPACE})"
+
 
 class ExtendedCasbinRule(models.Model):
     """Extended model for Casbin rules to store additional metadata.
@@ -182,6 +188,9 @@ class Meta:
         verbose_name = "Extended Casbin Rule"
         verbose_name_plural = "Extended Casbin Rules"
 
+    def __str__(self):
+        return f"ExtendedCasbinRule for {self.casbin_rule}"
+
     @classmethod
     def create_based_on_policy(
         cls,

openedx_authz/tests/test_migrations.py

@@ -721,7 +721,7 @@ def test_authz_migrate_course_authoring_command(self, mock_migrate):
         mock_migrate.assert_called_once()
         args, kwargs = mock_migrate.call_args
 
-        self.assertEqual(kwargs["delete_after_migration"], False)
+        self.assertEqual(kwargs["delete_after_migration"], True)
 
     @patch("openedx_authz.management.commands.authz_rollback_course_authoring.CourseAccessRole", CourseAccessRole)
     @patch("openedx_authz.management.commands.authz_rollback_course_authoring.migrate_authz_to_legacy_course_roles")
@@ -747,12 +747,8 @@ def test_authz_rollback_course_authoring_command(self, mock_rollback):
         with patch("builtins.input", return_value="yes"):
             call_command("authz_rollback_course_authoring", "--delete")
 
-        # First call (normal)
-        # Second call (with deletion=True)
-        self.assertEqual(mock_rollback.call_count, 2)
+        self.assertEqual(mock_rollback.call_count, 1)
 
-        first_call_kwargs = mock_rollback.call_args_list[0][1]
-        second_call_kwargs = mock_rollback.call_args_list[1][1]
+        call_kwargs = mock_rollback.call_args_list[0][1]
 
-        self.assertEqual(first_call_kwargs["delete_after_migration"], False)
-        self.assertEqual(second_call_kwargs["delete_after_migration"], True)
+        self.assertEqual(call_kwargs["delete_after_migration"], True)