Spaces:
Running
Running
| import json | |
| import unittest | |
| from fastapi import FastAPI | |
| from fastapi.testclient import TestClient | |
| from sqlalchemy import create_engine | |
| from sqlalchemy.orm import sessionmaker | |
| from sqlalchemy.pool import StaticPool | |
| from app.api.deps import require_admin | |
| from app.api.routes.admin import router | |
| from app.core.database import Base, get_db | |
| from app.models.models import Question, QuestionType, Test, User, UserRole | |
| class AdminQuestionFileUploadTests(unittest.TestCase): | |
| def setUp(self): | |
| self.engine = create_engine( | |
| "sqlite:///:memory:", | |
| connect_args={"check_same_thread": False}, | |
| poolclass=StaticPool, | |
| ) | |
| Base.metadata.create_all(self.engine) | |
| self.SessionLocal = sessionmaker(bind=self.engine) | |
| db = self.SessionLocal() | |
| try: | |
| self.admin = User(email="admin@example.com", full_name="Admin", role=UserRole.admin) | |
| self.test = Test(title="File import", created_by=1, total_marks=0.0) | |
| db.add_all([self.admin, self.test]) | |
| db.commit() | |
| db.refresh(self.admin) | |
| db.refresh(self.test) | |
| self.test_id = self.test.id | |
| finally: | |
| db.close() | |
| app = FastAPI() | |
| app.include_router(router) | |
| def override_get_db(): | |
| db = self.SessionLocal() | |
| try: | |
| yield db | |
| finally: | |
| db.close() | |
| def override_require_admin(): | |
| return self.admin | |
| app.dependency_overrides[get_db] = override_get_db | |
| app.dependency_overrides[require_admin] = override_require_admin | |
| self.client = TestClient(app) | |
| def test_upload_questions_file_imports_valid_questions(self): | |
| payload = { | |
| "questions": [ | |
| { | |
| "question_type": "mcq", | |
| "question_text": "Pick the stable sort.", | |
| "options": ["Quick sort", "Heap sort", "Merge sort", "Selection sort"], | |
| "correct_answer": "C", | |
| "marks": 2, | |
| "negative_marks": 0.66, | |
| "order_index": 4, | |
| "subject": "Algorithms", | |
| "topic": "Sorting", | |
| } | |
| ] | |
| } | |
| response = self.client.post( | |
| f"/admin/tests/{self.test_id}/questions/upload-file", | |
| files={"file": ("questions.json", json.dumps(payload), "application/json")}, | |
| ) | |
| self.assertEqual(response.status_code, 201) | |
| self.assertEqual(response.json()["imported_count"], 1) | |
| db = self.SessionLocal() | |
| try: | |
| question = db.query(Question).one() | |
| test = db.query(Test).filter(Test.id == self.test_id).one() | |
| self.assertEqual(question.question_type, QuestionType.mcq) | |
| self.assertEqual(question.correct_answer, "C") | |
| self.assertEqual(question.order_index, 4) | |
| self.assertEqual(test.total_marks, 2.0) | |
| finally: | |
| db.close() | |
| def test_upload_questions_file_accepts_nat_colon_range(self): | |
| payload = { | |
| "questions": [ | |
| { | |
| "question_type": "nat", | |
| "question_text": "Probability that the slot is empty.", | |
| "options": [], | |
| "correct_answer": "0.44:0.45", | |
| "marks": 2, | |
| "negative_marks": 0, | |
| "order_index": 17, | |
| } | |
| ] | |
| } | |
| response = self.client.post( | |
| f"/admin/tests/{self.test_id}/questions/upload-file", | |
| files={"file": ("questions.json", json.dumps(payload), "application/json")}, | |
| ) | |
| self.assertEqual(response.status_code, 201) | |
| self.assertEqual(response.json()["imported_count"], 1) | |
| db = self.SessionLocal() | |
| try: | |
| question = db.query(Question).one() | |
| self.assertEqual(question.question_type, QuestionType.nat) | |
| self.assertEqual(question.correct_answer, "0.44:0.45") | |
| self.assertEqual(question.order_index, 17) | |
| finally: | |
| db.close() | |
| def test_upload_questions_file_rejects_invalid_questions_without_insert(self): | |
| payload = [ | |
| { | |
| "question_type": "mcq", | |
| "question_text": "Bad answer", | |
| "options": ["A", "B", "C", "D"], | |
| "correct_answer": "Z", | |
| } | |
| ] | |
| response = self.client.post( | |
| f"/admin/tests/{self.test_id}/questions/upload-file", | |
| files={"file": ("questions.json", json.dumps(payload), "application/json")}, | |
| ) | |
| self.assertEqual(response.status_code, 422) | |
| self.assertIn("errors", response.json()["detail"]) | |
| db = self.SessionLocal() | |
| try: | |
| self.assertEqual(db.query(Question).count(), 0) | |
| finally: | |
| db.close() | |
| def test_admin_can_update_imported_question_and_total_marks(self): | |
| db = self.SessionLocal() | |
| try: | |
| question = Question( | |
| test_id=self.test_id, | |
| question_type=QuestionType.mcq, | |
| question_text="Imported question", | |
| options=["A1", "B1", "C1", "D1"], | |
| correct_answer="A", | |
| marks=2.0, | |
| negative_marks=0.66, | |
| order_index=0, | |
| ) | |
| test = db.query(Test).filter(Test.id == self.test_id).one() | |
| test.total_marks = 2.0 | |
| db.add(question) | |
| db.commit() | |
| db.refresh(question) | |
| question_id = question.id | |
| finally: | |
| db.close() | |
| payload = { | |
| "question_type": "msq", | |
| "question_text": "Corrected question", | |
| "options": ["A2", "B2", "C2", "D2"], | |
| "correct_answer": "A,C", | |
| "marks": 3.0, | |
| "negative_marks": 0.33, | |
| "subject": "Data Structures", | |
| "topic": "Trees", | |
| } | |
| response = self.client.patch( | |
| f"/admin/tests/{self.test_id}/questions/{question_id}", | |
| json=payload, | |
| ) | |
| self.assertEqual(response.status_code, 200) | |
| body = response.json() | |
| self.assertEqual(body["question_type"], "msq") | |
| self.assertEqual(body["question_text"], "Corrected question") | |
| self.assertEqual(body["options"], ["A2", "B2", "C2", "D2"]) | |
| self.assertEqual(body["correct_answer"], "A,C") | |
| self.assertEqual(body["negative_marks"], 0.0) | |
| db = self.SessionLocal() | |
| try: | |
| question = db.query(Question).filter(Question.id == question_id).one() | |
| test = db.query(Test).filter(Test.id == self.test_id).one() | |
| self.assertEqual(question.question_type, QuestionType.msq) | |
| self.assertEqual(question.marks, 3.0) | |
| self.assertEqual(test.total_marks, 3.0) | |
| finally: | |
| db.close() | |
| def test_update_question_rejects_invalid_answer(self): | |
| db = self.SessionLocal() | |
| try: | |
| question = Question( | |
| test_id=self.test_id, | |
| question_type=QuestionType.mcq, | |
| question_text="Imported question", | |
| options=["A1", "B1", "C1", "D1"], | |
| correct_answer="A", | |
| marks=2.0, | |
| negative_marks=0.66, | |
| order_index=0, | |
| ) | |
| db.add(question) | |
| db.commit() | |
| db.refresh(question) | |
| question_id = question.id | |
| finally: | |
| db.close() | |
| response = self.client.patch( | |
| f"/admin/tests/{self.test_id}/questions/{question_id}", | |
| json={ | |
| "question_type": "mcq", | |
| "question_text": "Still invalid", | |
| "options": ["A", "B", "C", "D"], | |
| "correct_answer": "Z", | |
| "marks": 2.0, | |
| "negative_marks": 0.66, | |
| }, | |
| ) | |
| self.assertEqual(response.status_code, 422) | |
| if __name__ == "__main__": | |
| unittest.main() | |