squash!: Support glob scopes

Author: rodmgwgu

openedx_authz/rest_api/v1/serializers.py

@@ -321,7 +321,7 @@ def get_external_key(self, obj: dict) -> str:
 
     def get_display_name(self, obj: dict) -> str:
         """Get the display name for the given scope."""
-        return str(obj.get("display_name") or "")
+        return str(obj.get("display_name_col") or "")
 
     def get_org(self, obj: dict) -> dict | None:
         """Get the org for the given scope."""

openedx_authz/rest_api/v1/views.py

@@ -23,7 +23,12 @@
 from rest_framework.views import APIView
 
 from openedx_authz import api
-from openedx_authz.api.data import ContentLibraryData, CourseOverviewData
+from openedx_authz.api.data import (
+    ContentLibraryData,
+    CourseOverviewData,
+    OrgContentLibraryGlobData,
+    OrgCourseOverviewGlobData,
+)
 from openedx_authz.api.users import get_scopes_for_user_and_permission
 from openedx_authz.api.utils import get_user_map
 from openedx_authz.constants import permissions
@@ -606,6 +611,7 @@ class AdminConsoleScopesAPIView(generics.ListAPIView):
     - search (Optional): Search term to filter scopes by display name
     - page (Optional): Page number for pagination
     - page_size (Optional): Number of items per page
+    - type (Optional): Filter scopes by type. Supported values are `course` and `library`.
     - management_permission_only (Optional): Filter scopes either by only the ones to which the user has "manage team"
         permissions (if true), or just "view team" permissions.
 
@@ -659,43 +665,57 @@ def get_serializer_context(self):
         context["org_map"] = Organization.objects.filter(active=True).in_bulk(field_name="short_name")
         return context
 
-    def _get_courses_queryset(self, allowed_ids: set | None = None, search: str = "") -> QuerySet:
+    def _get_courses_queryset(
+        self, allowed_ids: set | None = None, allowed_orgs: set | None = None, search: str = ""
+    ) -> QuerySet:
         """Return a CourseOverview queryset projected to the unified scope shape.
 
-        If allowed_ids is provided, filter to only those course keys.
+        If allowed_ids and/or allowed_orgs are provided, filter to matching courses.
         If search is provided, filter by display_name.
         """
         qs = CourseOverview.objects
-        if allowed_ids is not None:
-            qs = qs.filter(id__in=allowed_ids)
+        if allowed_ids is not None or allowed_orgs is not None:
+            org_filter = Q(org__in=allowed_orgs) if allowed_orgs else Q()
+            id_filter = Q(id__in=allowed_ids) if allowed_ids else Q()
+            qs = qs.filter(org_filter | id_filter)
         if search:
             qs = qs.filter(display_name__icontains=search)
         return qs.annotate(
-            scope_id=Cast("id", output_field=CharField()),
-            org_name=Cast("org", output_field=CharField()),
-            scope_type=Value("course", output_field=CharField()),
-        ).values("scope_id", "display_name", "org_name", "scope_type")
-
-    def _get_libraries_queryset(self, allowed_pairs: set | None = None, search: str = "") -> QuerySet:
+            scope_id=Cast("id", output_field=CharField(db_collation="utf8mb4_unicode_ci")),
+            display_name_col=Cast("display_name", output_field=CharField(db_collation="utf8mb4_unicode_ci")),
+            org_name=Cast("org", output_field=CharField(db_collation="utf8mb4_unicode_ci")),
+            scope_type=Value("course", output_field=CharField(db_collation="utf8mb4_unicode_ci")),
+        ).values("scope_id", "display_name_col", "org_name", "scope_type")
+
+    def _get_libraries_queryset(
+        self, allowed_pairs: set | None = None, allowed_orgs: set | None = None, search: str = ""
+    ) -> QuerySet:
         """Return a ContentLibrary queryset projected to the unified scope shape.
 
-        If allowed_pairs is provided, filter to only those (org, slug) pairs.
+        If allowed_pairs and/or allowed_orgs are provided, filter to matching libraries.
         If search is provided, filter by learning_package__title.
         """
         qs = ContentLibrary.objects
-        if allowed_pairs is not None:
-            if not allowed_pairs:
+        if allowed_pairs is not None or allowed_orgs is not None:
+            org_filter = Q(org__short_name__in=allowed_orgs) if allowed_orgs else Q()
+            pair_filter = (
+                reduce(operator.or_, (Q(org__short_name=org, slug=slug) for org, slug in allowed_pairs))
+                if allowed_pairs
+                else Q()
+            )
+            combined = org_filter | pair_filter
+            if not combined:
                 qs = qs.none()
             else:
