from datetime import timedelta from django.utils import timezone from rest_framework import status from rest_framework.test import APITestCase, APIClient from labs.models import ( User, Exam, ExamAttempt, ExamAttemptSection, ExamQuestion, ExamSection, ExamVariant, ) class ExamAttemptCreateEndpointTests(APITestCase): def setUp(self): self.client = APIClient() self.user = User.objects.create_user(email="test@example.com") self.client.force_login(self.user) def test_create_attempt_validates_variant_id(self): resp = self.client.post("/api/exams/attempts", data={}) self.assertEqual(resp.status_code, status.HTTP_400_BAD_REQUEST) self.assertIn("variant_id", resp.data) resp = self.client.post("/api/exams/attempts", data={"variant_id": 999999}) self.assertEqual(resp.status_code, status.HTTP_400_BAD_REQUEST) self.assertEqual( resp.data["variant_id"], ["The specified exam variant does not exist."] ) def test_create_attempt_rejects_unpublished_exam_variant(self): exam = Exam.objects.create(code="draft", credits=0, published=False) variant = ExamVariant.objects.create( exam=exam, language_code="en", title="Draft (EN)", duration_seconds=3600, pass_score=1, total_score=1, ) resp = self.client.post("/api/exams/attempts", data={"variant_id": variant.id}) self.assertEqual(resp.status_code, status.HTTP_400_BAD_REQUEST) self.assertEqual( resp.data["variant_id"], ["The specified exam variant does not exist."] ) def test_create_attempt_fails_when_already_passed(self): exam = Exam.objects.create(code="exam_passed", credits=0, published=True) variant = ExamVariant.objects.create( exam=exam, language_code="en", title="Exam Passed (EN)", duration_seconds=3600, pass_score=1, total_score=1, ) now = timezone.now() ExamAttempt.objects.create( user=self.user, exam=exam, variant=variant, language_code="en", status=ExamAttempt.Status.COMPLETED, started_at=now - timedelta(minutes=10), submitted_at=now, expires_at=now + timedelta(minutes=50), score=1, is_passed=True, ) resp = self.client.post("/api/exams/attempts", data={"variant_id": variant.id}) self.assertEqual(resp.status_code, status.HTTP_400_BAD_REQUEST) self.assertEqual(resp.data["non_field_errors"], ["You have already passed this exam."]) self.assertEqual(ExamAttempt.objects.filter(user=self.user, exam=exam).count(), 1) def test_create_attempt_fails_when_active_attempt_exists(self): blocking_exam = Exam.objects.create(code="blocking", credits=0, published=True) blocking_variant = ExamVariant.objects.create( exam=blocking_exam, language_code="en", title="Blocking (EN)", duration_seconds=3600, pass_score=1, total_score=1, ) now = timezone.now() ExamAttempt.objects.create( user=self.user, exam=blocking_exam, variant=blocking_variant, language_code="en", status=ExamAttempt.Status.IN_PROGRESS, started_at=now - timedelta(minutes=1), expires_at=now + timedelta(minutes=59), ) target_exam = Exam.objects.create(code="target", credits=0, published=True) target_variant = ExamVariant.objects.create( exam=target_exam, language_code="en", title="Target (EN)", duration_seconds=3600, pass_score=1, total_score=1, ) resp = self.client.post( "/api/exams/attempts", data={"variant_id": target_variant.id}, ) self.assertEqual(resp.status_code, status.HTTP_400_BAD_REQUEST) self.assertEqual( resp.data["non_field_errors"], ["You already have an active exam in progress, please finish it first."], ) def test_create_attempt_employee_only_rejected_for_non_employee(self): exam = Exam.objects.create( code="employee_exam", credits=0, employee_only=True, published=True ) variant = ExamVariant.objects.create( exam=exam, language_code="en", title="Employee Only (EN)", duration_seconds=3600, pass_score=1, total_score=1, ) resp = self.client.post("/api/exams/attempts", data={"variant_id": variant.id}) self.assertEqual(resp.status_code, status.HTTP_400_BAD_REQUEST) self.assertEqual( resp.data["non_field_errors"], ["This exam is for PingCAP employees only."] ) self.assertEqual(ExamAttempt.objects.filter(user=self.user, exam=exam).count(), 0) def test_create_attempt_employee_only_allowed_for_pingcap_employee(self): employee = User.objects.create_user(email="employee@pingcap.com") self.client.force_login(employee) exam = Exam.objects.create( code="employee_exam", credits=0, employee_only=True, published=True ) variant = ExamVariant.objects.create( exam=exam, language_code="en", title="Employee Only (EN)", duration_seconds=3600, pass_score=1, total_score=1, ) resp = self.client.post("/api/exams/attempts", data={"variant_id": variant.id}) self.assertEqual(resp.status_code, status.HTTP_200_OK) self.assertIsNotNone(resp.data.get("id")) self.assertEqual(resp.data["exam_id"], exam.id) self.assertEqual(ExamAttempt.objects.filter(user=employee, exam=exam).count(), 1) class ExamSectionsQuestionsFlowTests(APITestCase): def setUp(self): self.client = APIClient() self.user = User.objects.create_user(email="flow@example.com") self.client.force_login(self.user) self.exam = Exam.objects.create(code="exam_flow", credits=0, published=True) self.variant = ExamVariant.objects.create( exam=self.exam, language_code="en", title="Exam Flow (EN)", duration_seconds=3600, pass_score=1, total_score=2, ) self.instruction = ExamSection.objects.create( variant=self.variant, title="Instruction", section_type=ExamSection.SectionType.INSTRUCTION, content={"text": "Read this"}, display_order=0, ) self.questions = ExamSection.objects.create( variant=self.variant, title="Questions", section_type=ExamSection.SectionType.QUESTION, content={}, display_order=1, ) self.q1 = ExamQuestion.objects.create( variant=self.variant, section=self.questions, score=1, question_type=ExamQuestion.QuestionType.SINGLE_CHOICE, stem="Q1", content=[{"id": "A", "text": "A"}, {"id": "B", "text": "B"}], answer=["A"], display_order=0, ) self.q2 = ExamQuestion.objects.create( variant=self.variant, section=self.questions, score=1, question_type=ExamQuestion.QuestionType.MULTIPLE_CHOICE, stem="Q2", content=[{"id": "A", "text": "A"}, {"id": "B", "text": "B"}], answer=["A", "B"], display_order=1, ) resp = self.client.post( "/api/exams/attempts", data={"variant_id": self.variant.id}, ) self.assertEqual(resp.status_code, status.HTTP_200_OK) self.attempt_id = resp.data["id"] def test_attempt_detail_and_access_control(self): resp = self.client.get(f"/api/exams/attempts/{self.attempt_id}") self.assertEqual(resp.status_code, status.HTTP_200_OK) self.assertGreater(resp.data["remaining_seconds"], 0) other = User.objects.create_user(email="other@example.com") self.client.force_login(other) resp = self.client.get(f"/api/exams/attempts/{self.attempt_id}") self.assertEqual(resp.status_code, status.HTTP_404_NOT_FOUND) self.assertEqual(resp.data["detail"], "Exam attempt not found") def test_sections_questions_and_submit_flow(self): resp = self.client.get(f"/api/exams/attempts/{self.attempt_id}/sections") self.assertEqual(resp.status_code, status.HTTP_200_OK) self.assertEqual(len(resp.data), 2) self.assertEqual([s["section_id"] for s in resp.data], [self.instruction.id, self.questions.id]) self.assertEqual(resp.data[0]["status"], ExamAttemptSection.Status.AVAILABLE) self.assertEqual(resp.data[1]["status"], ExamAttemptSection.Status.LOCKED) resp = self.client.get( f"/api/exams/attempts/{self.attempt_id}/sections/{self.instruction.id}/questions" ) self.assertEqual(resp.status_code, status.HTTP_404_NOT_FOUND) resp = self.client.get( f"/api/exams/attempts/{self.attempt_id}/sections/{self.questions.id}" ) self.assertEqual(resp.status_code, status.HTTP_409_CONFLICT) self.assertEqual(resp.data["detail"], "This section is not available.") resp = self.client.get( f"/api/exams/attempts/{self.attempt_id}/sections/{self.instruction.id}" ) self.assertEqual(resp.status_code, status.HTTP_200_OK) self.assertEqual(resp.data["content"], {"text": "Read this"}) resp = self.client.post( f"/api/exams/attempts/{self.attempt_id}/sections/{self.instruction.id}/submit" ) self.assertEqual(resp.status_code, status.HTTP_204_NO_CONTENT) resp = self.client.get(f"/api/exams/attempts/{self.attempt_id}/sections") self.assertEqual(resp.status_code, status.HTTP_200_OK) self.assertEqual(resp.data[1]["status"], ExamAttemptSection.Status.AVAILABLE) resp = self.client.get( f"/api/exams/attempts/{self.attempt_id}/sections/{self.questions.id}" ) self.assertEqual(resp.status_code, status.HTTP_200_OK) self.assertIsNone(resp.data["content"]) resp = self.client.get( f"/api/exams/attempts/{self.attempt_id}/sections/{self.questions.id}/questions" ) self.assertEqual(resp.status_code, status.HTTP_200_OK) self.assertEqual([q["id"] for q in resp.data], [self.q1.id, self.q2.id]) self.assertFalse(resp.data[0]["is_answered"]) resp = self.client.get( f"/api/exams/attempts/{self.attempt_id}/sections/{self.questions.id}/questions/{self.q1.id}" ) self.assertEqual(resp.status_code, status.HTTP_200_OK) self.assertIsNone(resp.data["user_answer"]) resp = self.client.put( f"/api/exams/attempts/{self.attempt_id}/sections/{self.questions.id}/questions/{self.q1.id}", data={"answer": ["A"]}, format="json", ) self.assertEqual(resp.status_code, status.HTTP_204_NO_CONTENT) resp = self.client.get( f"/api/exams/attempts/{self.attempt_id}/sections/{self.questions.id}/questions" ) self.assertEqual(resp.status_code, status.HTTP_200_OK) self.assertTrue(resp.data[0]["is_answered"]) resp = self.client.put( f"/api/exams/attempts/{self.attempt_id}/sections/{self.questions.id}/questions/{self.q1.id}", data={"answer": []}, format="json", ) self.assertEqual(resp.status_code, status.HTTP_204_NO_CONTENT) resp = self.client.get( f"/api/exams/attempts/{self.attempt_id}/sections/{self.questions.id}/questions" ) self.assertEqual(resp.status_code, status.HTTP_200_OK) self.assertFalse(resp.data[0]["is_answered"]) resp = self.client.post( f"/api/exams/attempts/{self.attempt_id}/sections/{self.questions.id}/submit" ) self.assertEqual(resp.status_code, status.HTTP_204_NO_CONTENT) resp = self.client.get(f"/api/exams/attempts/{self.attempt_id}") self.assertEqual(resp.status_code, status.HTTP_200_OK) self.assertEqual(resp.data["status"], ExamAttempt.Status.COMPLETED) self.assertEqual(resp.data["remaining_seconds"], 0) self.assertIsNotNone(resp.data["submitted_at"]) def test_sections_list_requires_in_progress_attempt(self): db_attempt = ExamAttempt.objects.get(id=self.attempt_id) db_attempt.status = ExamAttempt.Status.COMPLETED db_attempt.save(update_fields=["status"]) resp = self.client.get(f"/api/exams/attempts/{self.attempt_id}/sections") self.assertEqual(resp.status_code, status.HTTP_403_FORBIDDEN) def test_expired_in_progress_attempt_rejects_sections_and_questions(self): db_attempt = ExamAttempt.objects.get(id=self.attempt_id) db_attempt.expires_at = timezone.now() - timedelta(seconds=1) db_attempt.save(update_fields=["expires_at"]) resp = self.client.get(f"/api/exams/attempts/{self.attempt_id}/sections") self.assertEqual(resp.status_code, status.HTTP_403_FORBIDDEN) self.assertEqual(resp.data["detail"], "This exam attempt has expired.") resp = self.client.get( f"/api/exams/attempts/{self.attempt_id}/sections/{self.questions.id}/questions" ) self.assertEqual(resp.status_code, status.HTTP_403_FORBIDDEN) self.assertEqual(resp.data["detail"], "This exam attempt has expired.") def test_cross_user_access_to_sections_and_questions_returns_404(self): other = User.objects.create_user(email="other-sections@example.com") self.client.force_login(other) resp = self.client.get(f"/api/exams/attempts/{self.attempt_id}/sections") self.assertEqual(resp.status_code, status.HTTP_404_NOT_FOUND) self.assertEqual(resp.data["detail"], "Exam attempt not found") resp = self.client.get( f"/api/exams/attempts/{self.attempt_id}/sections/{self.questions.id}/questions" ) self.assertEqual(resp.status_code, status.HTTP_404_NOT_FOUND) self.assertEqual(resp.data["detail"], "Section not found") def test_locked_section_submit_and_answer_return_409(self): resp = self.client.post( f"/api/exams/attempts/{self.attempt_id}/sections/{self.questions.id}/submit" ) self.assertEqual(resp.status_code, status.HTTP_409_CONFLICT) self.assertEqual(resp.data["detail"], "This section is not available.") resp = self.client.put( f"/api/exams/attempts/{self.attempt_id}/sections/{self.questions.id}/questions/{self.q1.id}", data={"answer": ["A"]}, format="json", ) self.assertEqual(resp.status_code, status.HTTP_409_CONFLICT) self.assertEqual(resp.data["detail"], "This section is not available.") def test_submit_same_section_twice_returns_409(self): resp = self.client.post( f"/api/exams/attempts/{self.attempt_id}/sections/{self.instruction.id}/submit" ) self.assertEqual(resp.status_code, status.HTTP_204_NO_CONTENT) resp = self.client.post( f"/api/exams/attempts/{self.attempt_id}/sections/{self.instruction.id}/submit" ) self.assertEqual(resp.status_code, status.HTTP_409_CONFLICT) self.assertEqual(resp.data["detail"], "This section is not available.")