diff --git a/lms/djangoapps/instructor/docs/references/instructor-v2-special-exams-api-spec.yaml b/lms/djangoapps/instructor/docs/references/instructor-v2-special-exams-api-spec.yaml index a778b04afd2b..5620c1eb75e8 100644 --- a/lms/djangoapps/instructor/docs/references/instructor-v2-special-exams-api-spec.yaml +++ b/lms/djangoapps/instructor/docs/references/instructor-v2-special-exams-api-spec.yaml @@ -488,6 +488,11 @@ definitions: type: string exam_id: type: integer + exam_name: + type: string + exam_type: + type: string + enum: [timed, proctored, practice] status: type: string start_time: @@ -501,6 +506,8 @@ definitions: allowed_time_limit_mins: type: integer x-nullable: true + ready_to_resume: + type: boolean ProctoringSettings: type: object diff --git a/lms/djangoapps/instructor/tests/test_api.py b/lms/djangoapps/instructor/tests/test_api.py index 869ea0235a1f..c23793b82099 100644 --- a/lms/djangoapps/instructor/tests/test_api.py +++ b/lms/djangoapps/instructor/tests/test_api.py @@ -4581,6 +4581,24 @@ def test_change_due_date_v2_success(self): assert get_extended_due(self.course, self.homework, self.user1) == due_date + def test_change_due_date_v2_without_reason(self): + """Test that reason is optional — both omitted and blank are accepted.""" + url = reverse('instructor_api_v2:change_due_date', kwargs={'course_id': str(self.course.id)}) + base_payload = { + 'email_or_username': self.user1.username, + 'block_id': str(self.homework.location), + 'due_datetime': '12/30/2013 00:00', + } + # Omitted reason + response = self.client.post(url, json.dumps(base_payload), content_type='application/json') + assert response.status_code == 200, response.content + + # Blank reason + response = self.client.post( + url, json.dumps({**base_payload, 'reason': ''}), content_type='application/json' + ) + assert response.status_code == 200, response.content + def test_change_due_date_v2_with_email(self): """Test due date change using email instead of username""" url = reverse('instructor_api_v2:change_due_date', kwargs={'course_id': str(self.course.id)}) diff --git a/lms/djangoapps/instructor/tests/test_api_v2.py b/lms/djangoapps/instructor/tests/test_api_v2.py index f2565b6bad5f..3a0d1be7f661 100644 --- a/lms/djangoapps/instructor/tests/test_api_v2.py +++ b/lms/djangoapps/instructor/tests/test_api_v2.py @@ -3,7 +3,7 @@ """ import json from datetime import datetime, timedelta -from unittest.mock import Mock, patch +from unittest.mock import MagicMock, Mock, patch from urllib.parse import urlencode from uuid import uuid4 @@ -12,6 +12,7 @@ from django.contrib.auth import get_user_model from django.http import Http404 from django.test import SimpleTestCase, override_settings +from django.test.client import Client as DjangoClient from django.urls import NoReverseMatch, reverse from edx_when.api import set_date_for_block, set_dates_for_course from opaque_keys import InvalidKeyError @@ -39,6 +40,7 @@ from lms.djangoapps.certificates.models import CertificateAllowlist, CertificateGenerationHistory from lms.djangoapps.certificates.tests.factories import GeneratedCertificateFactory from lms.djangoapps.courseware.models import StudentModule +from lms.djangoapps.courseware.tests.helpers import MasqueradeMixin from lms.djangoapps.instructor.access import ROLE_DISPLAY_NAMES from lms.djangoapps.instructor.permissions import InstructorPermission from lms.djangoapps.instructor.views.serializers_v2 import CourseInformationSerializerV2 @@ -50,7 +52,7 @@ @ddt.ddt -class CourseMetadataViewTest(SharedModuleStoreTestCase): +class CourseMetadataViewTest(SharedModuleStoreTestCase, MasqueradeMixin): """ Tests for the CourseMetadataView API endpoint. """ @@ -177,6 +179,9 @@ def test_get_course_metadata_as_instructor(self): assert 'studio_grading_url' in data assert 'admin_console_url' in data + # Verify current user's username is returned + assert data['username'] == self.instructor.username + assert data['studio_grading_url'] == f'http://localhost:2001/authoring/course/{self.course.id}/settings/grading' assert data['admin_console_url'] == 'http://localhost:2025/admin-console/authz' @@ -214,12 +219,13 @@ def test_get_course_metadata_as_staff(self): self.client.force_authenticate(user=self.staff) response = self.client.get(self._get_url()) - self.assertEqual(response.status_code, status.HTTP_200_OK) # noqa: PT009 + assert response.status_code == status.HTTP_200_OK data = response.data - self.assertEqual(data['course_id'], str(self.course_key)) # noqa: PT009 - self.assertIn('permissions', data) # noqa: PT009 + assert data['course_id'] == str(self.course_key) + assert 'permissions' in data # Staff should have staff permission - self.assertTrue(data['permissions']['staff']) # noqa: PT009 + assert data['permissions']['staff'] is True + assert data['username'] == self.staff.username def test_get_course_metadata_unauthorized(self): """ @@ -625,6 +631,31 @@ def test_pacing_self_for_self_paced_course(self): self.assertEqual(response.status_code, status.HTTP_200_OK) # noqa: PT009 self.assertEqual(response.data['pacing'], 'self') # noqa: PT009 + def test_masquerade_as_student_role_returns_403(self): + """ + Test that the endpoint returns 403 when a staff user masquerades as a student role. + """ + # Use Django's test Client for masquerade (MasqueradeMixin is incompatible with DRF APIClient) + original_client = self.client + self.client = DjangoClient() + self.client.login(username=self.staff.username, password='Password1234') + self.update_masquerade(course=self.course, role='student') + response = self.client.get(self._get_url()) + self.client = original_client + assert response.status_code == status.HTTP_403_FORBIDDEN + + def test_masquerade_as_specific_student_returns_403(self): + """ + Test that the endpoint returns 403 when a staff user masquerades as a specific student. + """ + original_client = self.client + self.client = DjangoClient() + self.client.login(username=self.staff.username, password='Password1234') + self.update_masquerade(course=self.course, username=self.student.username) + response = self.client.get(self._get_url()) + self.client = original_client + assert response.status_code == status.HTTP_403_FORBIDDEN + class BuildTabUrlTest(SimpleTestCase): """ @@ -2180,6 +2211,135 @@ def test_granted_exceptions_without_certificates(self): assert results['student2']['exception_notes'] == 'Special case' +class RegenerateCertificatesViewTest(SharedModuleStoreTestCase): + """ + Tests for the RegenerateCertificatesView API endpoint. + """ + + @classmethod + def setUpClass(cls): + super().setUpClass() + cls.course = CourseFactory.create( + org='edX', + number='TestX', + run='Test_Course', + display_name='Test Course', + ) + cls.course_key = cls.course.id + + def setUp(self): + super().setUp() + self.client = APIClient() + self.instructor = InstructorFactory.create(course_key=self.course_key) + self.student = UserFactory.create(username='student1', email='student1@example.com') + + # Enroll student + CourseEnrollmentFactory.create( + user=self.student, + course_id=self.course_key, + mode='verified', + is_active=True + ) + + def _get_url(self, course_id=None): + """Helper to get the API URL.""" + if course_id is None: + course_id = str(self.course_key) + return reverse('instructor_api_v2:regenerate_certificates', kwargs={'course_id': course_id}) + + @patch('lms.djangoapps.instructor.views.api_v2.task_api.generate_certificates_for_students') + def test_allowlisted_not_generated_passes_correct_student_set(self, mock_generate_certs): + """ + Test that student_set='allowlisted_not_generated' is passed correctly to the task layer. + + This test prevents future drift between the API layer and task layer if either + is renamed independently. + """ + # Mock the task API to return a fake InstructorTask + mock_task = MagicMock() + mock_task.task_id = 'test-task-id-123' + mock_generate_certs.return_value = mock_task + + # Authenticate and make the request + self.client.force_authenticate(user=self.instructor) + response = self.client.post( + self._get_url(), + data={'student_set': 'allowlisted_not_generated'}, + format='json' + ) + + # Assert the response is successful + assert response.status_code == status.HTTP_200_OK + assert response.data['task_id'] == 'test-task-id-123' + + # Assert the task API was called with the correct parameters + # Expected call signature: generate_certificates_for_students(request, course_key, student_set=...) + mock_generate_certs.assert_called_once() + call_args = mock_generate_certs.call_args + _, course_key_arg = call_args.args[:2] # Unpack request and course_key positional args + assert course_key_arg == self.course_key + assert call_args.kwargs['student_set'] == 'allowlisted_not_generated' + + @patch('lms.djangoapps.instructor.views.api_v2.task_api.generate_certificates_for_students') + def test_allowlisted_translates_to_all_allowlisted(self, mock_generate_certs): + """ + Test that student_set='allowlisted' is translated to 'all_allowlisted' for the task layer. + + This preserves the legacy translation from the pre-allowlist "whitelist" naming era. + """ + # Mock the task API to return a fake InstructorTask + mock_task = MagicMock() + mock_task.task_id = 'test-task-id-456' + mock_generate_certs.return_value = mock_task + + # Authenticate and make the request + self.client.force_authenticate(user=self.instructor) + response = self.client.post( + self._get_url(), + data={'student_set': 'allowlisted'}, + format='json' + ) + + # Assert the response is successful + assert response.status_code == status.HTTP_200_OK + + # Assert the task API was called with the translated value + mock_generate_certs.assert_called_once() + call_kwargs = mock_generate_certs.call_args.kwargs + assert call_kwargs['student_set'] == 'all_allowlisted' + + @patch('lms.djangoapps.instructor.views.api_v2.task_api.generate_certificates_for_students') + def test_all_students_omits_student_set_kwarg(self, mock_generate_certs): + """ + Test that student_set='all' calls the task layer without a student_set kwarg. + + This ensures the default behavior (generate for all enrolled students) is preserved. + """ + # Mock the task API to return a fake InstructorTask + mock_task = MagicMock() + mock_task.task_id = 'test-task-id-789' + mock_generate_certs.return_value = mock_task + + # Authenticate and make the request with student_set='all' + self.client.force_authenticate(user=self.instructor) + response = self.client.post( + self._get_url(), + data={'student_set': 'all'}, + format='json' + ) + + # Assert the response is successful + assert response.status_code == status.HTTP_200_OK + + # Assert the task API was called without student_set kwarg + # Expected call signature: generate_certificates_for_students(request, course_key) + mock_generate_certs.assert_called_once() + call_args = mock_generate_certs.call_args + _, course_key_arg = call_args.args[:2] # Unpack request and course_key positional args + assert course_key_arg == self.course_key + assert 'student_set' not in call_args.kwargs + + @ddt.ddt class CertificateGenerationHistoryViewTest(SharedModuleStoreTestCase): """ diff --git a/lms/djangoapps/instructor/tests/views/test_special_exams_api_v2.py b/lms/djangoapps/instructor/tests/views/test_special_exams_api_v2.py index 73ad265d928d..4304b004ccc6 100644 --- a/lms/djangoapps/instructor/tests/views/test_special_exams_api_v2.py +++ b/lms/djangoapps/instructor/tests/views/test_special_exams_api_v2.py @@ -7,11 +7,14 @@ from django.conf import settings from django.test.utils import override_settings from django.urls import reverse +from django.utils import timezone from edx_proctoring.api import ( add_allowance_for_user, create_exam, create_exam_attempt, + get_allowances_for_course, ) +from edx_proctoring.models import ProctoredExamStudentAttempt from rest_framework import status from rest_framework.test import APIClient @@ -84,18 +87,21 @@ def test_list_exams(self): assert timed['is_practice_exam'] is False assert timed['is_active'] is True assert timed['hide_after_due'] is False + assert timed['exam_type'] == 'timed' proctored = exams_by_name['Proctored Exam'] assert proctored['id'] == self.proctored_exam_id assert proctored['time_limit_mins'] == 90 assert proctored['is_proctored'] is True assert proctored['is_practice_exam'] is False + assert proctored['exam_type'] == 'proctored' practice = exams_by_name['Practice Exam'] assert practice['id'] == self.practice_exam_id assert practice['time_limit_mins'] == 30 assert practice['is_proctored'] is True assert practice['is_practice_exam'] is True + assert practice['exam_type'] == 'practice' def test_unauthenticated(self): self.client.force_authenticate(user=None) @@ -215,6 +221,7 @@ def test_reset_no_attempts(self): @override_settings(**PROCTORING_SETTINGS) @patch.dict(settings.FEATURES, {'ENABLE_SPECIAL_EXAMS': True}) +@ddt.ddt class SpecialExamAttemptsViewTest(ModuleStoreTestCase): """Tests for GET /api/instructor/v2/courses/{course_key}/special_exams/{exam_id}/attempts""" @@ -226,46 +233,66 @@ def setUp(self): self.student = UserFactory(username='student1', email='student1@example.com') self.client.force_authenticate(user=self.instructor) self.course_id = str(self.course.id) - self.exam_id = create_exam( + + def _create_exam(self, is_proctored=False, is_practice_exam=False, content_suffix='exam1'): + return create_exam( course_id=self.course_id, - content_id='block-v1:test+test+test+type@sequential+block@exam1', - exam_name='Midterm Exam', + content_id=f'block-v1:test+test+test+type@sequential+block@{content_suffix}', + exam_name='Test Exam', time_limit_mins=60, - is_proctored=False, + is_proctored=is_proctored, + is_practice_exam=is_practice_exam, ) - def _url(self, exam_id=None): + def _url(self, exam_id): return reverse('instructor_api_v2:special_exam_attempts', kwargs={ 'course_id': self.course_id, - 'exam_id': exam_id or self.exam_id, + 'exam_id': exam_id, }) - def test_list_attempts(self): - create_exam_attempt(self.exam_id, self.student.id) - response = self.client.get(self._url()) + @ddt.data( + (False, False, 'timed'), + (True, False, 'proctored'), + (True, True, 'practice'), + ) + @ddt.unpack + def test_attempt_exam_type(self, is_proctored, is_practice_exam, expected_type): + exam_id = self._create_exam(is_proctored=is_proctored, is_practice_exam=is_practice_exam) + create_exam_attempt(exam_id, self.student.id) + response = self.client.get(self._url(exam_id)) assert response.status_code == status.HTTP_200_OK data = response.json() assert data['count'] == 1 - assert data['results'][0]['exam_id'] == self.exam_id + assert data['results'][0]['exam_id'] == exam_id + assert data['results'][0]['exam_name'] == 'Test Exam' + assert data['results'][0]['exam_type'] == expected_type + assert data['results'][0]['ready_to_resume'] is False assert data['results'][0]['user']['username'] == 'student1' def test_list_attempts_filters_by_exam(self): """Only attempts for the requested exam_id are returned.""" - other_exam_id = create_exam( - course_id=self.course_id, - content_id='block-v1:test+test+test+type@sequential+block@exam2', - exam_name='Final Exam', - time_limit_mins=120, - is_proctored=False, - ) - create_exam_attempt(self.exam_id, self.student.id) + exam_id = self._create_exam(content_suffix='exam1') + other_exam_id = self._create_exam(content_suffix='exam2') + create_exam_attempt(exam_id, self.student.id) other_student = UserFactory(username='student2') create_exam_attempt(other_exam_id, other_student.id) - response = self.client.get(self._url()) + response = self.client.get(self._url(exam_id)) data = response.json() assert data['count'] == 1 - assert data['results'][0]['exam_id'] == self.exam_id + assert data['results'][0]['exam_id'] == exam_id + + def test_ready_to_resume_true(self): + """Verify ready_to_resume reflects the actual attempt state.""" + exam_id = self._create_exam() + attempt_id = create_exam_attempt(exam_id, self.student.id) + attempt = ProctoredExamStudentAttempt.objects.get(id=attempt_id) + attempt.ready_to_resume = True + attempt.save() + + response = self.client.get(self._url(exam_id)) + assert response.status_code == status.HTTP_200_OK + assert response.json()['results'][0]['ready_to_resume'] is True @override_settings(**PROCTORING_SETTINGS) @@ -416,6 +443,33 @@ def test_update_allowance(self): assert response.status_code == status.HTTP_200_OK assert response.json()['results'][0]['success'] is True + def test_grant_allowance_replaces_different_key(self): + """Granting an allowance with a different key replaces the existing one (one per user+exam).""" + self.client.post( + self._url(), + data={ + 'user_ids': [self.student.username], + 'allowance_type': 'additional_time_granted', + 'value': '30', + }, + format='json', + ) + response = self.client.post( + self._url(), + data={ + 'user_ids': [self.student.username], + 'allowance_type': 'review_policy_exception', + 'value': 'special review', + }, + format='json', + ) + assert response.status_code == status.HTTP_200_OK + assert response.json()['results'][0]['success'] is True + allowances = get_allowances_for_course(self.course_id) + user_allowances = [a for a in allowances if a['user']['username'] == self.student.username] + assert len(user_allowances) == 1 + assert user_allowances[0]['key'] == 'review_policy_exception' + def test_delete_allowance(self): add_allowance_for_user(self.exam_id, self.student.username, 'additional_time_granted', '30') response = self.client.delete( @@ -459,6 +513,7 @@ def test_delete_allowance_missing_fields(self): @override_settings(**PROCTORING_SETTINGS) @patch.dict(settings.FEATURES, {'ENABLE_SPECIAL_EXAMS': True}) +@ddt.ddt class CourseAllowancesViewTest(ModuleStoreTestCase): """Tests for GET /api/instructor/v2/courses/{course_key}/special_exams/allowances""" @@ -535,6 +590,26 @@ def test_bulk_create_allowances(self): assert len(data['results']) == 4 assert all(r['success'] is True for r in data['results']) + def test_bulk_create_allowances_replaces_different_key(self): + """Bulk-creating an allowance with a different key replaces the existing one.""" + add_allowance_for_user(self.exam_id, self.student.username, 'additional_time_granted', '30') + response = self.client.post( + self._url(), + data={ + 'exam_ids': [self.exam_id], + 'user_ids': [self.student.username], + 'allowance_type': 'review_policy_exception', + 'value': 'special review', + }, + format='json', + ) + assert response.status_code == status.HTTP_200_OK + assert response.json()['results'][0]['success'] is True + allowances = get_allowances_for_course(self.course_id) + user_allowances = [a for a in allowances if a['user']['username'] == self.student.username] + assert len(user_allowances) == 1 + assert user_allowances[0]['key'] == 'review_policy_exception' + def test_bulk_create_allowances_missing_fields(self): response = self.client.post( self._url(), @@ -543,9 +618,53 @@ def test_bulk_create_allowances_missing_fields(self): ) assert response.status_code == status.HTTP_400_BAD_REQUEST + @ddt.data( + 'username', + 'user.username', + 'email', + 'user.email', + 'exam_name', + 'proctored_exam.exam_name', + 'allowance_type', + 'key', + 'value', + ) + def test_sort_allowances(self, ordering): + """Verify all ordering fields are accepted and reverse correctly.""" + student2 = UserFactory(username='alice', email='alice@example.com') + exam_id_2 = create_exam( + course_id=self.course_id, + content_id='block-v1:test+test+test+type@sequential+block@exam2', + exam_name='AAA Exam', + time_limit_mins=30, + is_proctored=False, + ) + add_allowance_for_user(self.exam_id, self.student.username, 'additional_time_granted', '30') + add_allowance_for_user(exam_id_2, student2.username, 'review_policy_exception', '60') + asc_response = self.client.get(self._url(), {'ordering': ordering}) + desc_response = self.client.get(self._url(), {'ordering': f'-{ordering}'}) + assert asc_response.status_code == status.HTTP_200_OK + asc_results = asc_response.json()['results'] + desc_results = desc_response.json()['results'] + assert len(asc_results) == 2 + # All allowance fields differ between the two records, so order must reverse + assert asc_results[0] == desc_results[1] + assert asc_results[1] == desc_results[0] + + def test_sort_allowances_descending(self): + student2 = UserFactory(username='alice', email='alice@example.com') + add_allowance_for_user(self.exam_id, self.student.username, 'additional_time_granted', '30') + add_allowance_for_user(self.exam_id, student2.username, 'additional_time_granted', '60') + response = self.client.get(self._url(), {'ordering': '-username'}) + assert response.status_code == status.HTTP_200_OK + results = response.json()['results'] + assert results[0]['user']['username'] == 'student1' + assert results[1]['user']['username'] == 'alice' + @override_settings(**PROCTORING_SETTINGS) @patch.dict(settings.FEATURES, {'ENABLE_SPECIAL_EXAMS': True}) +@ddt.ddt class CourseExamAttemptsViewTest(ModuleStoreTestCase): """Tests for GET /api/instructor/v2/courses/{course_key}/special_exams/attempts""" @@ -577,6 +696,8 @@ def test_list_all_attempts(self): data = response.json() assert data['count'] == 1 assert data['results'][0]['exam_id'] == self.exam_id + assert data['results'][0]['exam_name'] == 'Midterm Exam' + assert data['results'][0]['ready_to_resume'] is False def test_search_attempts_by_username(self): create_exam_attempt(self.exam_id, self.student.id) @@ -589,3 +710,70 @@ def test_search_attempts_no_match(self): response = self.client.get(self._url(), {'search': 'nonexistent'}) assert response.status_code == status.HTTP_200_OK assert response.json()['count'] == 0 + + @ddt.data( + 'username', + 'user.username', + 'email', + 'user.email', + 'exam_name', + 'proctored_exam.exam_name', + 'time_limit', + 'proctored_exam.time_limit_mins', + 'type', + 'started_at', + 'start_time', + 'completed_at', + 'end_time', + 'status', + ) + def test_sort_attempts(self, ordering): + """Verify all ordering fields produce reversed results for asc vs desc.""" + student2 = UserFactory(username='student2', email='student2@example.com') + exam_id_2 = create_exam( + course_id=self.course_id, + content_id='block-v1:test+test+test+type@sequential+block@exam2', + exam_name='Final Exam', + time_limit_mins=120, + is_proctored=True, + ) + attempt_id_1 = create_exam_attempt(self.exam_id, self.student.id) + create_exam_attempt(exam_id_2, student2.id) + + # Give attempt 1 a distinct completed_at and status so all fields differ + attempt = ProctoredExamStudentAttempt.objects.get(id=attempt_id_1) + attempt.started_at = timezone.now() - timezone.timedelta(hours=1) + attempt.completed_at = timezone.now() + attempt.status = 'submitted' + attempt.save() + + asc_response = self.client.get(self._url(), {'ordering': ordering}) + desc_response = self.client.get(self._url(), {'ordering': f'-{ordering}'}) + assert asc_response.status_code == status.HTTP_200_OK + assert desc_response.status_code == status.HTTP_200_OK + asc_results = asc_response.json()['results'] + desc_results = desc_response.json()['results'] + assert len(asc_results) == 2 + assert asc_results[0] == desc_results[1] + assert asc_results[1] == desc_results[0] + + def test_sort_attempts_descending(self): + student2 = UserFactory(username='student2', email='student2@example.com') + create_exam_attempt(self.exam_id, self.student.id) + create_exam_attempt(self.exam_id, student2.id) + response = self.client.get(self._url(), {'ordering': '-username'}) + assert response.status_code == status.HTTP_200_OK + results = response.json()['results'] + assert results[0]['user']['username'] == 'student2' + assert results[1]['user']['username'] == 'student1' + + def test_ready_to_resume_true(self): + """Verify ready_to_resume reflects the actual attempt state.""" + attempt_id = create_exam_attempt(self.exam_id, self.student.id) + attempt = ProctoredExamStudentAttempt.objects.get(id=attempt_id) + attempt.ready_to_resume = True + attempt.save() + + response = self.client.get(self._url()) + assert response.status_code == status.HTTP_200_OK + assert response.json()['results'][0]['ready_to_resume'] is True diff --git a/lms/djangoapps/instructor/views/api_v2.py b/lms/djangoapps/instructor/views/api_v2.py index 1b699520ad19..00177e0c321a 100644 --- a/lms/djangoapps/instructor/views/api_v2.py +++ b/lms/djangoapps/instructor/views/api_v2.py @@ -43,13 +43,14 @@ ProctoredBaseException, ProctoredExamNotFoundException, ) +from edx_proctoring.models import ProctoredExamStudentAllowance from edx_rest_framework_extensions.paginators import DefaultPagination from edx_when import api as edx_when_api from opaque_keys import InvalidKeyError from opaque_keys.edx.keys import CourseKey, UsageKey from pytz import UTC from rest_framework import status -from rest_framework.exceptions import NotFound +from rest_framework.exceptions import NotFound, PermissionDenied from rest_framework.generics import GenericAPIView, ListAPIView from rest_framework.permissions import IsAuthenticated from rest_framework.response import Response @@ -83,6 +84,7 @@ from lms.djangoapps.course_home_api.toggles import course_home_mfe_progress_tab_is_active from lms.djangoapps.courseware.access import has_access from lms.djangoapps.courseware.courses import get_course_with_access +from lms.djangoapps.courseware.masquerade import get_masquerade_role, setup_masquerade from lms.djangoapps.courseware.models import StudentModule from lms.djangoapps.courseware.tabs import get_course_tab_list from lms.djangoapps.instructor import enrollment, permissions @@ -161,6 +163,7 @@ TaskStatusSerializer, ToggleCertificateGenerationSerializer, UnitExtensionSerializer, + derive_exam_type, ) from .tools import find_unit, get_units_with_due_date, keep_field_private, set_due_date_extension, title_or_url @@ -276,6 +279,17 @@ def get(self, request, course_id): course_key = CourseKey.from_string(course_id) course = get_course_by_id(course_key) + original_user_is_staff = has_access(request.user, 'staff', course_key).has_access + _, request.user = setup_masquerade( + request, + course_key, + staff_access=original_user_is_staff, + ) + + if get_masquerade_role(request.user, course_key) == 'student' or \ + not request.user.has_perm(permissions.VIEW_DASHBOARD, course): + raise PermissionDenied('This user does not have access to the instructor dashboard.') + tabs = get_course_tab_list(request.user, course) context = { 'tabs': tabs, @@ -1621,7 +1635,8 @@ class RegenerateCertificatesView(DeveloperErrorViewMixin, APIView): **Request Body Parameters** statuses (optional): List of certificate statuses to regenerate - student_set (optional): "all" for all learners, "allowlisted" for allowlisted learners only + student_set (optional): "all" for all learners, "allowlisted" for allowlisted learners only, + "allowlisted_not_generated" for allowlisted learners without certificates **Response Values** @@ -1684,6 +1699,13 @@ def post(self, request, course_id): course_key, student_set='all_allowlisted' ) + elif student_set == 'allowlisted_not_generated': + # Generate for allowlisted students who don't have certificates yet + task = task_api.generate_certificates_for_students( + request, + course_key, + student_set='allowlisted_not_generated' + ) elif statuses: # Regenerate for specified statuses task = task_api.regenerate_certificates( @@ -4232,6 +4254,23 @@ def patch(self, request, course_id): return Response(serializer.data, status=status.HTTP_200_OK) +def add_or_replace_allowance_for_user(exam_id, username_or_email, key, value): + """ + Add an allowance for a user on an exam, removing any existing allowance with a different key. + + Enforces one allowance per user per exam regardless of allowance type. If the user already + has an allowance for this exam with a different key, it is removed before the new one is created. + """ + user_id = get_user_by_username_or_email(username_or_email).id + + with transaction.atomic(): + for allowance in ProctoredExamStudentAllowance.get_allowances_for_user(exam_id, user_id): + if allowance.key != key: + remove_allowance_for_user(exam_id, user_id, allowance.key) + + add_allowance_for_user(exam_id, username_or_email, key, value) + + class ExamAllowanceView(DeveloperErrorViewMixin, APIView): """ Grant, update, or remove an allowance for a student on a proctored exam. @@ -4288,17 +4327,17 @@ def post(self, request, course_id, exam_id): validated = serializer.validated_data results = [] - for user_info in validated['user_ids']: + for username_or_email in validated['user_ids']: try: - add_allowance_for_user( + add_or_replace_allowance_for_user( int(exam_id), - user_info, + username_or_email, validated['allowance_type'], validated['value'], ) - results.append({'identifier': user_info, 'success': True}) - except ProctoredBaseException as err: - results.append({'identifier': user_info, 'success': False, 'error': str(err)}) + results.append({'identifier': username_or_email, 'success': True}) + except (ProctoredBaseException, User.DoesNotExist, User.MultipleObjectsReturned) as err: + results.append({'identifier': username_or_email, 'success': False, 'error': str(err)}) return Response( {'allowance_type': validated['allowance_type'], 'results': results}, @@ -4374,6 +4413,40 @@ def delete(self, request, course_id, exam_id): ) +def _sort_in_memory(items, ordering): + """ + Sort a list of dicts by the given ordering param. + + Supports dotted paths (e.g. 'user.username') and descending with '-' prefix. + + Note: Sorting is done in Python because edx_proctoring's API functions + (get_all_exam_attempts, get_allowances_for_course) return pre-serialized + lists of dicts with no sorting parameter. Database-level sorting would + require changes to the edx-proctoring package: + https://github.com/openedx/edx-proctoring/issues/1320 + """ + if not ordering: + return items + descending = ordering.startswith('-') + field = ordering.lstrip('-') + + def sort_key(item): + value = item + for part in field.split('.'): + if isinstance(value, dict): + value = value.get(part) + else: + value = None + break + # Return a tuple (is_none, value) so None values sort last + # and non-None values compare naturally within their type. + if value is None: + return (1, '') + return (0, value) + + return sorted(items, key=sort_key, reverse=descending) + + class CourseAllowancesView(DeveloperErrorViewMixin, ListAPIView): """ List or bulk-create exam allowances for a course. @@ -4382,13 +4455,34 @@ class CourseAllowancesView(DeveloperErrorViewMixin, ListAPIView): GET /api/instructor/v2/courses/{course_id}/special_exams/allowances GET /api/instructor/v2/courses/{course_id}/special_exams/allowances?search=student1 + GET /api/instructor/v2/courses/{course_id}/special_exams/allowances?ordering=-value POST /api/instructor/v2/courses/{course_id}/special_exams/allowances + + **Query Parameters** + + search (optional): Filter by username or email. + ordering (optional): Sort by field. Prefix with '-' for descending. + Valid values: username, email, exam_name, allowance_type, value. + page (optional): Page number for pagination. + page_size (optional): Number of results per page. """ permission_classes = (IsAuthenticated, permissions.InstructorPermission) permission_name = permissions.EXAM_RESULTS serializer_class = ExamAllowanceSerializer + ORDERING_FIELDS = { + 'username': 'user.username', + 'user.username': 'user.username', + 'email': 'user.email', + 'user.email': 'user.email', + 'exam_name': 'proctored_exam.exam_name', + 'proctored_exam.exam_name': 'proctored_exam.exam_name', + 'allowance_type': 'key', + 'key': 'key', + 'value': 'value', + } + def get_queryset(self): course_id = self.kwargs['course_id'] allowances = get_allowances_for_course(course_id) @@ -4399,6 +4493,11 @@ def get_queryset(self): if search in a.get('user', {}).get('username', '').lower() or search in a.get('user', {}).get('email', '').lower() ] + ordering = self.request.query_params.get('ordering', '') + field = ordering.lstrip('-') + if field in self.ORDERING_FIELDS: + prefix = '-' if ordering.startswith('-') else '' + allowances = _sort_in_memory(allowances, prefix + self.ORDERING_FIELDS[field]) return allowances def post(self, request, course_id): @@ -4410,17 +4509,24 @@ def post(self, request, course_id): validated = serializer.validated_data results = [] for exam_id in validated['exam_ids']: - for user_info in validated['user_ids']: + for username_or_email in validated['user_ids']: try: - add_allowance_for_user( + add_or_replace_allowance_for_user( exam_id, - user_info, + username_or_email, validated['allowance_type'], validated['value'], ) - results.append({'identifier': user_info, 'exam_id': exam_id, 'success': True}) - except ProctoredBaseException as err: - results.append({'identifier': user_info, 'exam_id': exam_id, 'success': False, 'error': str(err)}) + results.append({'identifier': username_or_email, 'exam_id': exam_id, 'success': True}) + except (ProctoredBaseException, User.DoesNotExist, User.MultipleObjectsReturned) as err: + results.append( + { + 'identifier': username_or_email, + 'exam_id': exam_id, + 'success': False, + 'error': str(err) + } + ) return Response({ 'allowance_type': validated['allowance_type'], @@ -4431,23 +4537,62 @@ def post(self, request, course_id): class CourseExamAttemptsView(DeveloperErrorViewMixin, ListAPIView): """ - List all exam attempts across all exams in a course with optional search and pagination. + List all exam attempts across all exams in a course with optional search, sorting, and pagination. **Example Requests** GET /api/instructor/v2/courses/{course_id}/special_exams/attempts GET /api/instructor/v2/courses/{course_id}/special_exams/attempts?search=student1 + GET /api/instructor/v2/courses/{course_id}/special_exams/attempts?ordering=-started_at GET /api/instructor/v2/courses/{course_id}/special_exams/attempts?page=2&page_size=50 + + **Query Parameters** + + search (optional): Filter by username or email. + ordering (optional): Sort by field. Prefix with '-' for descending. + Valid values: username, exam_name, time_limit, type, started_at, completed_at, status. + page (optional): Page number for pagination. + page_size (optional): Number of results per page. """ permission_classes = (IsAuthenticated, permissions.InstructorPermission) permission_name = permissions.EXAM_RESULTS serializer_class = ExamAttemptSerializer + ORDERING_FIELDS = { + 'username': 'user.username', + 'user.username': 'user.username', + 'email': 'user.email', + 'user.email': 'user.email', + 'exam_name': 'proctored_exam.exam_name', + 'proctored_exam.exam_name': 'proctored_exam.exam_name', + 'time_limit': 'proctored_exam.time_limit_mins', + 'proctored_exam.time_limit_mins': 'proctored_exam.time_limit_mins', + 'started_at': 'started_at', + 'start_time': 'started_at', + 'completed_at': 'completed_at', + 'end_time': 'completed_at', + 'status': 'status', + } + + @staticmethod + def _get_exam_type(attempt): + """Derive exam type string for sorting purposes.""" + return derive_exam_type(attempt.get('proctored_exam', {})) + def get_queryset(self): course_id = self.kwargs['course_id'] search = self.request.query_params.get('search', '').strip() if search: - # get_filtered_exam_attempts does server-side filtering by username/email - return get_filtered_exam_attempts(course_id, search) - return get_all_exam_attempts(course_id) + attempts = get_filtered_exam_attempts(course_id, search) + else: + attempts = get_all_exam_attempts(course_id) + ordering = self.request.query_params.get('ordering', '') + field = ordering.lstrip('-') + if field == 'type': + descending = ordering.startswith('-') + attempts = sorted(attempts, key=self._get_exam_type, reverse=descending) + elif field in self.ORDERING_FIELDS: + prefix = '-' if ordering.startswith('-') else '' + attempts = _sort_in_memory(attempts, prefix + self.ORDERING_FIELDS[field]) + return attempts diff --git a/lms/djangoapps/instructor/views/serializers_v2.py b/lms/djangoapps/instructor/views/serializers_v2.py index 89fa151716c5..37f8f376ba8a 100644 --- a/lms/djangoapps/instructor/views/serializers_v2.py +++ b/lms/djangoapps/instructor/views/serializers_v2.py @@ -51,6 +51,7 @@ class CourseInformationSerializerV2(serializers.Serializer): enrollment statistics, permissions, and dashboard configuration. """ course_id = serializers.SerializerMethodField(help_text="Course run key") + username = serializers.SerializerMethodField(help_text="Username of the current authenticated user") display_name = serializers.SerializerMethodField(help_text="Course display name") org = serializers.SerializerMethodField(help_text="Organization identifier") course_number = serializers.SerializerMethodField(help_text="Course number") @@ -328,6 +329,10 @@ def get_course_id(self, data): """Get course ID as string.""" return str(data['course'].id) + def get_username(self, data): + """Get the username of the current authenticated user.""" + return data['user'].username + def get_display_name(self, data): """Get course display name.""" return data['course'].display_name @@ -530,7 +535,7 @@ class BlockDueDateSerializerV2(serializers.Serializer): block_id (str): The ID related to the block that needs the due date update. due_datetime (str): The new due date and time for the block. email_or_username (str): The email or username of the student whose access is being modified. - reason (str): Reason why updating this. + reason (str, optional): Reason why updating this. """ block_id = serializers.CharField() due_datetime = serializers.CharField() @@ -538,7 +543,7 @@ class BlockDueDateSerializerV2(serializers.Serializer): max_length=255, help_text="Email or username of user to change access" ) - reason = serializers.CharField(required=False) + reason = serializers.CharField(required=False, allow_blank=True, default='') def validate_email_or_username(self, value): """ @@ -842,10 +847,13 @@ class RegenerateCertificatesSerializer(serializers.Serializer): help_text="Certificate statuses to regenerate" ) student_set = serializers.ChoiceField( - choices=['all', 'allowlisted'], + choices=['all', 'allowlisted', 'allowlisted_not_generated'], required=False, default='all', - help_text="Student set filter" + help_text=( + "Student set filter: 'all' for all students, 'allowlisted' for all allowlisted students, " + "'allowlisted_not_generated' for allowlisted students without certificates" + ) ) @@ -857,7 +865,7 @@ class LearnerInputSerializer(serializers.Serializer): required=True, max_length=255, allow_blank=False, - help_text="Username or email address of the learner" + help_text="Username or email address of the learner", ) def validate_email_or_username(self, value): @@ -1166,6 +1174,23 @@ def to_internal_value(self, data): return super().to_internal_value(data) +def derive_exam_type(exam_dict): + """ + Derive exam type string from proctoring flags. + + Args: + exam_dict: dict with 'is_proctored' and 'is_practice_exam' keys. + + Returns: + 'practice', 'proctored', or 'timed'. + """ + if exam_dict.get('is_practice_exam'): + return 'practice' + if exam_dict.get('is_proctored'): + return 'proctored' + return 'timed' + + class SpecialExamSerializer(serializers.Serializer): """Serializer for proctored/timed exam data from edx_proctoring.""" id = serializers.IntegerField() @@ -1174,13 +1199,17 @@ class SpecialExamSerializer(serializers.Serializer): exam_name = serializers.CharField() time_limit_mins = serializers.IntegerField() due_date = serializers.DateTimeField(allow_null=True, required=False) - exam_type = serializers.CharField(required=False, default='') + exam_type = serializers.SerializerMethodField() is_proctored = serializers.BooleanField() is_practice_exam = serializers.BooleanField() is_active = serializers.BooleanField() hide_after_due = serializers.BooleanField() backend = serializers.CharField(allow_null=True, required=False) + def get_exam_type(self, obj): + """Derive exam type from proctoring flags.""" + return derive_exam_type(obj) + class ExamAttemptUserSerializer(serializers.Serializer): """Serializer for user info within an exam attempt.""" @@ -1194,10 +1223,17 @@ class ExamAttemptSerializer(serializers.Serializer): id = serializers.IntegerField() user = ExamAttemptUserSerializer() exam_id = serializers.IntegerField(source='proctored_exam.id') + exam_name = serializers.CharField(source='proctored_exam.exam_name') + exam_type = serializers.SerializerMethodField() status = serializers.CharField() start_time = serializers.DateTimeField(source='started_at', allow_null=True, required=False) end_time = serializers.DateTimeField(source='completed_at', allow_null=True, required=False) allowed_time_limit_mins = serializers.IntegerField(allow_null=True, required=False) + ready_to_resume = serializers.BooleanField() + + def get_exam_type(self, obj): + """Derive exam type from proctored_exam flags.""" + return derive_exam_type(obj.get('proctored_exam', {})) class ProctoringSettingsSerializer(serializers.Serializer):