pdf-quiz-generator/backend/tests/test_quiz_builder.py
Daniel f696b99569 fix: close custom-test review gaps and verify grading
Handle ownerless question revocation, legacy hide sharing, private deletion, selected-set grading including skips, UI validation/reparent/delete safeguards and offline-safe hierarchy migration. Verified 14 backend tests in deployed image, 13 frontend tests/build and real disposable PostgreSQL migration round-trip. Related tutor/image privacy work remains before deployment.
2026-09-07 02:07:58 +02:00

355 lines
22 KiB
Python

"""Run: DATABASE_URL=sqlite:///:memory: PYTHONPATH=backend python -m unittest discover -s backend/tests.
No application startup, external services or AI calls; every test uses a disposable SQLite database.
"""
import os
os.environ["DATABASE_URL"] = "sqlite:///:memory:"
import sys
import unittest
from datetime import datetime, timedelta
from types import ModuleType
from unittest.mock import Mock, patch
from fastapi import FastAPI
from fastapi.testclient import TestClient
from sqlalchemy import create_engine
from sqlalchemy.orm import Session
from sqlalchemy.pool import StaticPool
from app.database import Base, get_db
from app.models.user import User
from app.models.question import Question
from app.models.question_category import QuestionCategory
from app.models.quiz import Quiz
from app.models.quiz_question_link import QuizQuestionLink
from app.models.attempt import QuizAttempt, AttemptAnswer
from app.models.favorite import Favorite
from app.models.course import Course, CourseEnrollment
from app.services.quiz_builder import category_descendants
from app.utils.auth import get_current_user
from app.routers import questions, question_categories, attempts, mobile
# Extraction is outside this milestone. Stub only its unused service import;
# the real routes, ORM, predicates, creation and authorization run below.
with patch.dict(sys.modules, {"app.services.quiz_service": ModuleType("app.services.quiz_service")}):
from app.routers import quizzes
class BuilderTests(unittest.TestCase):
def setUp(self):
self.engine = create_engine("sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool)
with self.engine.connect() as conn:
conn.exec_driver_sql("PRAGMA foreign_keys=ON")
Base.metadata.create_all(self.engine)
self.db = Session(self.engine)
self.owner = User(id=1, name="Owner", email="owner@example.test", hashed_password="unused")
self.peer = User(id=2, name="Peer", email="peer@example.test", hashed_password="unused")
self.mod = User(id=3, name="Mod", email="mod@example.test", hashed_password="unused", role="moderator")
self.db.add_all([self.owner, self.peer, self.mod])
self.db.flush()
self.course = Course(id=1, title="Course", user_id=3, status="published")
self.db.add(self.course)
self.db.add_all([QuestionCategory(id=1, name="Root", user_id=3),
QuestionCategory(id=2, name="Child", parent_id=1, user_id=3),
QuestionCategory(id=3, name="Leaf", parent_id=2, user_id=3),
QuestionCategory(id=4, name="Empty", user_id=3)])
self.db.add_all([Quiz(id=1, title="Origin", user_id=3, is_published=1),
Quiz(id=2, title="Course quiz", user_id=3, course_id=1, is_published=1, is_shared=1, allow_review=0)])
self.db.flush()
for qid, category, owner, shared, source in [
(1, 1, 3, 1, 1), (2, 2, 3, 1, 1), (3, 3, 1, 0, None),
(4, 2, 2, 0, None), (5, 2, 3, 1, 2), (6, None, 3, 1, None),
]:
self.db.add(Question(id=qid, question_category_id=category, user_id=owner,
is_shared=shared, source_quiz_id=source, question_text=f"Question {qid}",
question_type="mcq", options=["yes", "no"], correct_answer="yes",
explanation="Full explanation", image_path="q.png", explanation_image_path="answer.png"))
self.db.flush()
self.db.query(Question).filter(Question.id == 2).update({"is_shared": None})
self.db.add_all([QuizQuestionLink(quiz_id=1, question_id=1, position=0),
QuizQuestionLink(quiz_id=1, question_id=2, position=1),
QuizQuestionLink(quiz_id=2, question_id=5, position=0)])
self.db.commit()
self.user = self.owner
app = FastAPI()
for path, router in [("questions", questions.router), ("question-categories", question_categories.router),
("quizzes", quizzes.router), ("attempts", attempts.router), ("mobile", mobile.router)]:
app.include_router(router, prefix=f"/{path}")
app.dependency_overrides[get_db] = lambda: self.db
app.dependency_overrides[get_current_user] = lambda: self.user
self.client = TestClient(app)
def tearDown(self):
self.client.close()
self.db.close()
self.engine.dispose()
def generate(self, **overrides):
return self.client.post("/questions/builder", json={"title": " Test ", "count": 2, **overrides})
def count(self, **params):
response = self.client.get("/questions/builder/count", params=params)
self.assertEqual(response.status_code, 200, response.text)
return response.json()["count"]
def answer(self, qid, correct=False, day=0, completed=True, expired=0, quiz_id=1):
attempt = QuizAttempt(user_id=1, quiz_id=quiz_id, completed_at=datetime(2026, 1, 1) + timedelta(days=day) if completed else None,
expired=expired, total_questions=1, score=int(correct))
self.db.add(attempt)
self.db.flush()
self.db.add(AttemptAnswer(attempt_id=attempt.id, question_id=qid, is_correct=correct, user_answer="yes" if correct else "no"))
self.db.commit()
return attempt
def test_descendants_counts_visibility_and_invalid_categories(self):
self.assertEqual(category_descendants(self.db.query(QuestionCategory).all(), [1, 2]), {1, 2, 3})
self.assertEqual(self.count(category_ids=[1, 2]), 3)
self.assertEqual(self.count(category_ids=[1], is_shared=True), 2)
self.assertEqual(self.count(), 4)
cats = self.client.get("/question-categories/").json()
root = next(c for c in cats if c["id"] == 1)
leaf = next(c for c in cats if c["id"] == 3)
self.assertEqual(root["question_count"], 3)
self.assertEqual([b["id"] for b in leaf["breadcrumbs"]], [1, 2, 3])
self.assertEqual(self.client.get("/questions/builder/count?category_ids=999").status_code, 400)
self.assertEqual(self.generate(category_ids=[999]).status_code, 400)
self.assertEqual(self.client.get("/questions/bank/ids?category_ids=1,bad").status_code, 400)
self.assertEqual(self.client.get("/questions/bank?category_ids=-1").status_code, 400)
self.assertEqual(self.client.get("/questions/export/qti?question_ids=1,4").status_code, 400)
bank = self.client.get("/questions/bank", params={"category_ids": "1,2", "search_mode": "keyword"}).json()
self.assertEqual(bank["total"], 3)
self.assertEqual(set(self.client.get("/questions/bank/ids").json()), {1, 2, 3, 6})
def test_parent_cycle_missing_self_delete_and_saved_history(self):
saved = self.generate(category_ids=[1], count=3).json()["id"]
before = [q.question_id for q in self.db.query(QuizQuestionLink).filter_by(quiz_id=saved).order_by(QuizQuestionLink.position)]
attempt = self.answer(1, correct=True, quiz_id=saved)
self.user = self.mod
for parent in (1, 3, 999):
res = self.client.patch("/question-categories/1", json={"name": "Root", "parent_id": parent})
self.assertEqual(res.status_code, 400, res.text)
self.assertEqual(self.client.delete("/question-categories/1").status_code, 400)
self.assertEqual(self.client.delete("/question-categories/3?move_to=3").status_code, 400)
self.assertEqual(self.client.delete("/question-categories/3?move_to=999").status_code, 400)
res = self.client.patch("/question-categories/2", json={"name": "Moved", "parent_id": 4})
self.assertEqual(res.status_code, 200, res.text)
self.assertEqual([b["id"] for b in res.json()["breadcrumbs"]], [4, 2])
self.assertEqual(self.client.delete("/question-categories/3?move_to=4").status_code, 204)
self.assertEqual(before, [q.question_id for q in self.db.query(QuizQuestionLink).filter_by(quiz_id=saved).order_by(QuizQuestionLink.position)])
self.assertEqual(self.db.get(QuizAttempt, attempt.id).score, 1)
self.assertEqual(self.db.get(Question, 3).question_category_id, 4)
self.assertEqual(self.client.post("/question-categories/", json={"name": "New", "parent_id": 999}).status_code, 400)
def test_sampling_exact_zero_insufficient_stale_and_validation(self):
for payload, status in [({"category_ids": [4]}, 400), ({"count": 5}, 400), ({"expected_count": 99}, 409),
({"count": 0}, 422), ({"count": 201}, 422), ({"mode": "bad"}, 422),
({"title": " "}, 422), ({"title": "x" * 201}, 422), ({"time_limit_minutes": 0}, 422), ({"state": "bad"}, 422)]:
self.assertEqual(self.generate(**payload).status_code, status)
sampled = self.generate(count=2).json()
sampled_ids = [q.id for q in self.db.get(Quiz, sampled["id"]).questions]
self.assertEqual(len(sampled_ids), 2)
self.assertEqual(len(set(sampled_ids)), 2)
self.assertTrue(set(sampled_ids) <= {1, 2, 3, 6})
result = self.generate(count=4, expected_count=4, mode="learning")
self.assertEqual(result.status_code, 200, result.text)
quiz = self.db.get(Quiz, result.json()["id"])
self.assertEqual(quiz.title, "Test")
self.assertEqual(quiz.is_shared, 0)
self.assertEqual(quiz.is_published, 0)
self.assertEqual({q.id for q in quiz.questions}, {1, 2, 3, 6})
self.assertEqual(len(quiz.questions), 4)
self.assertEqual(quiz.questions[0].explanation_image_path, "answer.png")
first = self.client.post(f"/attempts/start?quiz_id={quiz.id}").json()
again = self.client.post(f"/attempts/start?quiz_id={quiz.id}").json()
self.assertEqual(first["id"], again["id"])
self.assertEqual({q["id"] for q in self.client.get(f"/quizzes/{quiz.id}").json()["questions"]}, {1, 2, 3, 6})
def test_latest_incorrect_unused_bookmarks(self):
self.answer(1, False, day=0)
self.answer(1, True, day=1)
self.answer(1, False, day=2, expired=1)
self.answer(2, False, day=0)
self.answer(2, True, day=2, completed=False)
self.answer(3, False, day=1, quiz_id=2) # course answers never count
self.db.add_all([Favorite(user_id=1, question_id=2), Favorite(user_id=1, question_id=5), Favorite(user_id=2, question_id=1)])
self.db.commit()
self.assertEqual(self.count(state="unused"), 2)
self.assertEqual(self.count(state="incorrect"), 1)
self.assertEqual(self.count(state="bookmarked"), 1)
res = self.generate(state="incorrect", count=1).json()
self.assertEqual([q.id for q in self.db.get(Quiz, res["id"]).questions], [2])
self.answer(2, True, day=3)
self.assertEqual(self.count(state="incorrect"), 0)
def test_explicit_ids_atomic_permissions_order_and_category_creator(self):
for ids in ([1, 4], [1, 5], [1, 999]):
before = self.db.query(Quiz).count()
res = self.client.post("/questions/from-bank", json={"title": "X", "question_ids": ids})
self.assertEqual(res.status_code, 400, res.text)
self.assertEqual(self.db.query(Quiz).count(), before)
res = self.client.post("/questions/from-bank", json={"title": "X", "question_ids": [2, 1, 2, 3]})
self.assertEqual(res.status_code, 200, res.text)
self.assertEqual([q.id for q in self.db.get(Quiz, res.json()["id"]).questions], [2, 1, 3])
res = self.client.post("/question-categories/3/create-quiz?title=Manual")
self.assertEqual(res.status_code, 200, res.text) # ordinary owner, no origin quiz
self.assertEqual(res.json()["questions_count"], 1)
def test_shared_private_revocation_and_mobile(self):
private_id = self.generate(category_ids=[3], count=1).json()["id"]
res = self.client.patch(f"/quizzes/{private_id}/share?shared=true")
self.assertEqual(res.status_code, 400)
self.assertEqual(self.client.post("/questions/from-bank", json={"title": "X", "question_ids": [3], "is_shared": True}).status_code, 400)
shared_id = self.generate(is_shared=True, count=2, category_ids=[1]).json()["id"]
self.user = self.peer
self.assertEqual(self.client.get(f"/quizzes/{private_id}").status_code, 403)
self.assertEqual(self.client.post(f"/quizzes/{private_id}/shuffle").status_code, 403)
self.assertEqual(self.client.post(f"/attempts/start?quiz_id={private_id}").status_code, 403)
self.assertEqual(self.client.get(f"/quizzes/{shared_id}").status_code, 200)
self.assertEqual(self.client.get(f"/mobile/quizzes/{shared_id}").status_code, 200)
self.assertNotIn(private_id, [q["id"] for q in self.client.get("/quizzes/").json()])
self.assertEqual(self.client.patch(f"/quizzes/{shared_id}/share?shared=false").status_code, 403)
attempt = self.client.post(f"/attempts/start?quiz_id={shared_id}").json()["id"]
self.user = self.owner
self.assertEqual(self.client.patch(f"/quizzes/{shared_id}/share?shared=false").status_code, 200)
self.user = self.peer
for url in (f"/quizzes/{shared_id}", f"/quizzes/{shared_id}/review", f"/attempts/progress?quiz_id={shared_id}",
f"/attempts/quiz/{shared_id}/in-progress", f"/attempts/{attempt}"):
self.assertEqual(self.client.get(url).status_code, 403, url)
self.assertEqual(self.client.get(f"/mobile/quizzes/{shared_id}").status_code, 404)
self.assertEqual(self.client.get("/attempts/in-progress").json(), [])
self.assertEqual(self.client.post("/attempts/progress", json={"quiz_id": shared_id,
"attempt_id": attempt, "answers": {}, "current_idx": 0, "mode": "timed"}).status_code, 403)
self.assertEqual(self.client.post("/mobile/attempts", json={"quiz_id": shared_id, "answers": []}).status_code, 404)
self.assertNotIn(shared_id, [q["id"] for q in self.client.get("/mobile/sync").json()["quizzes"]])
self.assertEqual(self.client.post(f"/attempts/{attempt}/submit", json={"answers": []}).status_code, 403)
self.assertEqual(self.client.post(f"/attempts/start?quiz_id={shared_id}").status_code, 403)
def test_question_revocation_legacy_publication_and_course_enrollment(self):
saved_id = self.generate(is_shared=True, category_ids=[1], count=2).json()["id"]
self.user = self.peer
self.assertEqual(self.client.get("/quizzes/1").status_code, 200) # published legacy quiz
self.db.get(Question, 1).is_shared = 0
self.db.commit()
self.assertEqual(self.client.get("/quizzes/1").status_code, 403)
self.user = self.owner
self.assertEqual(self.client.get(f"/quizzes/{saved_id}").status_code, 403)
self.user = self.peer
self.assertEqual(self.client.get("/quizzes/2").status_code, 403)
self.assertEqual(self.client.post("/attempts/start?quiz_id=2").status_code, 403)
self.db.add(CourseEnrollment(course_id=1, user_id=2))
self.db.commit()
self.assertEqual(self.client.get("/quizzes/2").status_code, 200)
self.assertEqual(self.client.get("/quizzes/2?study=true").status_code, 403)
self.assertEqual(self.client.get("/quizzes/2/review").status_code, 403)
self.assertEqual(self.client.get("/mobile/quizzes/2").status_code, 404)
self.assertNotIn(2, [q["id"] for q in self.client.get("/quizzes/").json()])
self.user = self.mod
self.assertEqual(self.client.patch("/quizzes/2/share?shared=true").status_code, 400)
self.assertEqual(self.client.get("/quizzes/1").status_code, 200)
def test_ownerless_question_revocation_denies_saved_owner(self):
self.db.get(Question, 1).user_id = None
self.db.commit()
saved = self.generate(is_shared=True, category_ids=[1], count=2).json()["id"]
self.db.get(Question, 1).is_shared = 0
self.db.commit()
for user in (self.owner, self.peer):
self.user = user
self.assertEqual(self.client.get(f"/quizzes/{saved}").status_code, 403)
self.assertEqual(self.client.post(f"/attempts/start?quiz_id={saved}").status_code, 403)
self.assertEqual(self.client.get(f"/mobile/quizzes/{saved}").status_code, 404)
self.assertNotIn(saved, [q["id"] for q in self.client.get("/quizzes/").json()])
def test_legacy_hide_revokes_both_flags(self):
self.db.get(Quiz, 1).is_shared = 1
self.db.commit()
self.user = self.mod
response = self.client.patch("/quizzes/1/publish?published=false")
self.assertEqual(response.status_code, 200)
self.assertEqual(response.json()["is_shared"], 0)
self.user = self.peer
self.assertNotIn(1, [q["id"] for q in self.client.get("/quizzes/").json()])
self.assertEqual(self.client.get("/quizzes/1").status_code, 403)
self.assertEqual(self.client.get("/mobile/quizzes/1").status_code, 404)
self.assertEqual(self.client.post("/attempts/start?quiz_id=1").status_code, 403)
def test_peer_cannot_delete_private_manual_question_or_history(self):
saved = self.generate(category_ids=[3], count=1).json()["id"]
attempt = self.answer(3, True, quiz_id=saved)
self.user = self.peer
self.assertEqual(self.client.delete("/questions/3").status_code, 403)
self.assertIsNotNone(self.db.get(Question, 3))
self.assertEqual(self.db.query(QuizQuestionLink).filter_by(quiz_id=saved, question_id=3).count(), 1)
self.assertEqual(self.db.query(AttemptAnswer).filter_by(attempt_id=attempt.id, question_id=3).count(), 1)
def test_real_submissions_record_skips_and_latest_outcome(self):
saved = self.generate(is_shared=True, category_ids=[1], count=2).json()["id"]
first = self.client.post(f"/attempts/start?quiz_id={saved}").json()["id"]
self.assertEqual(self.client.get(f"/attempts/{first}").json()["answers"], [])
redis = Mock()
with patch.dict(sys.modules, {"redis": redis}):
result = self.client.post(f"/attempts/{first}/submit", json={"answers": [{"question_id": 1, "user_answer": "yes"}]})
self.assertEqual(result.status_code, 200, result.text)
self.assertEqual((result.json()["score"], result.json()["total_questions"]), (1, 2))
self.assertEqual(self.db.query(AttemptAnswer).filter_by(attempt_id=first).count(), 2)
skipped = self.db.query(AttemptAnswer).filter_by(attempt_id=first, question_id=2).one()
self.assertEqual((skipped.user_answer, skipped.is_correct), ("", False))
self.assertEqual(self.count(state="incorrect"), 1)
self.assertEqual(self.count(state="unused"), 2)
second = self.client.post(f"/attempts/start?quiz_id={saved}&fresh=true").json()["id"]
with patch.dict(sys.modules, {"redis": redis}):
result = self.client.post(f"/attempts/{second}/submit", json={"answers": []})
self.assertEqual(result.status_code, 200, result.text)
self.assertEqual(result.json()["score"], 0)
self.assertEqual(self.count(state="incorrect"), 2)
self.assertEqual(self.count(state="unused"), 2)
def test_submission_rejects_duplicates_and_out_of_pool_atomically(self):
saved = self.generate(is_shared=True, category_ids=[1], count=2).json()["id"]
aid = self.client.post(f"/attempts/start?quiz_id={saved}").json()["id"]
attempt = self.db.get(QuizAttempt, aid)
attempt.selected_question_ids = [1]
attempt.total_questions = 1
self.db.commit()
for ids in ([1, 1], [2], [999]):
result = self.client.post(f"/attempts/{aid}/submit", json={"answers": [{"question_id": qid, "user_answer": "yes"} for qid in ids]})
self.assertEqual(result.status_code, 400, result.text)
self.assertEqual(self.db.query(AttemptAnswer).filter_by(attempt_id=aid).count(), 0)
self.assertIsNone(self.db.get(QuizAttempt, aid).completed_at)
with patch.dict(sys.modules, {"redis": Mock()}):
result = self.client.post(f"/attempts/{aid}/submit", json={"answers": [{"question_id": 1, "user_answer": "yes"}]})
self.assertEqual(result.status_code, 200, result.text)
self.assertEqual((result.json()["score"], result.json()["total_questions"], result.json()["percentage"]), (1, 1, 100))
self.assertEqual([q["question_id"] for q in result.json()["answers"]], [1])
self.assertEqual([q["question_id"] for q in self.client.get(f"/attempts/{aid}").json()["answers"]], [1])
def test_mobile_and_expiry_use_same_selected_question_grading(self):
import json
saved = self.generate(is_shared=True, category_ids=[1], count=2).json()["id"]
before = self.db.query(QuizAttempt).count()
for selected, answers in [([1, 1], []), ([999], []), ([1], [2]), ([1], [1, 1])]:
result = self.client.post("/mobile/attempts", json={"quiz_id": saved, "selected_question_ids": selected,
"answers": [{"question_id": qid, "user_answer": "yes"} for qid in answers]})
self.assertEqual(result.status_code, 400, result.text)
self.assertEqual(self.db.query(QuizAttempt).count(), before)
result = self.client.post("/mobile/attempts", json={"quiz_id": saved, "answers": [{"question_id": 1, "user_answer": "yes"}]})
self.assertEqual(result.status_code, 200, result.text)
self.assertEqual((result.json()["score"], result.json()["total_questions"]), (1, 2))
self.assertEqual(self.db.query(AttemptAnswer).filter_by(attempt_id=result.json()["id"]).count(), 2)
aid = self.client.post(f"/attempts/start?quiz_id={saved}&fresh=true").json()["id"]
attempt = self.db.get(QuizAttempt, aid)
attempt.selected_question_ids = [1]
attempt.total_questions = 1
self.db.commit()
redis = Mock()
redis.from_url.return_value.get.return_value = json.dumps({"answers": {}, "total_time": 1, "started_at": "2000-01-01T00:00:00+00:00"})
with patch.dict(sys.modules, {"redis": redis}):
response = self.client.get(f"/attempts/progress?quiz_id={saved}")
self.assertEqual(response.status_code, 200, response.text)
self.assertIsNone(response.json())
self.assertEqual(self.db.get(QuizAttempt, aid).expired, 1)
self.assertEqual(self.db.query(AttemptAnswer).filter_by(attempt_id=aid).count(), 1)
self.assertFalse(self.db.query(AttemptAnswer).filter_by(attempt_id=aid).one().is_correct)
if __name__ == "__main__":
unittest.main()