refactor: add tests for getting assignments for role

mariajgrimaldi · mariajgrimaldi · commit f2c164fd7aaa · 2025-10-01T18:11:34.000+02:00
diff --git a/openedx_authz/api/data.py b/openedx_authz/api/data.py
@@ -32,10 +32,16 @@ class AuthZData:
     Attributes:
         NAMESPACE: The namespace prefix for the data type (e.g., 'user', 'role').
         SEPARATOR: The separator between the namespace and the identifier (e.g., ':', '@').
+
+    Subclasses are automatically registered by their NAMESPACE for factory pattern.
     """
 
     SEPARATOR: str = "@"
-    NAMESPACE: str = None  # To be defined in subclasses
+    NAMESPACE: str = None
+
+    # TODO: Implement factory method to return correct subclass based on NAMESPACE prefix.
+    # This would allow initializing with either subject or scope, etc. and returning the correct subclass.
+    # So we don't have to manage each subclass separately or hardcoded anywhere.
 
 
 @define
@@ -45,21 +51,25 @@ class ScopeData(AuthZData):
     Attributes:
         scope_id: The scope identifier (e.g., 'org@Demo').
 
-    This class assumes that the scope is already namespaced appropriately
-    before being passed in, as scopes can vary widely (e.g., courses, organizations).
+    Acts as a factory: automatically returns the correct subclass based on the scope_id prefix.
     """
 
-    NAMESPACE: str = "sc"  # Generic scope namespace, should be overridden by specific scope types
+    NAMESPACE: str = "sc"
     scope_id: str = ""
-    name: str = ""  # Optional human-readable name
+    name: str = ""
 
     def __attrs_post_init__(self):
         """Ensure scope ID has appropriate namespace prefix."""
         if not self.scope_id:
             self.scope_id = f"{self.NAMESPACE}{self.SEPARATOR}{self.name}".lower()
 
         # Allow reverse lookup of name from scope_id
-        if not self.name and self.scope_id and self.NAMESPACE and self.scope_id.startswith(f"{self.NAMESPACE}{self.SEPARATOR}"):
+        if (
+            not self.name
+            and self.scope_id
+            and self.NAMESPACE
+            and self.scope_id.startswith(f"{self.NAMESPACE}{self.SEPARATOR}")
+        ):
             self.name = self.scope_id.split(self.SEPARATOR, 1)[1].lower()
 
 
@@ -81,7 +91,9 @@ def __attrs_post_init__(self):
             self.scope_id = f"{self.NAMESPACE}{self.SEPARATOR}{self.library_id}".lower()
 
         # Allow reverse lookup of library_id from scope_id
-        if not self.library_id and self.scope_id.startswith(f"{self.NAMESPACE}{self.SEPARATOR}"):
+        if not self.library_id and self.scope_id.startswith(
+            f"{self.NAMESPACE}{self.SEPARATOR}"
+        ):
             self.library_id = self.scope_id.split(self.SEPARATOR, 1)[1].lower()
 
 
@@ -92,27 +104,22 @@ class SubjectData(AuthZData):
     Attributes:
         subject_id: The subject identifier namespaced (e.g., 'user@john_doe').
 
-    This class assumes that the subject was already namespaced by their own
-    type (e.g., 'user@', 'group@') before being passed in since subjects can be
-    users, groups, or other entities.
+    Acts as a factory: automatically returns the correct subclass based on the subject_id prefix.
     """
 
-    NAMESPACE: str = (
-        "sub"  # Generic subject namespace, should be overridden by specific subject types
-    )
+    NAMESPACE: str = "sub"
     subject_id: str = ""
-    name: str = ""  # Optional human-readable name
+    name: str = ""
 
     def __attrs_post_init__(self):
-        """Ensure subject ID has appropriate namespace prefix.
-
-        This allows initialization with either name= or subject_id= parameter.
-        """
+        """Ensure subject ID has appropriate namespace prefix."""
         if not self.subject_id:
             self.subject_id = f"{self.NAMESPACE}{self.SEPARATOR}{self.name}".lower()
 
         # Allow reverse lookup of name from subject_id
-        if not self.name and self.subject_id.startswith(f"{self.NAMESPACE}{self.SEPARATOR}"):
+        if not self.name and self.subject_id.startswith(
+            f"{self.NAMESPACE}{self.SEPARATOR}"
+        ):
             self.name = self.subject_id.split(self.SEPARATOR, 1)[1].lower()
 
 
@@ -140,7 +147,9 @@ def __attrs_post_init__(self):
             self.subject_id = f"{self.NAMESPACE}{self.SEPARATOR}{self.username}".lower()
 
         # Allow reverse lookup of username from subject_id
-        if not self.username and self.subject_id.startswith(f"{self.NAMESPACE}{self.SEPARATOR}"):
+        if not self.username and self.subject_id.startswith(
+            f"{self.NAMESPACE}{self.SEPARATOR}"
+        ):
             self.username = self.subject_id.split(self.SEPARATOR, 1)[1].lower()
 
 
@@ -165,7 +174,9 @@ def __attrs_post_init__(self):
             self.action_id = f"{self.NAMESPACE}{self.SEPARATOR}{self.name}".lower()
 
         # Allow reverse lookup of name from action_id
-        if not self.name and self.action_id.startswith(f"{self.NAMESPACE}{self.SEPARATOR}"):
+        if not self.name and self.action_id.startswith(
+            f"{self.NAMESPACE}{self.SEPARATOR}"
+        ):
             self.name = self.action_id.split(self.SEPARATOR, 1)[1].lower()
 
 
@@ -219,13 +230,18 @@ def __attrs_post_init__(self):
 
         This allows initialization with either name= or role_id= parameter.
         """
