refactor: use only runtime utilities to filter by course namespace (glob or specific key)

Author: mariajgrimaldi

openedx_authz/api/roles.py

@@ -41,6 +41,7 @@
     "get_subject_role_assignments",
     "get_subject_role_assignments_for_role_in_scope",
     "get_subject_role_assignments_in_scope",
+    "get_all_role_assignments_per_scope_type",
     "unassign_role_from_subject_in_scope",
     "unassign_subject_from_all_roles",
 ]
@@ -527,3 +528,22 @@ def unassign_subject_from_all_roles(subject: SubjectData) -> bool:
     """
     enforcer = AuthzEnforcer.get_enforcer()
     return enforcer.remove_filtered_grouping_policy(GroupingPolicyIndex.SUBJECT.value, subject.namespaced_key)
+
+
+def get_all_role_assignments_per_scope_type(scope_type: type[ScopeData]) -> list[RoleAssignmentData]:
+    """Get all role assignments for a specific scope type.
+
+    Loads all grouping policies from the enforcer and filters in Python. Casbin policies
+    store full scope keys (e.g., 'course-v1^course-v1:Org+Course+Run'), so there is no
+    way to query by scope type directly so the filtering must happen here.
+
+    Args:
+        scope_type: A ScopeData subclass (not an instance) used to match by NAMESPACE.
+
+    Returns:
+        list[RoleAssignmentData]: All assignments whose scope matches the given scope type.
+    """
+    return [
+        role_assignment for role_assignment in get_role_assignments()
+        if role_assignment.scope.NAMESPACE == scope_type.NAMESPACE
+    ]

openedx_authz/engine/utils.py

@@ -10,11 +10,11 @@
 from casbin import Enforcer
 
 from openedx_authz.api.data import CourseOverviewData, OrgCourseOverviewGlobData
+from openedx_authz.api.roles import get_all_role_assignments_per_scope_type
 from openedx_authz.api.users import (
     assign_role_to_user_in_scope,
     batch_assign_role_to_users_in_scope,
     batch_unassign_role_from_users,
-    get_user_role_assignments,
 )
 from openedx_authz.constants.roles import (
     LEGACY_COURSE_ROLE_EQUIVALENCES,
@@ -203,6 +203,8 @@ def migrate_legacy_course_roles_to_authz(course_access_role_model, course_id_lis
         raise ValueError(
             "At least one of course_id_list or org_id must be provided to limit the scope of the migration."
         )
+
+    # TODO: not sure if we should keep the startswith here
     course_access_role_filter = {}
 
     if org_id:
@@ -286,6 +288,12 @@ def migrate_authz_to_legacy_course_roles(
     This is essentially the reverse of migrate_legacy_course_roles_to_authz and is intended
     for rollback purposes in case of migration issues.
 
+    To build each CourseAccessRole entry, the function needs:
+    - A user: resolved from role assignments in scopes linked to courses.
+    - A scope: either a CourseOverviewData (course-level) or OrgCourseOverviewGlobData (org-level glob),
+      filtered by course_id or org_id if provided.
+    - A role: a role external key that maps to a legacy role in COURSE_ROLE_EQUIVALENCES.
+
     param course_access_role_model: It should be the CourseAccessRole model. This is passed in because the function
     is intended to run within a Django migration context, where direct model imports can cause issues.
     param user_subject_model: It should be the UserSubject model. This is passed in because the function
@@ -300,70 +308,77 @@ def migrate_authz_to_legacy_course_roles(
             "At least one of course_id_list or org_id must be provided to limit the scope of the rollback migration."
         )
 
-    # 1. Get all users with course-related permissions in the new model by filtering
-    # UserSubjects that are linked to CourseScopes with a valid course overview.
-    course_subject_filter = {
-        "casbin_rules__scope__coursescope__course_overview__isnull": False,
-    }
+    # CourseOverviewData and OrgCourseOverviewGlobData share the same namespace,
+    # so filtering by CourseOverviewData captures both course-level and org-level glob assignments.
+    role_assignments = get_all_role_assignments_per_scope_type(scope_type=CourseOverviewData)
 
+    # Two cases here:
+    # 1. If org_id is provided, we filter by org_id which will include both org-level glob scopes and course-level scopes linked to that org
+    # 2. If only course_id_list is provided, we filter by course_id which will include only course-level scopes linked to those course_ids since
+    # org-level glob scopes don't have course_id in their scope object
     if org_id:
-        course_subject_filter["casbin_rules__scope__coursescope__course_overview__org"] = org_id
+        role_assignments = [
+            role_assignment
+            for role_assignment in role_assignments
+            if role_assignment.scope.org == org_id
+        ]
 
     if course_id_list and not org_id:
-        # Only filter by course_id if org_id is not provided,
-        # otherwise we will filter by org_id which is more efficient
-        course_subject_filter["casbin_rules__scope__coursescope__course_overview__id__in"] = course_id_list
-
-    course_subjects = user_subject_model.objects.filter(**course_subject_filter).select_related("user").distinct()
+        role_assignments = [
+            role_assignment
+            for role_assignment in role_assignments
+            if isinstance(role_assignment.scope, CourseOverviewData)
+            and role_assignment.scope.external_key in course_id_list
+        ]
 
     roles_with_errors = []
     roles_with_no_errors = []
     unassignments = defaultdict(list)
 
-    for course_subject in course_subjects:
-        user = course_subject.user
-        user_external_key = user.username
+    for role_assignment in role_assignments:
+
+        # Per valid role assignment, create corresponding CourseAccessRole entry
+        try:
+            user_external_key = role_assignment.subject.external_key
+            role_external_key = role_assignment.roles[0].external_key
+            scope_external_key = role_assignment.scope.external_key
+
+            course_access_role_kwargs = {
+                "user": user_subject_model.objects.get(user__username=user_external_key).user,
+                "role": COURSE_ROLE_EQUIVALENCES[role_external_key],
+            }
+
+            # Here we prioritize course_id over org for scope since course-level scope is more specific
+            # and also both are not needed to create a valid CourseAccessRole entry
+            if isinstance(role_assignment.scope, CourseOverviewData):
+                course_access_role_kwargs["course_id"] = scope_external_key
+            elif isinstance(role_assignment.scope, OrgCourseOverviewGlobData):
+                course_access_role_kwargs["org"] = role_assignment.scope.org
+            else:
+                logger.error(
+                    f"Unexpected scope type: {type(role_assignment.scope)} for RoleAssignment with scope: {scope_external_key}"
+                )
+                roles_with_errors.append(role_assignment)
+                continue
+
+            course_access_role_model.objects.create(**course_access_role_kwargs)
+            roles_with_no_errors.append(role_assignment)
 
-        # 2. Get all role assignments for the user
-        role_assignments = get_user_role_assignments(user_external_key=user_external_key)
+            logger.info(
+                f"Successfully rolled back RoleAssignment for User: {user_external_key} "
+                f"in Role: {role_external_key} and Scope: {scope_external_key} "
+                f"to legacy CourseAccessRole entry."
+            )
 
-        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
+            if delete_after_migration:
+                unassignments[(role_external_key, scope_external_key)].append(user_external_key)
 
-            scope = assignment.scope.external_key
-
-            course_overview = assignment.scope.get_object()
-
-            for role in assignment.roles:
-                legacy_role = COURSE_ROLE_EQUIVALENCES.get(role.external_key)
-                if legacy_role is None:
-                    logger.error(f"Unknown role: {role} for User: {user_external_key}")
-                    roles_with_errors.append((user_external_key, role.external_key, scope))
-                    continue
-
-                try:
-                    # Create legacy CourseAccessRole entry
-                    course_access_role_model.objects.get_or_create(
-                        user=user,
-                        org=course_overview.org,
-                        course_id=scope,
-                        role=legacy_role,
-                    )
-                    roles_with_no_errors.append((user_external_key, role.external_key, scope))
-                except Exception as e:  # pylint: disable=broad-exception-caught
-                    logger.error(
-                        f"Error creating CourseAccessRole for User: "
-                        f"{user_external_key}, Role: {legacy_role}, Course: {scope}: {e}"
-                    )
-                    roles_with_errors.append((user_external_key, role.external_key, scope))
-                    continue
-
-                # If we successfully created the legacy role, we can add this role assignment
-                # to the unassignment list if delete_after_migration is True
-                if delete_after_migration:
-                    unassignments[(role.external_key, scope)].append(user_external_key)
+        except Exception as e:
+            logger.error(
+                f"Error rolling back RoleAssignment for User: {role_assignment.subject.external_key} "
+                f"in Role: {role_assignment.roles[0].external_key} and Scope: {role_assignment.scope.external_key}: {e}"
+            )
+            roles_with_errors.append(role_assignment)
 
     # Once the loop is done, we can log summary of unassignments
     # and perform batch unassignment if delete_after_migration is True