feat: add bulk certificate exceptions CSV upload endpoint

lms/djangoapps/instructor/tests/test_certificates_api_v2.py

@@ -1,6 +1,7 @@
 """
 Unit tests for instructor API v2 certificate management endpoints.
 """
+from io import BytesIO
 from unittest.mock import patch
 
 from django.urls import reverse
@@ -266,6 +267,233 @@ def test_delete_successful(self, mock_get_entry, mock_remove):
         mock_remove.assert_called_once_with(self.enrolled_student, self.course.id)
 
 
+class BulkCertificateExceptionsViewTest(SharedModuleStoreTestCase):
+    """Tests for BulkCertificateExceptionsView."""
+
+    @classmethod
+    def setUpClass(cls):
+        super().setUpClass()
+        cls.course = CourseFactory.create()
+
+    def setUp(self):
+        super().setUp()
+        self.client = APIClient()
+        self.instructor = InstructorFactory.create(course_key=self.course.id)
+        self.student = UserFactory.create()
+        self.url = reverse(
+            'instructor_api_v2:bulk_certificate_exceptions',
+            kwargs={'course_id': str(self.course.id)}
+        )
+
+    def _create_csv_file(self, content):
+        """Helper to create a CSV file upload."""
+        csv_file = BytesIO(content.encode('utf-8'))
+        csv_file.name = 'test.csv'
+        return csv_file
+
+    def test_permission_required(self):
+        """Test that only instructors can upload bulk exceptions."""
+        self.client.force_authenticate(user=self.student)
+        csv_file = self._create_csv_file('user1,notes1')
+        response = self.client.post(self.url, {'file': csv_file}, format='multipart')
+        assert response.status_code == status.HTTP_403_FORBIDDEN
+
+    def test_no_file_uploaded(self):
+        """Test error when no file is uploaded."""
+        self.client.force_authenticate(user=self.instructor)
+        response = self.client.post(self.url, {}, format='multipart')
+        assert response.status_code == status.HTTP_400_BAD_REQUEST
+        assert 'No file uploaded' in response.data['message']
+
+    def test_non_csv_file_type(self):
+        """Test error when uploaded file is not CSV."""
+        self.client.force_authenticate(user=self.instructor)
+        txt_file = BytesIO(b'user1,notes1')
+        txt_file.name = 'test.txt'
+        response = self.client.post(self.url, {'file': txt_file}, format='multipart')
+        assert response.status_code == status.HTTP_400_BAD_REQUEST
+        assert 'CSV format' in response.data['message']
+
+    def test_empty_csv(self):
+        """Test error when CSV file is empty."""
+        self.client.force_authenticate(user=self.instructor)
+        csv_file = self._create_csv_file('')
+        response = self.client.post(self.url, {'file': csv_file}, format='multipart')
+        assert response.status_code == status.HTTP_400_BAD_REQUEST
+        assert 'empty' in response.data['message']
+
+    def test_csv_with_only_empty_rows(self):
+        """Test error when CSV contains only empty rows."""
+        self.client.force_authenticate(user=self.instructor)
+        csv_file = self._create_csv_file('\n\n  \n')
+        response = self.client.post(self.url, {'file': csv_file}, format='multipart')
+        assert response.status_code == status.HTTP_400_BAD_REQUEST
+        assert 'empty' in response.data['message']
+
+    @patch('lms.djangoapps.instructor.views.api_v2.certs_api.create_or_update_certificate_allowlist_entry')
+    def test_happy_path_csv(self, mock_create):
+        """Test successful bulk upload with valid CSV."""
+        student1 = UserFactory.create(username='student1')
+        student2 = UserFactory.create(username='student2', email='[email protected]')
+        CourseEnrollmentFactory.create(user=student1, course_id=self.course.id)
+        CourseEnrollmentFactory.create(user=student2, course_id=self.course.id)
+
+        self.client.force_authenticate(user=self.instructor)
+        csv_content = 'student1,First student notes\[email protected],Second student notes'
+        csv_file = self._create_csv_file(csv_content)
+        response = self.client.post(self.url, {'file': csv_file}, format='multipart')
+
+        assert response.status_code == status.HTTP_200_OK
+        assert len(response.data['success']) == 2
+        assert 'student1' in response.data['success']
+        assert '[email protected]' in response.data['success']
+        assert len(response.data['errors']) == 0
+        assert mock_create.call_count == 2
+
+    @patch('lms.djangoapps.instructor.views.api_v2.certs_api.create_or_update_certificate_allowlist_entry')
+    def test_csv_without_notes_column(self, mock_create):
+        """Test CSV with only username column (no notes)."""
+        student1 = UserFactory.create(username='student1')
+        CourseEnrollmentFactory.create(user=student1, course_id=self.course.id)
+
+        self.client.force_authenticate(user=self.instructor)
+        csv_content = 'student1'
+        csv_file = self._create_csv_file(csv_content)
+        response = self.client.post(self.url, {'file': csv_file}, format='multipart')
+
+        assert response.status_code == status.HTTP_200_OK
+        assert len(response.data['success']) == 1
+        # Verify empty notes were passed
+        call_args = mock_create.call_args
+        assert call_args[0][2] == ''  # notes parameter
+
+    def test_unresolvable_learners(self):
+        """Test error handling for users that don't exist."""
+        self.client.force_authenticate(user=self.instructor)
+        csv_content = 'nonexistent1,notes1\nnonexistent2,notes2'
+        csv_file = self._create_csv_file(csv_content)
+        response = self.client.post(self.url, {'file': csv_file}, format='multipart')
+
+        assert response.status_code == status.HTTP_200_OK
+        assert len(response.data['success']) == 0
+        assert len(response.data['errors']) == 2
+        assert any('nonexistent1' in str(err) for err in response.data['errors'])
+        assert any('nonexistent2' in str(err) for err in response.data['errors'])
+
+    @patch('lms.djangoapps.instructor.views.api_v2.certs_api.create_or_update_certificate_allowlist_entry')
+    def test_partial_success(self, mock_create):
+        """Test mix of valid and invalid learners in CSV."""
+        student1 = UserFactory.create(username='valid_user')
+        CourseEnrollmentFactory.create(user=student1, course_id=self.course.id)
+
+        self.client.force_authenticate(user=self.instructor)
+        csv_content = 'valid_user,Valid notes\ninvalid_user,Invalid notes'
+        csv_file = self._create_csv_file(csv_content)
+        response = self.client.post(self.url, {'file': csv_file}, format='multipart')
+
+        assert response.status_code == status.HTTP_200_OK
+        assert len(response.data['success']) == 1
+        assert 'valid_user' in response.data['success']
+        assert len(response.data['errors']) == 1
+        assert any('invalid_user' in str(err) for err in response.data['errors'])
+        mock_create.assert_called_once()
+
+    @patch('lms.djangoapps.instructor.views.api_v2.certs_api.create_or_update_certificate_allowlist_entry')
+    def test_duplicate_csv_identifiers(self, mock_create):
+        """Test that duplicate identifiers use last occurrence's notes."""
+        student1 = UserFactory.create(username='student1')
+        CourseEnrollmentFactory.create(user=student1, course_id=self.course.id)
+
+        self.client.force_authenticate(user=self.instructor)
+        # Same identifier twice with different notes
+        csv_content = 'student1,First notes\nstudent1,Last notes'
+        csv_file = self._create_csv_file(csv_content)
+        response = self.client.post(self.url, {'file': csv_file}, format='multipart')
+
+        assert response.status_code == status.HTTP_200_OK
+        assert len(response.data['success']) == 1
+        # Verify the last notes value was used (dict behavior)
+        call_args = mock_create.call_args
+        assert call_args[0][2] == 'Last notes'
+
+    @patch('lms.djangoapps.instructor.views.api_v2.certs_api.create_or_update_certificate_allowlist_entry')
+    def test_csv_with_empty_notes(self, mock_create):
+        """Test CSV rows with empty notes column."""
+        student1 = UserFactory.create(username='student1')
+        CourseEnrollmentFactory.create(user=student1, course_id=self.course.id)
+
+        self.client.force_authenticate(user=self.instructor)
+        csv_content = 'student1,'
+        csv_file = self._create_csv_file(csv_content)
+        response = self.client.post(self.url, {'file': csv_file}, format='multipart')
+
+        assert response.status_code == status.HTTP_200_OK
+        assert len(response.data['success']) == 1
+        call_args = mock_create.call_args
+        assert call_args[0][2] == ''
+
+    @patch('lms.djangoapps.instructor.views.api_v2.certs_api.create_or_update_certificate_allowlist_entry')
+    def test_unenrolled_learner(self, mock_create):
+        """Test error when learner exists but is not enrolled in course."""
+        UserFactory.create(username='unenrolled')
+        # Don't enroll the student
+
+        self.client.force_authenticate(user=self.instructor)
+        csv_content = 'unenrolled,notes'
+        csv_file = self._create_csv_file(csv_content)
+        response = self.client.post(self.url, {'file': csv_file}, format='multipart')
+
+        assert response.status_code == status.HTTP_200_OK
+        assert len(response.data['success']) == 0
+        assert len(response.data['errors']) == 1
+        assert 'not enrolled' in response.data['errors'][0]['message']
+        mock_create.assert_not_called()
+
+    @patch('lms.djangoapps.instructor.views.api_v2.certs_api.create_or_update_certificate_allowlist_entry')
+    def test_learner_with_active_invalidation(self, mock_create):
+        """Test error when learner has an active certificate invalidation."""
+        student1 = UserFactory.create(username='invalidated')
+        CourseEnrollmentFactory.create(user=student1, course_id=self.course.id)
+        cert = GeneratedCertificateFactory.create(
+            user=student1,
+            course_id=self.course.id,
+            status=CertificateStatuses.unavailable
+        )
+        CertificateInvalidation.objects.create(
+            generated_certificate=cert,
+            invalidated_by=self.instructor,
+            notes='Test invalidation',
+            active=True
+        )
+
+        self.client.force_authenticate(user=self.instructor)
+        csv_content = 'invalidated,notes'
+        csv_file = self._create_csv_file(csv_content)
+        response = self.client.post(self.url, {'file': csv_file}, format='multipart')
+
+        assert response.status_code == status.HTTP_200_OK
+        assert len(response.data['success']) == 0
+        assert len(response.data['errors']) == 1
+        assert 'invalidation' in response.data['errors'][0]['message']
+        mock_create.assert_not_called()
+
+    @patch('lms.djangoapps.instructor.views.api_v2.certs_api.create_or_update_certificate_allowlist_entry')
+    def test_csv_with_utf8_bom(self, mock_create):
+        """Test CSV file with UTF-8 BOM is handled correctly."""
+        student1 = UserFactory.create(username='student1')
+        CourseEnrollmentFactory.create(user=student1, course_id=self.course.id)
+
+        self.client.force_authenticate(user=self.instructor)
+        # UTF-8 BOM + CSV content
+        csv_content = '\ufeffstudent1,notes'
+        csv_file = self._create_csv_file(csv_content)
+        response = self.client.post(self.url, {'file': csv_file}, format='multipart')
+
+        assert response.status_code == status.HTTP_200_OK
+        assert len(response.data['success']) == 1
+        mock_create.assert_called_once()
+
+
 class CertificateInvalidationsViewTest(SharedModuleStoreTestCase):
     """Tests for CertificateInvalidationsView."""
 