-        if not self.role_id or not self.role_id.startswith(f"{self.NAMESPACE}{self.SEPARATOR}"):
+        if not self.role_id or not self.role_id.startswith(
+            f"{self.NAMESPACE}{self.SEPARATOR}"
+        ):
             self.role_id = f"{self.NAMESPACE}{self.SEPARATOR}{self.name}".lower()
 
         # Allow reverse lookup of name from role_id
-        if not self.name and self.role_id.startswith(f"{self.NAMESPACE}{self.SEPARATOR}"):
+        if not self.name and self.role_id.startswith(
+            f"{self.NAMESPACE}{self.SEPARATOR}"
+        ):
             self.name = self.role_id.split(self.SEPARATOR, 1)[1].lower()
 
+
 @define
 class RoleAssignmentData(AuthZData):
     """A role assignment is the assignment of a role to a subject in a specific scope.
diff --git a/openedx_authz/api/roles.py b/openedx_authz/api/roles.py
@@ -247,7 +247,9 @@ def get_subject_role_assignments(subject: SubjectData) -> list[RoleAssignmentDat
         GroupingPolicyIndex.SUBJECT.value, subject.subject_id
     ):
         role = RoleData(role_id=policy[GroupingPolicyIndex.ROLE.value])
-        role.permissions = get_permissions_for_roles(role)[role.name][  # Index by role name for readability
+        role.permissions = get_permissions_for_roles(role)[
+            role.name
+        ][  # Index by role name for readability
             "permissions"
         ]
 
@@ -307,7 +309,7 @@ def get_subjects_role_assignments_for_role_in_scope(
     """
     role_assignments = []
     for subject in enforcer.get_users_for_role_in_domain(role.role_id, scope.scope_id):