-                qs = qs.filter(reduce(operator.or_, (Q(org__short_name=org, slug=slug) for org, slug in allowed_pairs)))
+                qs = qs.filter(combined)
         if search:
             qs = qs.filter(learning_package__title__icontains=search)
         return qs.annotate(
-            scope_id=Cast("slug", output_field=CharField()),
-            display_name=Cast("learning_package__title", output_field=CharField()),
-            org_name=Cast("org__short_name", output_field=CharField()),
-            scope_type=Value("library", output_field=CharField()),
-        ).values("scope_id", "display_name", "org_name", "scope_type")
+            scope_id=Cast("slug", output_field=CharField(db_collation="utf8mb4_unicode_ci")),
+            display_name_col=Cast("learning_package__title", output_field=CharField(db_collation="utf8mb4_unicode_ci")),
+            org_name=Cast("org__short_name", output_field=CharField(db_collation="utf8mb4_unicode_ci")),
+            scope_type=Value("library", output_field=CharField(db_collation="utf8mb4_unicode_ci")),
+        ).values("scope_id", "display_name_col", "org_name", "scope_type")
 
     def _build_queryset(self, courses_qs: QuerySet | None, libraries_qs: QuerySet | None) -> QuerySet:
         """Union the provided querysets and sort by org. If only one is provided, return it sorted directly."""
@@ -730,24 +750,28 @@ def get_permission(scope_cls):
         # Resolve allowed scopes from Casbin in a single call per scope type.
         courses_qs = None
         if scope_type != "library":
+            allowed_course_scopes = get_scopes_for_user_and_permission(
+                user.username, get_permission(CourseOverviewData).identifier
+            )
             allowed_course_ids = {
-                s.external_key
-                for s in get_scopes_for_user_and_permission(
-                    user.username, get_permission(CourseOverviewData).identifier
-                )
+                s.external_key for s in allowed_course_scopes if not isinstance(s, OrgCourseOverviewGlobData)
             }
-            courses_qs = self._get_courses_queryset(allowed_course_ids, search=search)
+            allowed_course_orgs = {s.org for s in allowed_course_scopes if isinstance(s, OrgCourseOverviewGlobData)}
+            courses_qs = self._get_courses_queryset(allowed_course_ids, allowed_course_orgs, search=search)
 
         libraries_qs = None
         if scope_type != "course":
+            allowed_library_scopes = get_scopes_for_user_and_permission(
+                user.username, get_permission(ContentLibraryData).identifier
+            )
             allowed_library_pairs = {
                 # Library external keys have the format lib:<org>:<slug>
                 (s.external_key.split(":")[1], s.external_key.split(":")[2])
-                for s in get_scopes_for_user_and_permission(
-                    user.username, get_permission(ContentLibraryData).identifier
-                )
+                for s in allowed_library_scopes
+                if not isinstance(s, OrgContentLibraryGlobData)
             }
-            libraries_qs = self._get_libraries_queryset(allowed_library_pairs, search=search)
+            allowed_library_orgs = {s.org for s in allowed_library_scopes if isinstance(s, OrgContentLibraryGlobData)}
+            libraries_qs = self._get_libraries_queryset(allowed_library_pairs, allowed_library_orgs, search=search)
 
         # Union the requested querysets and sort by org at the DB level.
         return self._build_queryset(courses_qs, libraries_qs)

openedx_authz/tests/rest_api/test_views.py

@@ -20,6 +20,10 @@
 from rest_framework.test import APIClient
 
 from openedx_authz import api
+from openedx_authz.api.data import (
+    OrgContentLibraryGlobData,
+    OrgCourseOverviewGlobData,
+)
 from openedx_authz.api.users import assign_role_to_user_in_scope
 from openedx_authz.constants import permissions, roles
 from openedx_authz.models.scopes import get_content_library_model, get_course_overview_model
@@ -937,10 +941,20 @@ def setUp(self):
 
         # Default combined result used by most tests.
         self.fake_scopes = [
-            {"scope_id": self.COURSE_ORG1, "display_name": "Course Org1", "org_name": "Org1", "scope_type": "course"},
-            {"scope_id": "LIB1", "display_name": "Library LIB1", "org_name": "Org1", "scope_type": "library"},
-            {"scope_id": self.COURSE_ORG2, "display_name": "Course Org2", "org_name": "Org2", "scope_type": "course"},
-            {"scope_id": "LIB2", "display_name": "Library LIB2", "org_name": "Org2", "scope_type": "library"},
+            {
+                "scope_id": self.COURSE_ORG1,
+                "display_name_col": "Course Org1",
+                "org_name": "Org1",
+                "scope_type": "course",
+            },
+            {"scope_id": "LIB1", "display_name_col": "Library LIB1", "org_name": "Org1", "scope_type": "library"},
+            {
+                "scope_id": self.COURSE_ORG2,
+                "display_name_col": "Course Org2",
+                "org_name": "Org2",
+                "scope_type": "course",
+            },
+            {"scope_id": "LIB2", "display_name_col": "Library LIB2", "org_name": "Org2", "scope_type": "library"},
         ]
 
         # Patch _build_queryset so tests don't need real DB querysets.
@@ -955,21 +969,26 @@ def setUp(self):
         # The stub ContentLibrary uses 'title' directly instead of 'learning_package__title'.
         # Patch _get_libraries_queryset to use the stub-compatible field, aliased as 'display_name'
         # to match the column name the union and serializer expect.
