From a2684c7d1e3e1bb526015ef035040272be829780 Mon Sep 17 00:00:00 2001 From: Brian Buck Date: Wed, 11 Mar 2026 13:42:48 -0600 Subject: [PATCH 1/3] feat: Add enrollments v2 list endpoint --- .../tests/test_enrollment_list_api.py | 468 ++++++++++++++++++ lms/djangoapps/instructor/views/api.py | 129 ++++- lms/djangoapps/instructor/views/api_urls.py | 1 + lms/djangoapps/instructor/views/serializer.py | 12 + 4 files changed, 608 insertions(+), 2 deletions(-) create mode 100644 lms/djangoapps/instructor/tests/test_enrollment_list_api.py diff --git a/lms/djangoapps/instructor/tests/test_enrollment_list_api.py b/lms/djangoapps/instructor/tests/test_enrollment_list_api.py new file mode 100644 index 000000000000..722b06ea2c25 --- /dev/null +++ b/lms/djangoapps/instructor/tests/test_enrollment_list_api.py @@ -0,0 +1,468 @@ +""" +Unit tests for instructor API enrollment list endpoints with search and pagination. +""" +import json +from django.urls import reverse +from common.djangoapps.student.roles import CourseBetaTesterRole +from common.djangoapps.student.tests.factories import ( + CourseEnrollmentFactory, + InstructorFactory, + UserFactory +) +from xmodule.modulestore.tests.django_utils import SharedModuleStoreTestCase +from xmodule.modulestore.tests.factories import CourseFactory +from lms.djangoapps.courseware.tests.helpers import LoginEnrollmentTestCase + + +class TestListCourseRoleMembersWithPagination(SharedModuleStoreTestCase, LoginEnrollmentTestCase): + """ + Test the list_course_role_members endpoint with search and pagination functionality. + """ + @classmethod + def setUpClass(cls): + super().setUpClass() + cls.course = CourseFactory.create() + + def setUp(self): + super().setUp() + self.instructor = InstructorFactory(course_key=self.course.id) + self.client.login(username=self.instructor.username, password=self.TEST_PASSWORD) + self.url = reverse('list_course_role_members', kwargs={'course_id': str(self.course.id)}) + + # Create beta testers for testing + self.beta_testers = [] + beta_role = CourseBetaTesterRole(self.course.id) + for i in range(25): + user = UserFactory( + username=f'beta_user_{i}', + email=f'beta{i}@example.com', + first_name=f'Beta{i}', + last_name=f'Tester{i}' + ) + beta_role.add_users(user) + self.beta_testers.append(user) + + def test_list_beta_testers_without_pagination(self): + """Test listing beta testers without pagination parameters (backward compatibility).""" + response = self.client.post(self.url, {'rolename': 'beta'}) + assert response.status_code == 200 + + res_json = json.loads(response.content.decode('utf-8')) + assert res_json['course_id'] == str(self.course.id) + assert 'beta' in res_json + assert res_json['count'] == 25 + assert res_json['num_pages'] == 2 # 25 items with default page_size of 20 + assert res_json['current_page'] == 1 + assert len(res_json['beta']) == 20 # First page with default page_size + + def test_list_beta_testers_with_pagination(self): + """Test listing beta testers with pagination.""" + response = self.client.post(self.url, { + 'rolename': 'beta', + 'page': 1, + 'page_size': 10 + }) + assert response.status_code == 200 + + res_json = json.loads(response.content.decode('utf-8')) + assert res_json['count'] == 25 + assert res_json['num_pages'] == 3 # 25 items / 10 per page + assert res_json['current_page'] == 1 + assert len(res_json['beta']) == 10 + + def test_list_beta_testers_second_page(self): + """Test listing beta testers on second page.""" + response = self.client.post(self.url, { + 'rolename': 'beta', + 'page': 2, + 'page_size': 10 + }) + assert response.status_code == 200 + + res_json = json.loads(response.content.decode('utf-8')) + assert res_json['current_page'] == 2 + assert len(res_json['beta']) == 10 + + def test_list_beta_testers_last_page(self): + """Test listing beta testers on last page with partial results.""" + response = self.client.post(self.url, { + 'rolename': 'beta', + 'page': 3, + 'page_size': 10 + }) + assert response.status_code == 200 + + res_json = json.loads(response.content.decode('utf-8')) + assert res_json['current_page'] == 3 + assert len(res_json['beta']) == 5 # Last page has 5 items + + def test_list_beta_testers_beyond_last_page(self): + """Test requesting a page beyond the last page returns empty results.""" + response = self.client.post(self.url, { + 'rolename': 'beta', + 'page': 10, + 'page_size': 10 + }) + assert response.status_code == 200 + + res_json = json.loads(response.content.decode('utf-8')) + assert res_json['current_page'] == 10 + assert len(res_json['beta']) == 0 + + def test_list_beta_testers_search_by_username(self): + """Test searching beta testers by username.""" + response = self.client.post(self.url, { + 'rolename': 'beta', + 'search': 'beta_user_1', + 'page': 1, + 'page_size': 10 + }) + assert response.status_code == 200 + + res_json = json.loads(response.content.decode('utf-8')) + # Should match beta_user_1, beta_user_10-19 (11 total) + assert res_json['count'] == 11 + assert len(res_json['beta']) == 10 # First page + for user in res_json['beta']: + assert 'beta_user_1' in user['username'] + + def test_list_beta_testers_search_by_email(self): + """Test searching beta testers by email.""" + response = self.client.post(self.url, { + 'rolename': 'beta', + 'search': 'beta5@example.com', + 'page': 1, + 'page_size': 10 + }) + assert response.status_code == 200 + + res_json = json.loads(response.content.decode('utf-8')) + assert res_json['count'] == 1 + assert len(res_json['beta']) == 1 + assert res_json['beta'][0]['email'] == 'beta5@example.com' + + def test_list_beta_testers_search_by_first_name(self): + """Test searching beta testers by first name.""" + response = self.client.post(self.url, { + 'rolename': 'beta', + 'search': 'Beta2', + 'page': 1, + 'page_size': 10 + }) + assert response.status_code == 200 + + res_json = json.loads(response.content.decode('utf-8')) + # Should match Beta2, Beta20-24 (6 total) + assert res_json['count'] == 6 + for user in res_json['beta']: + assert 'Beta2' in user['first_name'] + + def test_list_beta_testers_search_case_insensitive(self): + """Test that search is case-insensitive.""" + response = self.client.post(self.url, { + 'rolename': 'beta', + 'search': 'BETA_USER_3', + 'page': 1, + 'page_size': 10 + }) + assert response.status_code == 200 + + res_json = json.loads(response.content.decode('utf-8')) + assert res_json['count'] == 1 + assert res_json['beta'][0]['username'] == 'beta_user_3' + + def test_list_beta_testers_search_no_results(self): + """Test searching with no matching results.""" + response = self.client.post(self.url, { + 'rolename': 'beta', + 'search': 'nonexistent', + 'page': 1, + 'page_size': 10 + }) + assert response.status_code == 200 + + res_json = json.loads(response.content.decode('utf-8')) + assert res_json['count'] == 0 + assert len(res_json['beta']) == 0 + + def test_list_beta_testers_empty_search(self): + """Test that empty search returns all results.""" + response = self.client.post(self.url, { + 'rolename': 'beta', + 'search': '', + 'page': 1, + 'page_size': 10 + }) + assert response.status_code == 200 + + res_json = json.loads(response.content.decode('utf-8')) + assert res_json['count'] == 25 + + def test_list_beta_testers_max_page_size(self): + """Test that page_size is capped at maximum.""" + response = self.client.post(self.url, { + 'rolename': 'beta', + 'page': 1, + 'page_size': 100 + }) + assert response.status_code == 200 + + res_json = json.loads(response.content.decode('utf-8')) + assert len(res_json['beta']) == 25 # All results fit in max page size + + def test_list_beta_testers_invalid_page_size(self): + """Test with invalid page_size (should fail validation).""" + response = self.client.post(self.url, { + 'rolename': 'beta', + 'page': 1, + 'page_size': 0 + }) + assert response.status_code == 400 + + def test_list_beta_testers_invalid_page(self): + """Test with invalid page number (should fail validation).""" + response = self.client.post(self.url, { + 'rolename': 'beta', + 'page': 0, + 'page_size': 10 + }) + assert response.status_code == 400 + + +class TestListCourseEnrollments(SharedModuleStoreTestCase, LoginEnrollmentTestCase): + """ + Test the list_course_enrollments endpoint with search and pagination functionality. + """ + @classmethod + def setUpClass(cls): + super().setUpClass() + cls.course = CourseFactory.create() + + def setUp(self): + super().setUp() + self.instructor = InstructorFactory(course_key=self.course.id) + self.client.login(username=self.instructor.username, password=self.TEST_PASSWORD) + self.url = reverse('list_course_enrollments', kwargs={'course_id': str(self.course.id)}) + + # Create enrollments for testing + self.enrolled_users = [] + for i in range(30): + user = UserFactory( + username=f'student_{i}', + email=f'student{i}@example.com', + first_name=f'Student{i}', + last_name=f'Learner{i}' + ) + CourseEnrollmentFactory( + user=user, + course_id=self.course.id, + is_active=True + ) + self.enrolled_users.append(user) + + # Create some inactive enrollments (should not be included) + for i in range(5): + user = UserFactory( + username=f'inactive_{i}', + email=f'inactive{i}@example.com' + ) + CourseEnrollmentFactory( + user=user, + course_id=self.course.id, + is_active=False + ) + + def test_list_enrollments_without_pagination(self): + """Test listing enrollments without pagination parameters.""" + response = self.client.post(self.url, {}) + assert response.status_code == 200 + + res_json = json.loads(response.content.decode('utf-8')) + assert res_json['course_id'] == str(self.course.id) + assert 'enrollments' in res_json + # 30 active student enrollments (InstructorFactory does not create an enrollment) + assert res_json['count'] == 30 + assert res_json['num_pages'] == 2 # 30 items with default page_size of 20 + assert res_json['current_page'] == 1 + assert len(res_json['enrollments']) == 20 + + def test_list_enrollments_with_pagination(self): + """Test listing enrollments with pagination.""" + response = self.client.post(self.url, { + 'page': 1, + 'page_size': 10 + }) + assert response.status_code == 200 + + res_json = json.loads(response.content.decode('utf-8')) + assert res_json['count'] == 30 + assert res_json['num_pages'] == 3 # 30 items / 10 per page = 3 pages + assert res_json['current_page'] == 1 + assert len(res_json['enrollments']) == 10 + + def test_list_enrollments_second_page(self): + """Test listing enrollments on second page.""" + response = self.client.post(self.url, { + 'page': 2, + 'page_size': 10 + }) + assert response.status_code == 200 + + res_json = json.loads(response.content.decode('utf-8')) + assert res_json['current_page'] == 2 + assert len(res_json['enrollments']) == 10 + + def test_list_enrollments_last_page(self): + """Test listing enrollments on last page with partial results.""" + response = self.client.post(self.url, { + 'page': 3, + 'page_size': 10 + }) + assert response.status_code == 200 + + res_json = json.loads(response.content.decode('utf-8')) + assert res_json['current_page'] == 3 + assert len(res_json['enrollments']) == 10 # Last page has 10 items + + def test_list_enrollments_search_by_username(self): + """Test searching enrollments by username.""" + response = self.client.post(self.url, { + 'search': 'student_2', + 'page': 1, + 'page_size': 10 + }) + assert response.status_code == 200 + + res_json = json.loads(response.content.decode('utf-8')) + # Should match student_2, student_20-29 (11 total) + assert res_json['count'] == 11 + for user in res_json['enrollments']: + assert 'student_2' in user['username'] + + def test_list_enrollments_search_by_email(self): + """Test searching enrollments by email.""" + response = self.client.post(self.url, { + 'search': 'student7@example.com', + 'page': 1, + 'page_size': 10 + }) + assert response.status_code == 200 + + res_json = json.loads(response.content.decode('utf-8')) + assert res_json['count'] == 1 + assert res_json['enrollments'][0]['email'] == 'student7@example.com' + + def test_list_enrollments_search_by_first_name(self): + """Test searching enrollments by first name.""" + response = self.client.post(self.url, { + 'search': 'Student1', + 'page': 1, + 'page_size': 20 + }) + assert response.status_code == 200 + + res_json = json.loads(response.content.decode('utf-8')) + # Should match Student1, Student10-19 (11 total) + assert res_json['count'] == 11 + + def test_list_enrollments_search_case_insensitive(self): + """Test that search is case-insensitive.""" + response = self.client.post(self.url, { + 'search': 'STUDENT_5', + 'page': 1, + 'page_size': 10 + }) + assert response.status_code == 200 + + res_json = json.loads(response.content.decode('utf-8')) + assert res_json['count'] == 1 + assert res_json['enrollments'][0]['username'] == 'student_5' + + def test_list_enrollments_search_no_results(self): + """Test searching with no matching results.""" + response = self.client.post(self.url, { + 'search': 'nonexistent', + 'page': 1, + 'page_size': 10 + }) + assert response.status_code == 200 + + res_json = json.loads(response.content.decode('utf-8')) + assert res_json['count'] == 0 + assert len(res_json['enrollments']) == 0 + + def test_list_enrollments_excludes_inactive(self): + """Test that inactive enrollments are not included.""" + response = self.client.post(self.url, { + 'search': 'inactive', + 'page': 1, + 'page_size': 10 + }) + assert response.status_code == 200 + + res_json = json.loads(response.content.decode('utf-8')) + # Should not find any inactive enrollments + assert res_json['count'] == 0 + + def test_list_enrollments_empty_course(self): + """Test listing enrollments for a course with no enrollments.""" + empty_course = CourseFactory.create() + empty_instructor = InstructorFactory(course_key=empty_course.id) + self.client.login(username=empty_instructor.username, password=self.TEST_PASSWORD) + + url = reverse('list_course_enrollments', kwargs={'course_id': str(empty_course.id)}) + response = self.client.post(url, {}) + assert response.status_code == 200 + + res_json = json.loads(response.content.decode('utf-8')) + # InstructorFactory does not create an enrollment, so empty course has 0 + assert res_json['count'] == 0 + assert len(res_json['enrollments']) == 0 + + def test_list_enrollments_max_page_size(self): + """Test that page_size is capped at maximum.""" + response = self.client.post(self.url, { + 'page': 1, + 'page_size': 100 + }) + assert response.status_code == 200 + + res_json = json.loads(response.content.decode('utf-8')) + assert len(res_json['enrollments']) == 30 # All results fit in max page size + + def test_list_enrollments_invalid_page_size(self): + """Test with invalid page_size (should fail validation).""" + response = self.client.post(self.url, { + 'page': 1, + 'page_size': 0 + }) + assert response.status_code == 400 + + def test_list_enrollments_invalid_page(self): + """Test with invalid page number (should fail validation).""" + response = self.client.post(self.url, { + 'page': -1, + 'page_size': 10 + }) + assert response.status_code == 400 + + def test_list_enrollments_permission_required(self): + """Test that non-instructor users cannot access the endpoint.""" + student = UserFactory() + self.client.login(username=student.username, password=self.TEST_PASSWORD) + + response = self.client.post(self.url, {}) + assert response.status_code == 403 + + def test_list_enrollments_ordered_by_username(self): + """Test that enrollments are ordered by username.""" + response = self.client.post(self.url, { + 'page': 1, + 'page_size': 5 + }) + assert response.status_code == 200 + + res_json = json.loads(response.content.decode('utf-8')) + usernames = [user['username'] for user in res_json['enrollments']] + # Check that usernames are in alphabetical order + assert usernames == sorted(usernames) diff --git a/lms/djangoapps/instructor/views/api.py b/lms/djangoapps/instructor/views/api.py index 09b91d362b28..b089e48fbc05 100644 --- a/lms/djangoapps/instructor/views/api.py +++ b/lms/djangoapps/instructor/views/api.py @@ -22,6 +22,7 @@ from django.core.exceptions import MultipleObjectsReturned, ObjectDoesNotExist, PermissionDenied, ValidationError from django.core.validators import validate_email from django.db import IntegrityError, transaction +from django.db.models import Q from django.http import QueryDict, HttpResponse, HttpResponseBadRequest, HttpResponseForbidden, HttpResponseNotFound from django.shortcuts import redirect from django.urls import reverse @@ -103,6 +104,7 @@ BlockDueDateSerializer, CertificateSerializer, CertificateStatusesSerializer, + EnrollmentListSerializer, ForumRoleNameSerializer, ListInstructorTaskInputSerializer, ModifyAccessSerializer, @@ -1050,6 +1052,11 @@ class ListCourseRoleMembersView(APIView): rolename is one of ['instructor', 'staff', 'beta', 'ccx_coach'] + Supports optional search and pagination parameters: + - search: Filter users by username, email, first_name, or last_name + - page: Page number (default: 1) + - page_size: Number of results per page (default: 20, max: 100) + Returns JSON of the form { "course_id": "some/course/id", "staff": [ @@ -1059,7 +1066,10 @@ class ListCourseRoleMembersView(APIView): "first_name": "Joe", "last_name": "Shmoe", } - ] + ], + "count": 10, + "num_pages": 1, + "current_page": 1 } """ permission_classes = (IsAuthenticated, permissions.InstructorPermission) @@ -1087,13 +1097,128 @@ def post(self, request, course_id): role_serializer = RoleNameSerializer(data=request.data) role_serializer.is_valid(raise_exception=True) rolename = role_serializer.data['rolename'] + search = role_serializer.data.get('search', '').strip() + page = role_serializer.data.get('page', 1) + page_size = role_serializer.data.get('page_size', 20) users = list_with_level(course.id, rolename) - serializer = UserSerializer(users, many=True) + + # Apply search filter + if search: + users = [ + user for user in users + if search.lower() in user.username.lower() + or search.lower() in user.email.lower() + or search.lower() in (user.first_name or '').lower() + or search.lower() in (user.last_name or '').lower() + ] + + # Calculate pagination + total_count = len(users) + num_pages = (total_count + page_size - 1) // page_size if page_size > 0 else 1 + start_idx = (page - 1) * page_size + end_idx = start_idx + page_size + paginated_users = users[start_idx:end_idx] + + serializer = UserSerializer(paginated_users, many=True) response_payload = { 'course_id': str(course_id), rolename: serializer.data, + 'count': total_count, + 'num_pages': num_pages, + 'current_page': page, + } + + return Response(response_payload, status=status.HTTP_200_OK) + + +class ListCourseEnrollmentsView(APIView): + """ + View to list all enrollments (learners/students) for a specific course. + Requires the user to have instructor access. + + Supports optional search and pagination parameters: + - search: Filter users by username, email, first_name, or last_name + - page: Page number (default: 1) + - page_size: Number of results per page (default: 20, max: 100) + + Returns JSON of the form { + "course_id": "some/course/id", + "enrollments": [ + { + "username": "student1", + "email": "student1@example.org", + "first_name": "Jane", + "last_name": "Doe", + } + ], + "count": 100, + "num_pages": 5, + "current_page": 1 + } + """ + permission_classes = (IsAuthenticated, permissions.InstructorPermission) + permission_name = permissions.VIEW_ENROLLMENTS + + @method_decorator(ensure_csrf_cookie) + def post(self, request, course_id): + """ + Handles POST request to list course enrollments. + + Args: + request (HttpRequest): The request object containing user data. + course_id (str): The ID of the course to list enrollments for. + + Returns: + Response: A Response object containing the list of enrollments or an error message. + + Raises: + Http404: If the course does not exist. + """ + course_key = CourseKey.from_string(course_id) + course = get_course_with_access( + request.user, 'instructor', course_key, depth=None + ) + + enrollment_serializer = EnrollmentListSerializer(data=request.data) + enrollment_serializer.is_valid(raise_exception=True) + search = enrollment_serializer.data.get('search', '').strip() + page = enrollment_serializer.data.get('page', 1) + page_size = enrollment_serializer.data.get('page_size', 20) + + # Get all active enrollments for the course + enrollments = CourseEnrollment.objects.filter( + course_id=course_key, + is_active=True + ).select_related('user').order_by('user__username') + + # Apply search filter + if search: + enrollments = enrollments.filter( + Q(user__username__icontains=search) | + Q(user__email__icontains=search) | + Q(user__first_name__icontains=search) | + Q(user__last_name__icontains=search) + ) + + # Calculate pagination + total_count = enrollments.count() + num_pages = (total_count + page_size - 1) // page_size if page_size > 0 else 1 + start_idx = (page - 1) * page_size + end_idx = start_idx + page_size + paginated_enrollments = enrollments[start_idx:end_idx] + + # Extract user data from enrollments + users = [enrollment.user for enrollment in paginated_enrollments] + serializer = UserSerializer(users, many=True) + + response_payload = { + 'course_id': str(course_key), + 'enrollments': serializer.data, + 'count': total_count, + 'num_pages': num_pages, + 'current_page': page, } return Response(response_payload, status=status.HTTP_200_OK) diff --git a/lms/djangoapps/instructor/views/api_urls.py b/lms/djangoapps/instructor/views/api_urls.py index b61120e58c13..45089749d4d6 100644 --- a/lms/djangoapps/instructor/views/api_urls.py +++ b/lms/djangoapps/instructor/views/api_urls.py @@ -72,6 +72,7 @@ path('students_update_enrollment', api.StudentsUpdateEnrollmentView.as_view(), name='students_update_enrollment'), path('register_and_enroll_students', api.RegisterAndEnrollStudents.as_view(), name='register_and_enroll_students'), path('list_course_role_members', api.ListCourseRoleMembersView.as_view(), name='list_course_role_members'), + path('list_course_enrollments', api.ListCourseEnrollmentsView.as_view(), name='list_course_enrollments'), path('modify_access', api.ModifyAccess.as_view(), name='modify_access'), path('bulk_beta_modify_access', api.BulkBetaModifyAccess.as_view(), name='bulk_beta_modify_access'), path('get_problem_responses', api.GetProblemResponses.as_view(), name='get_problem_responses'), diff --git a/lms/djangoapps/instructor/views/serializer.py b/lms/djangoapps/instructor/views/serializer.py index 9b83acdf6156..229a49dc9849 100644 --- a/lms/djangoapps/instructor/views/serializer.py +++ b/lms/djangoapps/instructor/views/serializer.py @@ -29,6 +29,9 @@ class RoleNameSerializer(serializers.Serializer): # pylint: disable=abstract-me """ rolename = serializers.CharField(help_text=_("Role name")) + search = serializers.CharField(required=False, allow_blank=True, help_text=_("Search term")) + page = serializers.IntegerField(required=False, min_value=1, help_text=_("Page number")) + page_size = serializers.IntegerField(required=False, min_value=1, max_value=100, help_text=_("Page size")) def validate_rolename(self, value): """ @@ -45,6 +48,15 @@ class Meta: fields = ['username', 'email', 'first_name', 'last_name'] +class EnrollmentListSerializer(serializers.Serializer): # pylint: disable=abstract-method + """ + Serializer for enrollment list request parameters. + """ + search = serializers.CharField(required=False, allow_blank=True, help_text=_("Search term")) + page = serializers.IntegerField(required=False, min_value=1, help_text=_("Page number")) + page_size = serializers.IntegerField(required=False, min_value=1, max_value=100, help_text=_("Page size")) + + class UniqueStudentIdentifierSerializer(serializers.Serializer): """ Serializer for identifying unique_student. From 6e08a92711e56ae5ec36c969a4825eff559bd29c Mon Sep 17 00:00:00 2001 From: Brian Buck Date: Wed, 11 Mar 2026 14:03:06 -0600 Subject: [PATCH 2/3] fix: Update test JSON structure. --- lms/djangoapps/instructor/tests/test_api.py | 10 ++++++++-- 1 file changed, 8 insertions(+), 2 deletions(-) diff --git a/lms/djangoapps/instructor/tests/test_api.py b/lms/djangoapps/instructor/tests/test_api.py index 4d9f6f83b9e8..898b64f3e9b3 100644 --- a/lms/djangoapps/instructor/tests/test_api.py +++ b/lms/djangoapps/instructor/tests/test_api.py @@ -2425,7 +2425,10 @@ def test_list_course_role_members_staff(self): 'first_name': self.other_staff.first_name, 'last_name': self.other_staff.last_name, } - ] + ], + 'count': 1, + 'num_pages': 1, + 'current_page': 1, } res_json = json.loads(response.content.decode('utf-8')) assert res_json == expected @@ -2440,7 +2443,10 @@ def test_list_course_role_members_beta(self): # check response content expected = { 'course_id': str(self.course.id), - 'beta': [] + 'beta': [], + 'count': 0, + 'num_pages': 0, + 'current_page': 1, } res_json = json.loads(response.content.decode('utf-8')) assert res_json == expected From 95675b0440b459c454ddafb76acdab13ed195d0d Mon Sep 17 00:00:00 2001 From: Brian Buck Date: Thu, 19 Mar 2026 14:20:27 -0600 Subject: [PATCH 3/3] feat: Migrate to v2 api --- .../instructor/tests/test_api_v2.py | 302 +++++++++++++++++- lms/djangoapps/instructor/views/api.py | 2 +- lms/djangoapps/instructor/views/api_urls.py | 10 + lms/djangoapps/instructor/views/api_v2.py | 205 ++++++++++++ .../instructor/views/serializers_v2.py | 55 ++++ 5 files changed, 572 insertions(+), 2 deletions(-) diff --git a/lms/djangoapps/instructor/tests/test_api_v2.py b/lms/djangoapps/instructor/tests/test_api_v2.py index 61797e1e9354..ae08900ba6a6 100644 --- a/lms/djangoapps/instructor/tests/test_api_v2.py +++ b/lms/djangoapps/instructor/tests/test_api_v2.py @@ -16,7 +16,7 @@ from rest_framework.test import APIClient, APITestCase from edx_when.api import set_dates_for_course, set_date_for_block -from common.djangoapps.student.roles import CourseDataResearcherRole, CourseInstructorRole +from common.djangoapps.student.roles import CourseBetaTesterRole, CourseDataResearcherRole, CourseInstructorRole from common.djangoapps.student.tests.factories import ( AdminFactory, CourseEnrollmentFactory, @@ -1734,3 +1734,303 @@ def test_extension_data_structure(self, mock_title_or_url, mock_get_units, mock_ self.assertIsInstance(extension['email'], str) self.assertIsInstance(extension['unit_title'], str) self.assertIsInstance(extension['unit_location'], str) + + +class CourseEnrollmentsViewTest(SharedModuleStoreTestCase): + """Tests for the CourseEnrollmentsView v2 GET endpoint.""" + + @classmethod + def setUpClass(cls): + super().setUpClass() + cls.course = CourseFactory.create() + + def setUp(self): + super().setUp() + self.client = APIClient() + self.instructor = InstructorFactory(course_key=self.course.id) + self.url = reverse( + 'instructor_api_v2:course_enrollments', + kwargs={'course_id': str(self.course.id)} + ) + + self.enrolled_users = [] + for i in range(30): + user = UserFactory( + username=f'student_{i}', + email=f'student{i}@example.com', + first_name=f'Student{i}', + last_name=f'Learner{i}' + ) + CourseEnrollmentFactory( + user=user, + course_id=self.course.id, + is_active=True + ) + self.enrolled_users.append(user) + + # Inactive enrollments should not appear + for i in range(5): + user = UserFactory( + username=f'inactive_{i}', + email=f'inactive{i}@example.com' + ) + CourseEnrollmentFactory( + user=user, + course_id=self.course.id, + is_active=False + ) + + def test_unauthenticated_returns_401(self): + response = self.client.get(self.url) + self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED) + + def test_student_returns_403(self): + student = UserFactory() + self.client.force_authenticate(user=student) + response = self.client.get(self.url) + self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN) + + def test_default_pagination(self): + self.client.force_authenticate(user=self.instructor) + response = self.client.get(self.url) + self.assertEqual(response.status_code, status.HTTP_200_OK) + data = response.data + self.assertEqual(data['course_id'], str(self.course.id)) + self.assertEqual(data['count'], 30) + self.assertEqual(data['num_pages'], 2) + self.assertEqual(data['current_page'], 1) + self.assertEqual(len(data['enrollments']), 20) + + def test_custom_pagination(self): + self.client.force_authenticate(user=self.instructor) + response = self.client.get(self.url, {'page': 1, 'page_size': 10}) + data = response.data + self.assertEqual(data['count'], 30) + self.assertEqual(data['num_pages'], 3) + self.assertEqual(len(data['enrollments']), 10) + + def test_second_page(self): + self.client.force_authenticate(user=self.instructor) + response = self.client.get(self.url, {'page': 2, 'page_size': 10}) + data = response.data + self.assertEqual(data['current_page'], 2) + self.assertEqual(len(data['enrollments']), 10) + + def test_last_page_partial(self): + self.client.force_authenticate(user=self.instructor) + response = self.client.get(self.url, {'page': 3, 'page_size': 10}) + data = response.data + self.assertEqual(data['current_page'], 3) + self.assertEqual(len(data['enrollments']), 10) + + def test_search_by_username(self): + self.client.force_authenticate(user=self.instructor) + response = self.client.get(self.url, {'search': 'student_2', 'page_size': 100}) + data = response.data + # Matches student_2, student_20..student_29 = 11 + self.assertEqual(data['count'], 11) + for user in data['enrollments']: + self.assertIn('student_2', user['username']) + + def test_search_by_email(self): + self.client.force_authenticate(user=self.instructor) + response = self.client.get(self.url, {'search': 'student7@example.com'}) + data = response.data + self.assertEqual(data['count'], 1) + self.assertEqual(data['enrollments'][0]['email'], 'student7@example.com') + + def test_search_case_insensitive(self): + self.client.force_authenticate(user=self.instructor) + response = self.client.get(self.url, {'search': 'STUDENT_5'}) + data = response.data + self.assertEqual(data['count'], 1) + self.assertEqual(data['enrollments'][0]['username'], 'student_5') + + def test_search_no_results(self): + self.client.force_authenticate(user=self.instructor) + response = self.client.get(self.url, {'search': 'nonexistent'}) + data = response.data + self.assertEqual(data['count'], 0) + self.assertEqual(len(data['enrollments']), 0) + + def test_excludes_inactive_enrollments(self): + self.client.force_authenticate(user=self.instructor) + response = self.client.get(self.url, {'search': 'inactive'}) + data = response.data + self.assertEqual(data['count'], 0) + + def test_invalid_page_size(self): + self.client.force_authenticate(user=self.instructor) + response = self.client.get(self.url, {'page_size': 0}) + self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) + + def test_invalid_page(self): + self.client.force_authenticate(user=self.instructor) + response = self.client.get(self.url, {'page': -1}) + self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) + + def test_ordered_by_username(self): + self.client.force_authenticate(user=self.instructor) + response = self.client.get(self.url, {'page_size': 5}) + data = response.data + usernames = [u['username'] for u in data['enrollments']] + self.assertEqual(usernames, sorted(usernames)) + + def test_staff_can_access(self): + staff = StaffFactory(course_key=self.course.id) + self.client.force_authenticate(user=staff) + response = self.client.get(self.url) + self.assertEqual(response.status_code, status.HTTP_200_OK) + + def test_includes_mode_field(self): + self.client.force_authenticate(user=self.instructor) + response = self.client.get(self.url, {'page_size': 1}) + enrollment = response.data['enrollments'][0] + self.assertIn('mode', enrollment) + + def test_includes_is_beta_tester_field(self): + self.client.force_authenticate(user=self.instructor) + response = self.client.get(self.url, {'page_size': 100}) + for enrollment in response.data['enrollments']: + self.assertIn('is_beta_tester', enrollment) + self.assertFalse(enrollment['is_beta_tester']) + + def test_beta_tester_flag_true(self): + beta_role = CourseBetaTesterRole(self.course.id) + target_user = self.enrolled_users[0] + beta_role.add_users(target_user) + + self.client.force_authenticate(user=self.instructor) + response = self.client.get(self.url, {'search': target_user.username}) + data = response.data + self.assertEqual(data['count'], 1) + self.assertTrue(data['enrollments'][0]['is_beta_tester']) + + +class RoleMembersViewTest(SharedModuleStoreTestCase): + """Tests for the RoleMembersView v2 GET endpoint.""" + + @classmethod + def setUpClass(cls): + super().setUpClass() + cls.course = CourseFactory.create() + + def setUp(self): + super().setUp() + self.client = APIClient() + self.instructor = InstructorFactory(course_key=self.course.id) + self.url = reverse( + 'instructor_api_v2:role_members', + kwargs={'course_id': str(self.course.id)} + ) + + self.beta_testers = [] + beta_role = CourseBetaTesterRole(self.course.id) + for i in range(25): + user = UserFactory( + username=f'beta_user_{i}', + email=f'beta{i}@example.com', + first_name=f'Beta{i}', + last_name=f'Tester{i}' + ) + beta_role.add_users(user) + self.beta_testers.append(user) + + def test_unauthenticated_returns_401(self): + response = self.client.get(self.url, {'rolename': 'beta'}) + self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED) + + def test_student_returns_403(self): + student = UserFactory() + self.client.force_authenticate(user=student) + response = self.client.get(self.url, {'rolename': 'beta'}) + self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN) + + def test_missing_rolename_returns_400(self): + self.client.force_authenticate(user=self.instructor) + response = self.client.get(self.url) + self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) + + def test_invalid_rolename_returns_400(self): + self.client.force_authenticate(user=self.instructor) + response = self.client.get(self.url, {'rolename': 'invalid_role'}) + self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) + + def test_list_beta_testers_default_pagination(self): + self.client.force_authenticate(user=self.instructor) + response = self.client.get(self.url, {'rolename': 'beta'}) + self.assertEqual(response.status_code, status.HTTP_200_OK) + data = response.data + self.assertEqual(data['course_id'], str(self.course.id)) + self.assertIn('beta', data) + self.assertEqual(data['count'], 25) + self.assertEqual(data['num_pages'], 2) + self.assertEqual(data['current_page'], 1) + self.assertEqual(len(data['beta']), 20) + + def test_list_beta_testers_custom_pagination(self): + self.client.force_authenticate(user=self.instructor) + response = self.client.get(self.url, {'rolename': 'beta', 'page': 1, 'page_size': 10}) + data = response.data + self.assertEqual(data['count'], 25) + self.assertEqual(data['num_pages'], 3) + self.assertEqual(len(data['beta']), 10) + + def test_list_beta_testers_last_page(self): + self.client.force_authenticate(user=self.instructor) + response = self.client.get(self.url, {'rolename': 'beta', 'page': 3, 'page_size': 10}) + data = response.data + self.assertEqual(data['current_page'], 3) + self.assertEqual(len(data['beta']), 5) + + def test_beyond_last_page_empty(self): + self.client.force_authenticate(user=self.instructor) + response = self.client.get(self.url, {'rolename': 'beta', 'page': 10, 'page_size': 10}) + data = response.data + self.assertEqual(len(data['beta']), 0) + + def test_search_by_username(self): + self.client.force_authenticate(user=self.instructor) + response = self.client.get(self.url, {'rolename': 'beta', 'search': 'beta_user_1', 'page_size': 100}) + data = response.data + # Matches beta_user_1, beta_user_10..beta_user_19 = 11 + self.assertEqual(data['count'], 11) + for user in data['beta']: + self.assertIn('beta_user_1', user['username']) + + def test_search_by_email(self): + self.client.force_authenticate(user=self.instructor) + response = self.client.get(self.url, {'rolename': 'beta', 'search': 'beta5@example.com'}) + data = response.data + self.assertEqual(data['count'], 1) + self.assertEqual(data['beta'][0]['email'], 'beta5@example.com') + + def test_search_case_insensitive(self): + self.client.force_authenticate(user=self.instructor) + response = self.client.get(self.url, {'rolename': 'beta', 'search': 'BETA_USER_3'}) + data = response.data + self.assertEqual(data['count'], 1) + self.assertEqual(data['beta'][0]['username'], 'beta_user_3') + + def test_search_no_results(self): + self.client.force_authenticate(user=self.instructor) + response = self.client.get(self.url, {'rolename': 'beta', 'search': 'nonexistent'}) + data = response.data + self.assertEqual(data['count'], 0) + self.assertEqual(len(data['beta']), 0) + + def test_list_staff_role(self): + self.client.force_authenticate(user=self.instructor) + response = self.client.get(self.url, {'rolename': 'staff'}) + self.assertEqual(response.status_code, status.HTTP_200_OK) + self.assertIn('staff', response.data) + + def test_invalid_page_size(self): + self.client.force_authenticate(user=self.instructor) + response = self.client.get(self.url, {'rolename': 'beta', 'page_size': 0}) + self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) + + def test_invalid_page(self): + self.client.force_authenticate(user=self.instructor) + response = self.client.get(self.url, {'rolename': 'beta', 'page': 0}) + self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) diff --git a/lms/djangoapps/instructor/views/api.py b/lms/djangoapps/instructor/views/api.py index b089e48fbc05..4eb7b80c7a09 100644 --- a/lms/djangoapps/instructor/views/api.py +++ b/lms/djangoapps/instructor/views/api.py @@ -1178,7 +1178,7 @@ def post(self, request, course_id): """ course_key = CourseKey.from_string(course_id) course = get_course_with_access( - request.user, 'instructor', course_key, depth=None + request.user, 'staff', course_key, depth=None ) enrollment_serializer = EnrollmentListSerializer(data=request.data) diff --git a/lms/djangoapps/instructor/views/api_urls.py b/lms/djangoapps/instructor/views/api_urls.py index 45089749d4d6..4abe5cbda475 100644 --- a/lms/djangoapps/instructor/views/api_urls.py +++ b/lms/djangoapps/instructor/views/api_urls.py @@ -66,6 +66,16 @@ api_v2.ORASummaryView.as_view(), name='ora_summary' ), + re_path( + rf'^courses/{COURSE_ID_PATTERN}/enrollments$', + api_v2.CourseEnrollmentsView.as_view(), + name='course_enrollments' + ), + re_path( + rf'^courses/{COURSE_ID_PATTERN}/role_members$', + api_v2.RoleMembersView.as_view(), + name='role_members' + ), ] urlpatterns = [ diff --git a/lms/djangoapps/instructor/views/api_v2.py b/lms/djangoapps/instructor/views/api_v2.py index dd399dcb064a..16f14cf95f73 100644 --- a/lms/djangoapps/instructor/views/api_v2.py +++ b/lms/djangoapps/instructor/views/api_v2.py @@ -34,12 +34,17 @@ from xmodule.modulestore.django import modulestore from xmodule.modulestore.exceptions import ItemNotFoundError +from django.db.models import Q + +from common.djangoapps.student.models import CourseEnrollment +from common.djangoapps.student.roles import CourseBetaTesterRole from common.djangoapps.util.json_request import JsonResponseBadRequest from openedx.core.djangoapps.course_groups.cohorts import is_course_cohorted from openedx.core.djangoapps.site_configuration import helpers as configuration_helpers from lms.djangoapps.courseware.tabs import get_course_tab_list from lms.djangoapps.instructor import permissions +from lms.djangoapps.instructor.access import list_with_level from lms.djangoapps.instructor.views.api import _display_unit, get_student_from_identifier from lms.djangoapps.instructor.views.instructor_task_helpers import extract_task_features from lms.djangoapps.instructor_task import api as task_api @@ -56,7 +61,10 @@ InstructorTaskListSerializer, CourseInformationSerializerV2, BlockDueDateSerializerV2, + EnrollmentListRequestSerializerV2, + RoleMembersRequestSerializerV2, UnitExtensionSerializer, + UserSerializerV2, ORASerializer, ORASummarySerializer, ) @@ -1097,3 +1105,200 @@ def get(self, request, *args, **kwargs): serializer = self.get_serializer(items) return Response(serializer.data) + + +class CourseEnrollmentsView(DeveloperErrorViewMixin, APIView): + """ + **Use Cases** + + List all active enrollments for a course with optional search and pagination. + + **Query Parameters** + + - search: Filter by username, email, first name, or last name + - page: Page number (default: 1) + - page_size: Number of results per page (default: 20, max: 100) + """ + permission_classes = (IsAuthenticated, permissions.InstructorPermission) + permission_name = permissions.VIEW_ENROLLMENTS + + @apidocs.schema( + parameters=[ + apidocs.string_parameter( + 'course_id', + apidocs.ParameterLocation.PATH, + description="Course key for the course.", + ), + apidocs.string_parameter( + 'search', + apidocs.ParameterLocation.QUERY, + description="Filter by username, email, first name, or last name.", + ), + apidocs.string_parameter( + 'page', + apidocs.ParameterLocation.QUERY, + description="Page number (default: 1).", + ), + apidocs.string_parameter( + 'page_size', + apidocs.ParameterLocation.QUERY, + description="Number of results per page (default: 20, max: 100).", + ), + ], + responses={ + 200: "Paginated list of active enrollments.", + 401: "The requesting user is not authenticated.", + 403: "The requesting user lacks access to the course.", + }, + ) + def get(self, request, course_id): + """ + List active enrollments for the given course. + """ + course_key = CourseKey.from_string(course_id) + + serializer = EnrollmentListRequestSerializerV2(data=request.query_params) + if not serializer.is_valid(): + return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST) + + search = serializer.validated_data['search'].strip() + page = serializer.validated_data['page'] + page_size = serializer.validated_data['page_size'] + + enrollments = CourseEnrollment.objects.filter( + course_id=course_key, + is_active=True + ).select_related('user').order_by('user__username') + + if search: + enrollments = enrollments.filter( + Q(user__username__icontains=search) + | Q(user__email__icontains=search) + | Q(user__first_name__icontains=search) + | Q(user__last_name__icontains=search) + ) + + total_count = enrollments.count() + num_pages = (total_count + page_size - 1) // page_size if page_size > 0 else 1 + start_idx = (page - 1) * page_size + end_idx = start_idx + page_size + paginated_enrollments = enrollments[start_idx:end_idx] + + # Batch-fetch beta tester user IDs for the course (single query) + beta_tester_ids = set( + CourseBetaTesterRole(course_key).users_with_role().values_list('id', flat=True) + ) + + enrollment_data = [] + for enrollment in paginated_enrollments: + user = enrollment.user + enrollment_data.append({ + 'username': user.username, + 'email': user.email, + 'first_name': user.first_name, + 'last_name': user.last_name, + 'mode': enrollment.mode, + 'is_beta_tester': user.id in beta_tester_ids, + }) + + return Response({ + 'course_id': str(course_key), + 'enrollments': enrollment_data, + 'count': total_count, + 'num_pages': num_pages, + 'current_page': page, + }) + + +class RoleMembersView(DeveloperErrorViewMixin, APIView): + """ + **Use Cases** + + List users with a specific role for a course with optional search and pagination. + + **Query Parameters** + + - rolename: Required. One of 'instructor', 'staff', 'beta', 'ccx_coach' + - search: Filter by username, email, first name, or last name + - page: Page number (default: 1) + - page_size: Number of results per page (default: 20, max: 100) + """ + permission_classes = (IsAuthenticated, permissions.InstructorPermission) + permission_name = permissions.EDIT_COURSE_ACCESS + + @apidocs.schema( + parameters=[ + apidocs.string_parameter( + 'course_id', + apidocs.ParameterLocation.PATH, + description="Course key for the course.", + ), + apidocs.string_parameter( + 'rolename', + apidocs.ParameterLocation.QUERY, + description="Role name: one of 'instructor', 'staff', 'beta', 'ccx_coach'.", + ), + apidocs.string_parameter( + 'search', + apidocs.ParameterLocation.QUERY, + description="Filter by username, email, first name, or last name.", + ), + apidocs.string_parameter( + 'page', + apidocs.ParameterLocation.QUERY, + description="Page number (default: 1).", + ), + apidocs.string_parameter( + 'page_size', + apidocs.ParameterLocation.QUERY, + description="Number of results per page (default: 20, max: 100).", + ), + ], + responses={ + 200: "Paginated list of users with the specified role.", + 401: "The requesting user is not authenticated.", + 403: "The requesting user lacks access to the course.", + }, + ) + def get(self, request, course_id): + """ + List users with the specified role for the given course. + """ + course_key = CourseKey.from_string(course_id) + + serializer = RoleMembersRequestSerializerV2(data=request.query_params) + if not serializer.is_valid(): + return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST) + + rolename = serializer.validated_data['rolename'] + search = serializer.validated_data['search'].strip() + page = serializer.validated_data['page'] + page_size = serializer.validated_data['page_size'] + + users = list_with_level(course_key, rolename) + + if search: + search_lower = search.lower() + users = [ + user for user in users + if search_lower in user.username.lower() + or search_lower in user.email.lower() + or search_lower in (user.first_name or '').lower() + or search_lower in (user.last_name or '').lower() + ] + + total_count = len(users) + num_pages = (total_count + page_size - 1) // page_size if page_size > 0 else 1 + start_idx = (page - 1) * page_size + end_idx = start_idx + page_size + paginated_users = users[start_idx:end_idx] + + user_serializer = UserSerializerV2(paginated_users, many=True) + + return Response({ + 'course_id': str(course_key), + rolename: user_serializer.data, + 'count': total_count, + 'num_pages': num_pages, + 'current_page': page, + }) diff --git a/lms/djangoapps/instructor/views/serializers_v2.py b/lms/djangoapps/instructor/views/serializers_v2.py index 04c97450b2f5..4b7abf63bb1c 100644 --- a/lms/djangoapps/instructor/views/serializers_v2.py +++ b/lms/djangoapps/instructor/views/serializers_v2.py @@ -13,6 +13,8 @@ from edx_when.api import is_enabled_for_course from rest_framework import serializers +from django.contrib.auth import get_user_model + from common.djangoapps.course_modes.models import CourseMode from common.djangoapps.student.models import CourseEnrollment from common.djangoapps.student.roles import ( @@ -33,8 +35,12 @@ from openedx.core.djangoapps.django_comment_common.models import FORUM_ROLE_ADMINISTRATOR from xmodule.modulestore.django import modulestore +from lms.djangoapps.instructor.access import ROLES + from .tools import get_student_from_identifier, parse_datetime, DashboardError +User = get_user_model() + log = logging.getLogger(__name__) @@ -487,3 +493,52 @@ class ORASummarySerializer(serializers.Serializer): waiting = serializers.IntegerField() staff = serializers.IntegerField() final_grade_received = serializers.IntegerField() + + +class PaginationRequestSerializerV2(serializers.Serializer): + """ + Base serializer for paginated list requests with optional search. + """ + search = serializers.CharField( + required=False, + allow_blank=True, + default='', + help_text="Filter by username, email, first name, or last name" + ) + page = serializers.IntegerField( + required=False, + min_value=1, + default=1, + help_text="Page number (default: 1)" + ) + page_size = serializers.IntegerField( + required=False, + min_value=1, + max_value=100, + default=20, + help_text="Number of results per page (default: 20, max: 100)" + ) + + +class UserSerializerV2(serializers.ModelSerializer): + """Serializer for user data in enrollment and role member lists.""" + class Meta: + model = User + fields = ['username', 'email', 'first_name', 'last_name'] + + +class EnrollmentListRequestSerializerV2(PaginationRequestSerializerV2): + """Request serializer for the enrollment list endpoint.""" + + +class RoleMembersRequestSerializerV2(PaginationRequestSerializerV2): + """Request serializer for the role members list endpoint.""" + rolename = serializers.CharField( + required=True, + help_text="Role name: one of 'instructor', 'staff', 'beta', 'ccx_coach'" + ) + + def validate_rolename(self, value): + if value not in ROLES: + raise serializers.ValidationError(_("Invalid role name.")) + return value