diff --git a/openedx/core/djangoapps/enrollments/v2/tests/test_views.py b/openedx/core/djangoapps/enrollments/v2/tests/test_views.py index 414aa8075926..d6891b35600a 100644 --- a/openedx/core/djangoapps/enrollments/v2/tests/test_views.py +++ b/openedx/core/djangoapps/enrollments/v2/tests/test_views.py @@ -14,10 +14,11 @@ from django.test import override_settings from django.urls import reverse +from opaque_keys.edx.keys import CourseKey from rest_framework import status from rest_framework.test import APITestCase -from common.djangoapps.student.tests.factories import AdminFactory, UserFactory +from common.djangoapps.student.tests.factories import AdminFactory, CourseEnrollmentFactory, UserFactory from openedx.core.djangoapps.enrollments.v2.views import EnrollmentViewSet from openedx.core.djangolib.testing.utils import skip_unless_lms @@ -287,3 +288,79 @@ def test_minimal_view_collapses_course_details_to_course_id(self, mock_list, moc assert {r["course_id"] for r in response.data["results"]} == { "course-v1:org+a+r", "course-v1:org+b+r", } + + +# --------------------------------------------------------------------------- +# EnrollmentsAdminListView (GET /enrollments/) +# --------------------------------------------------------------------------- +@skip_unless_lms +class TestEnrollmentsAdminListView(APITestCase): + """ + Regression tests for the admin enrollment list. + + Covers endpoint access, the pass-through queryset scoping, the query-param + filters (applied in ``filter_queryset``, so form validation still yields a + 400), and the ADR 0033 ``Deprecation`` header. Uses real + ``CourseEnrollment`` rows (SQL, MongoDB-free). + """ + + def setUp(self): + super().setUp() + self.admin = AdminFactory.create(password="test") + self.user = UserFactory.create(password="test") + self.url = reverse("v2:enrollment-v2-admin-list") + self.course_a = CourseKey.from_string("course-v1:edX+A+run") + self.course_b = CourseKey.from_string("course-v1:edX+B+run") + self.learner_a = UserFactory.create() + self.learner_b = UserFactory.create() + CourseEnrollmentFactory.create(user=self.learner_a, course_id=self.course_a) + CourseEnrollmentFactory.create(user=self.learner_b, course_id=self.course_b) + + def test_unauthenticated_gets_401(self): + assert self.client.get(self.url).status_code == status.HTTP_401_UNAUTHORIZED + + def test_non_admin_gets_403(self): + """Endpoint-access layer: IsAdminUser rejects a regular user.""" + self.client.force_authenticate(user=self.user) + assert self.client.get(self.url).status_code == status.HTTP_403_FORBIDDEN + + def test_admin_sees_all_rows(self): + """Record-visibility layer is a pass-through: admin sees every enrollment.""" + self.client.force_authenticate(user=self.admin) + response = self.client.get(self.url) + assert response.status_code == status.HTTP_200_OK + assert response.data["count"] == 2 + + def test_filter_by_course_key_narrows(self): + """User-driven filter (filter_queryset) still narrows by course_key.""" + self.client.force_authenticate(user=self.admin) + response = self.client.get(self.url, {"course_key": str(self.course_a)}) + assert response.status_code == status.HTTP_200_OK + assert response.data["count"] == 1 + + def test_filter_by_username_narrows(self): + """User-driven filter (filter_queryset) still narrows by username.""" + self.client.force_authenticate(user=self.admin) + response = self.client.get(self.url, {"username": self.learner_b.username}) + assert response.status_code == status.HTTP_200_OK + assert response.data["count"] == 1 + + def test_non_admin_cannot_list_another_users_enrollments(self): + """A regular user cannot read someone else's enrollments by naming them in the filter.""" + self.client.force_authenticate(user=self.learner_a) + response = self.client.get(self.url, {"username": self.learner_b.username}) + assert response.status_code == status.HTTP_403_FORBIDDEN + assert "results" not in response.data + + def test_invalid_course_key_gets_400(self): + """Form validation moved to filter_queryset must still surface as a 400.""" + self.client.force_authenticate(user=self.admin) + response = self.client.get(self.url, {"course_key": "not-a-course-key"}) + assert response.status_code == status.HTTP_400_BAD_REQUEST + + def test_legacy_course_id_emits_deprecation_header(self): + """ADR 0033: the legacy ``course_id`` alias still emits the Deprecation header.""" + self.client.force_authenticate(user=self.admin) + response = self.client.get(self.url, {"course_id": str(self.course_a)}) + assert response.status_code == status.HTTP_200_OK + assert "Deprecation" in response.headers diff --git a/openedx/core/djangoapps/enrollments/v2/views.py b/openedx/core/djangoapps/enrollments/v2/views.py index 4b13e247d8f6..e03e04d04951 100644 --- a/openedx/core/djangoapps/enrollments/v2/views.py +++ b/openedx/core/djangoapps/enrollments/v2/views.py @@ -47,6 +47,7 @@ from edx_rest_framework_extensions.auth.jwt.authentication import JwtAuthentication from edx_rest_framework_extensions.mixins import StandardizedErrorMixin from edx_rest_framework_extensions.paginators import DefaultPagination, IterablePaginationMixin +from edx_rest_framework_extensions.scoping import FullScopePolicy, ScopedQuerysetMixin from edx_rest_framework_extensions.shaping import MinimalViewMixin from opaque_keys import InvalidKeyError from opaque_keys.edx.keys import CourseKey @@ -671,8 +672,15 @@ def get(self, request, course_id=None): 403: _RESP_FORBIDDEN, }, ) -class EnrollmentsAdminListView(StandardizedErrorMixin, ListAPIView): - """Admin-only paginated enrollment list with OEP-68 filter aliases.""" +class EnrollmentsAdminListView(ScopedQuerysetMixin, StandardizedErrorMixin, ListAPIView): + """ + Admin-only paginated enrollment list with OEP-68 filter aliases. + + Authorization is layered: ``permission_classes`` gates access to the + endpoint, ``ScopedQuerysetMixin`` applies ``scoping_policy`` to the base + ``queryset``, and ``filter_queryset`` narrows the scoped rows by the + caller's query parameters. + """ # ADR 0034 — JWT + cross-domain session (BearerAuthenticationAllowInactiveUser # removed per OEP-0042). EnrollmentCrossDomainSessionAuth retained because the @@ -687,6 +695,12 @@ class EnrollmentsAdminListView(StandardizedErrorMixin, ListAPIView): serializer_class = CourseEnrollmentsApiListSerializer pagination_class = EnrollmentsAdminListPagination + queryset = CourseEnrollment.objects.all().select_related("user", "course") + # Platform admins may see every enrollment, so the policy is a pass-through. The + # scoping layer stays wired so a narrower policy can be dropped in without touching + # the filtering below. + scoping_policy = FullScopePolicy() + # ADR 0033 §3 — whitelist of allowed values for the ``ordering`` param. ALLOWED_ORDERING_FIELDS = frozenset({"created", "-created", "id", "-id"}) @@ -696,12 +710,12 @@ class EnrollmentsAdminListView(StandardizedErrorMixin, ListAPIView): ("course_ids", "course_keys"), ) - def get_queryset(self): + def filter_queryset(self, queryset): + """Narrow the scoped queryset by the caller-supplied query parameters.""" form = EnrollmentsAdminListForm(self.request.query_params) if not form.is_valid(): raise ValidationError(form.errors) - queryset = CourseEnrollment.objects.all().select_related("user", "course") course_id = form.cleaned_data.get("course_id") course_ids = form.cleaned_data.get("course_ids") usernames = form.cleaned_data.get("username")