-        def stub_get_libraries_queryset(_, allowed_pairs=None, search=""):  # pylint: disable=unused-argument
+        def stub_get_libraries_queryset(_, allowed_pairs=None, allowed_orgs=None, search=""):  # pylint: disable=unused-argument
             qs = ContentLibrary.objects
-            if allowed_pairs is not None:
-                if not allowed_pairs:
+            if allowed_pairs is not None or allowed_orgs is not None:
+                org_filter = Q(org__short_name__in=allowed_orgs) if allowed_orgs else Q()
+                pair_filter = (
+                    reduce(operator.or_, (Q(org__short_name=org, slug=slug) for org, slug in allowed_pairs))
+                    if allowed_pairs
+                    else Q()
+                )
+                combined = org_filter | pair_filter
+                if not combined:
                     qs = qs.none()
                 else:
-                    qs = qs.filter(
-                        reduce(operator.or_, (Q(org__short_name=org, slug=slug) for org, slug in allowed_pairs))
-                    )
+                    qs = qs.filter(combined)
             return qs.annotate(
                 scope_id=Cast("slug", output_field=CharField()),
-                display_name=Cast("title", output_field=CharField()),
+                display_name_col=Cast("title", output_field=CharField()),
                 org_name=Cast("org__short_name", output_field=CharField()),
                 scope_type=Value("library", output_field=CharField()),
-            ).values("scope_id", "display_name", "org_name", "scope_type")
+            ).values("scope_id", "display_name_col", "org_name", "scope_type")
 
         self.libraries_qs_patcher = patch.object(
             views.AdminConsoleScopesAPIView,
@@ -1103,9 +1122,9 @@ def test_search_filters_by_display_name(self):
     def test_pagination(self, query_params: dict, expected_page_count: int, has_next: bool):
         """Results are paginated correctly."""
         mixed = [
-            {"scope_id": self.COURSE_ORG1, "display_name": "Course 1", "org_name": "Org1", "scope_type": "course"},
-            {"scope_id": "LIB1", "display_name": "Library 1", "org_name": "Org1", "scope_type": "library"},
-            {"scope_id": self.COURSE_ORG2, "display_name": "Course 2", "org_name": "Org2", "scope_type": "course"},
+            {"scope_id": self.COURSE_ORG1, "display_name_col": "Course 1", "org_name": "Org1", "scope_type": "course"},
+            {"scope_id": "LIB1", "display_name_col": "Library 1", "org_name": "Org1", "scope_type": "library"},
+            {"scope_id": self.COURSE_ORG2, "display_name_col": "Course 2", "org_name": "Org2", "scope_type": "course"},
         ]
         self.build_qs_patcher.stop()
         with patch.object(views.AdminConsoleScopesAPIView, "_build_queryset", return_value=mixed):
@@ -1266,6 +1285,48 @@ def test_empty_allowed_library_pairs_returns_no_libraries(self):
         self.assertEqual(response.status_code, status.HTTP_200_OK)
         self.assertEqual(response.data["count"], 0)
 
+    def test_org_glob_scope_returns_all_org_libraries(self):
+        """A user with an org-level glob permission (lib:ORG:*) sees all libraries in that org."""
+        user = User.objects.get(username="regular_1")
+        self.client.force_authenticate(user=user)
+        self.build_qs_patcher.stop()
+
+        # Simulate get_scopes_for_user_and_permission returning an org-level glob.
+        glob_scope = OrgContentLibraryGlobData(external_key="lib:Org1:*")
+        with patch(
+            "openedx_authz.rest_api.v1.views.get_scopes_for_user_and_permission",
+            return_value=[glob_scope],
+        ):
+            response = self.client.get(self.url, {"type": "library"})
+
+        self.build_qs_patcher.start()
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        external_keys = [item["external_key"] for item in response.data["results"]]
+        self.assertIn(self.LIBRARY_ORG1, external_keys)
+        self.assertNotIn(self.LIBRARY_ORG2, external_keys)
+
+    def test_org_glob_scope_returns_all_org_courses(self):
+        """A user with an org-level glob permission (course-v1:ORG+*) sees all courses in that org."""
+        user = User.objects.get(username="regular_9")
+        self.client.force_authenticate(user=user)
+        self.build_qs_patcher.stop()
+        self.libraries_qs_patcher.stop()
+
+        # Simulate get_scopes_for_user_and_permission returning an org-level glob.
+        glob_scope = OrgCourseOverviewGlobData(external_key="course-v1:Org1+*")
+        with patch(
+            "openedx_authz.rest_api.v1.views.get_scopes_for_user_and_permission",
+            return_value=[glob_scope],
+        ):
+            response = self.client.get(self.url, {"type": "course"})
+
+        self.libraries_qs_patcher.start()
+        self.build_qs_patcher.start()
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        external_keys = [item["external_key"] for item in response.data["results"]]
+        self.assertIn(self.COURSE_ORG1, external_keys)
+        self.assertNotIn(self.COURSE_ORG2, external_keys)
+
     def test_manage_permission_only_uses_manage_permission(self):
         """management_permission_only=true calls get_admin_manage_permission, not get_admin_view_permission."""
         user = User.objects.get(username="regular_1")