-        if subject.startswith(RoleData.NAMESPACE):
+        if subject.startswith(f"{RoleData.NAMESPACE}@"):
             # Skip roles that are also subjects
             continue
         role_assignments.append(
diff --git a/openedx_authz/api/users.py b/openedx_authz/api/users.py
@@ -9,7 +9,13 @@
 (e.g., 'user@john_doe').
 """
 
-from openedx_authz.api.data import RoleAssignmentData, RoleData, ScopeData, SubjectData, UserData
+from openedx_authz.api.data import (
+    RoleAssignmentData,
+    RoleData,
+    ScopeData,
+    SubjectData,
+    UserData,
+)
 from openedx_authz.api.roles import (
     assign_role_to_subject_in_scope,
     batch_assign_role_to_subjects_in_scope,
@@ -104,7 +110,9 @@ def get_user_role_assignments(username: str) -> list[RoleAssignmentData]:
     return get_subject_role_assignments(UserData(username=username))
 
 
-def get_user_role_assignments_in_scope(username: str, scope: str) -> list[RoleAssignmentData]:
+def get_user_role_assignments_in_scope(
+    username: str, scope: str
+) -> list[RoleAssignmentData]:
     """Get the roles assigned to a user in a specific scope.
 
     Args:
@@ -118,7 +126,10 @@ def get_user_role_assignments_in_scope(username: str, scope: str) -> list[RoleAs
         UserData(username=username), ScopeData(name=scope)
     )
 
-def get_user_role_assignments_for_role_in_scope(role_name:str, scope:str) -> list[RoleAssignmentData]:
+
+def get_user_role_assignments_for_role_in_scope(
+    role_name: str, scope: str
+) -> list[RoleAssignmentData]:
     """Get all users assigned to a specific role across all scopes.
 
     Args:
@@ -128,4 +139,20 @@ def get_user_role_assignments_for_role_in_scope(role_name:str, scope:str) -> lis
     Returns:
         list[dict]: A list of user names and all their metadata assigned to the role.
     """
-    return get_subjects_role_assignments_for_role_in_scope(RoleData(name=role_name), ScopeData(name=scope))
+    # TODO: this SHOULD definitely be managed in a better way by using class inheritance and factories
+    # But for now we'll keep it simple and explicit
+    user_role_assignments = []
+    for role_assignment in get_subjects_role_assignments_for_role_in_scope(
+        RoleData(name=role_name), ScopeData(name=scope)
+    ):
+        print(role_assignment)
+        user_role_assignments.append(
+            RoleAssignmentData(
+                subject=UserData(
+                    subject_id=role_assignment.subject.subject_id
+                ),  # TODO: this gets the username from the subject_id
+                role=role_assignment.role,
+                scope=role_assignment.scope,
+            )
+        )
+    return user_role_assignments
diff --git a/openedx_authz/settings/common.py b/openedx_authz/settings/common.py
@@ -22,7 +22,9 @@ def plugin_settings(settings):
         settings.INSTALLED_APPS.append(casbin_adapter_app)
 
     # Add Casbin configuration
-    settings.CASBIN_MODEL = os.path.join(ROOT_DIRECTORY, "engine", "config", "model.conf")
+    settings.CASBIN_MODEL = os.path.join(
+        ROOT_DIRECTORY, "engine", "config", "model.conf"
+    )
     settings.CASBIN_WATCHER_ENABLED = True
     # TODO: Replace with a more dynamic configuration
     # Redis host and port are temporarily loaded here for the MVP
diff --git a/openedx_authz/tests/api/test_roles.py b/openedx_authz/tests/api/test_roles.py
@@ -117,11 +117,6 @@ def setUpClass(cls):
                 "role_name": "library_collaborator",
                 "scope_name": "math_advanced",
             },
-            {
-                "subject_name": "Henry",
-                "role_name": "library_collaborator",
-                "scope_name": "math_advanced",
-            },
             # Hierarchical scope_id assignments - different specificity levels
             {
                 "subject_name": "Ivy",
@@ -195,7 +190,7 @@ def setUpClass(cls):
                 "subject_name": "Frank",
                 "role_name": "library_user",
                 "scope_name": "project_epsilon",
-            }
+            },
         ]
         cls._seed_database_with_policies()
         cls._assign_roles_to_users(assignments=assignments)
@@ -449,9 +444,7 @@ def test_get_permissions_for_roles(self, role_name, expected_permissions):
             "library_user",
             "english_101",
             [
-                PermissionData(
-                    action=ActionData(name="view_library"), effect="allow"
-                ),
+                PermissionData(action=ActionData(name="view_library"), effect="allow"),
                 PermissionData(
                     action=ActionData(name="view_library_team"), effect="allow"
                 ),
@@ -474,9 +467,7 @@ def test_get_permissions_for_roles(self, role_name, expected_permissions):
                     action=ActionData(name="publish_library_content"),
                     effect="allow",
                 ),