lms/djangoapps/instructor/views/api_urls.py

@@ -96,6 +96,11 @@
         api_v2.CertificateExceptionsView.as_view(),
         name='certificate_exceptions'
     ),
+    re_path(
+        rf'^courses/{COURSE_ID_PATTERN}/certificates/exceptions/bulk$',
+        api_v2.BulkCertificateExceptionsView.as_view(),
+        name='bulk_certificate_exceptions'
+    ),
     re_path(
         rf'^courses/{COURSE_ID_PATTERN}/certificates/invalidations$',
         api_v2.CertificateInvalidationsView.as_view(),

lms/djangoapps/instructor/views/api_v2.py

@@ -2119,6 +2119,120 @@ def _validate_certificates_for_invalidation(learner_to_user, course_key):
     return certificates_to_invalidate, errors
 
 
+class BulkCertificateExceptionsView(DeveloperErrorViewMixin, APIView):
+    """
+    View to grant certificate exceptions via CSV upload.
+
+    **Example Requests**
+
+        POST /api/instructor/v2/courses/{course_id}/certificates/exceptions/bulk
+
+    **POST Request Body**
+
+        Form data with CSV file uploaded as 'file' field.
+        CSV format: username_or_email,notes (optional second column)
+
+    **Returns**
+
+        * 200: OK - Bulk exceptions processed with success/error details
+        * 400: Bad Request - Invalid CSV file or format
+        * 401: Unauthorized - User is not authenticated
+        * 403: Forbidden - User lacks instructor permissions
+    """
+    permission_classes = (IsAuthenticated, permissions.InstructorPermission)
+    permission_name = permissions.CERTIFICATE_EXCEPTION_VIEW
+
+    def post(self, request, course_id):
+        """Grant certificate exceptions via CSV upload."""
+        course_key = CourseKey.from_string(course_id)
+        # Validate that the course exists
+        get_course_by_id(course_key)
+
+        # Check if file was uploaded
+        if 'file' not in request.FILES:
+            return Response(
+                {'message': _('No file uploaded')},
+                status=status.HTTP_400_BAD_REQUEST
+            )
+
+        uploaded_file = request.FILES['file']
+
+        # Validate file type
+        if not uploaded_file.name.endswith('.csv'):
+            return Response(
+                {'message': _('File must be in CSV format')},
+                status=status.HTTP_400_BAD_REQUEST
+            )
+
+        results = {
+            'success': [],
+            'errors': []
+        }
+
+        try:
+            file_content = uploaded_file.read().decode('utf-8-sig')
+            csv_reader = list(csv.reader(file_content.splitlines()))
+        except (UnicodeDecodeError, csv.Error) as exc:
+            log.exception("Error processing CSV file for certificate exceptions")
+            return Response(
+                {'message': _('Error processing CSV file: {error}').format(error=str(exc))},
+                status=status.HTTP_400_BAD_REQUEST
+            )
+
+        learners_with_notes = []
+        for row in csv_reader:
+            if not row or not row[0].strip():
+                continue  # Skip empty rows
+
+            learner = row[0].strip()
+            notes = row[1].strip() if len(row) > 1 and row[1].strip() else ''
+
+            learners_with_notes.append((learner, notes))
+
+        if not learners_with_notes:
+            return Response(
+                {'message': _('CSV file is empty or contains no valid entries')},
+                status=status.HTTP_400_BAD_REQUEST
+            )
+
+        # Extract learners for resolution and build a notes lookup
+        learners = [learner for learner, _ in learners_with_notes]
+        notes_by_learner = dict(learners_with_notes)
+
+        # Resolve all usernames/emails to users upfront
+        learner_to_user, user_errors = _resolve_learners_to_users(learners)
+        results['errors'].extend(user_errors)
+
+        # Validate learners for certificate exceptions
+        exceptions_to_create, validation_errors = _validate_learners_for_certificate_exceptions(
+            learner_to_user, course_key
+        )
+        results['errors'].extend(validation_errors)
+
+        # Create all exceptions using the certificates API
+        for learner, user in exceptions_to_create:
+            notes = notes_by_learner.get(learner, '')
+
+            try:
+                certs_api.create_or_update_certificate_allowlist_entry(user, course_key, notes)
+                log.info(
+                    "Certificate exception granted for user %s (%s) in course %s by %s via CSV upload",
+                    user.id, learner, course_key, request.user.username
+                )
+                results['success'].append(learner)
+            except Exception as exc:  # pylint: disable=broad-except
+                log.exception(
+                    "Error creating certificate exception for user %s in course %s",
+                    user.id, course_key
+                )
+                results['errors'].append({
+                    'learner': learner,
+                    'message': str(exc)
+                })
+
+        return Response(results, status=status.HTTP_200_OK)
+
+
 class CertificateInvalidationsView(DeveloperErrorViewMixin, APIView):
     """
     View to invalidate or re-validate certificates.