-                PermissionData(
-                    action=ActionData(name="edit_library"), effect="allow"
-                ),
+                PermissionData(action=ActionData(name="edit_library"), effect="allow"),
                 PermissionData(
                     action=ActionData(name="manage_library_tags"),
                     effect="allow",
@@ -581,7 +572,9 @@ def test_get_roles_in_scope(self, scope_name, expected_roles):
         # Need to cheat here and use library data class to get lib@* scope_name
         # TODO: it'd be better to have our own policies for testing but for now we're using
         # the existing ones in authz.policy
-        roles_in_scope = get_role_definitions_in_scope(ContentLibraryData(library_id=scope_name))
+        roles_in_scope = get_role_definitions_in_scope(
+            ContentLibraryData(library_id=scope_name)
+        )
 
         role_names = {role.name for role in roles_in_scope}
         self.assertEqual(role_names, expected_roles)
@@ -595,7 +588,6 @@ def test_get_roles_in_scope(self, scope_name, expected_roles):
         ("eve", "chemistry_501", {"library_author"}),
         ("eve", "biology_601", {"library_user"}),
         ("grace", "math_advanced", {"library_collaborator"}),
-        ("henry", "math_advanced", {"library_collaborator"}),
         ("ivy", "cs_101", {"library_admin"}),
         ("jack", "cs_101", {"library_author"}),
         ("kate", "cs_101", {"library_user"}),
@@ -797,9 +789,7 @@ def test_get_subject_role_assignments_in_scope(
         ("non_existent_user", []),
     )
     @unpack
-    def test_get_all_role_assignments_scopes(
-        self, subject_name, expected_roles
-    ):
+    def test_get_all_role_assignments_scopes(self, subject_name, expected_roles):
         """Test retrieving all roles assigned to a subject across all scopes.
 
         Expected result:
@@ -851,7 +841,7 @@ def test_get_role_assignments_in_scope(self, role_name, scope_name, expected_cou
         Expected result:
             - The number of role assignments in the given scope is correctly retrieved.
         """
-        role_assignments = get_subject_role_assignments_for_role_in_scope(
+        role_assignments = get_subjects_role_assignments_for_role_in_scope(
             RoleData(name=role_name), ScopeData(name=scope_name)
         )
 
diff --git a/openedx_authz/tests/api/test_users.py b/openedx_authz/tests/api/test_users.py
@@ -57,9 +57,7 @@ def test_assign_role_to_user_in_scope(self, username, role, scope_name, batch):
             - The role is successfully assigned to the user in the specified scope.
         """
         if batch:
-            batch_assign_role_to_users(
-                users=username, role_name=role, scope=scope_name
-            )
+            batch_assign_role_to_users(users=username, role_name=role, scope=scope_name)
             for user in username:
                 user_roles = get_user_role_assignments_in_scope(
                     username=user, scope=scope_name
@@ -77,7 +75,7 @@ def test_assign_role_to_user_in_scope(self, username, role, scope_name, batch):
             self.assertIn(role, role_names)
 
     @data(
-        (["Grace", "Henry"], "library_collaborator", "math_advanced", True),
+        (["Grace"], "library_collaborator", "math_advanced", True),
         (["Liam", "Maya"], "library_author", "art_101", True),
         ("Alice", "library_admin", "math_101", False),
         ("Bob", "library_author", "history_201", False),
@@ -101,9 +99,7 @@ def test_unassign_role_from_user(self, username, role, scope_name, batch):
                 role_names = {assignment.role.name for assignment in user_roles}
                 self.assertNotIn(role, role_names)
         else:
-            unassign_role_from_user(
-                user=username, role_name=role, scope=scope_name
-            )
+            unassign_role_from_user(user=username, role_name=role, scope=scope_name)
             user_roles = get_user_role_assignments_in_scope(
                 username=username, scope=scope_name
             )
@@ -136,7 +132,9 @@ def test_get_user_role_assignments(self, username, expected_roles):
         ("Grace", "math_advanced", {"library_collaborator"}),
     )
     @unpack
-    def test_get_user_role_assignments_in_scope(self, username, scope_name, expected_roles):
+    def test_get_user_role_assignments_in_scope(
+        self, username, scope_name, expected_roles
+    ):
         """Test retrieving role assignments for a user within a specific scope.
 
         Expected result:
@@ -150,14 +148,15 @@ def test_get_user_role_assignments_in_scope(self, username, scope_name, expected
         role_names = {assignment.role.name for assignment in user_roles}
         self.assertEqual(role_names, expected_roles)
 
-
     @data(
-        ("library_admin", "math_101", {"Alice", "Eve"}),
-        ("library_author", "history_201", {"Bob"}),
-        ("library_collaborator", "math_advanced", {"Grace"}),
+        ("library_admin", "math_101", {"alice"}),
+        ("library_author", "history_201", {"bob"}),
+        ("library_collaborator", "math_advanced", {"grace"}),
     )
     @unpack
-    def test_get_user_role_assignments_for_role_in_scope(self, role_name, scope_name, expected_users):
+    def test_get_user_role_assignments_for_role_in_scope(
+        self, role_name, scope_name, expected_users
+    ):
         """Test retrieving all users assigned to a specific role within a specific scope.
 
         Expected result:
@@ -168,5 +167,8 @@ def test_get_user_role_assignments_for_role_in_scope(self, role_name, scope_name
             role_name=role_name, scope=scope_name
         )
 
-        assigned_usernames = {assignment.subject.username for assignment in user_assignments}
+        assigned_usernames = {
+            assignment.subject.username for assignment in user_assignments
+        }
+
         self.assertEqual(assigned_usernames, expected_users)
