3279e14bb2
There will be no courses. What was there: one draft called "jk" with two empty lessons, and 4,000 lines of code around it — courses, modules, lessons, enrolments, per-lesson progress, SCORM, BigBlueButton, completion certificates, three React pages, a router, two models. Its real cost was everywhere else. Every query that measured practice had to remember `Quiz.course_id.is_(None)`, and forgetting it in one place would have silently mixed course attempts into a learner's analytics; the bank predicate carried a subquery to exclude a course's own questions from every search, recommendation and share; quiz access had a second, parallel rule about enrolment. All of that is gone, so the remaining rules say what they mean. `quizzes.allow_review` goes with it. It was only ever enforced for a course quiz, so it had become a promise nothing keeps — the public session page was still offering "no answer review" about sessions that review fine. The fixtures' question 5 lived in a course quiz and stood for "a question that exists but is not in your bank". There is no such thing now — a question is in the bank unless it is deleted — so the counts it kept out of the numbers are back in, and the tests that turned on it now turn on deletion or on the attempt that actually holds a question. Files the LMS uploaded stay on disk and stay protected: LEGACY_LMS_PREFIXES in app/utils/upload_access.py is what keeps them unreachable. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01TqXevQJhxFrM7jJg82cgZN
359 lines
18 KiB
Python
359 lines
18 KiB
Python
"""Quiz session management feed and question-manager bulk editing.
|
|
|
|
Run: DATABASE_URL=sqlite:///:memory: PYTHONPATH=backend python -m unittest discover -s backend/tests
|
|
No application startup, external services or AI calls; a disposable SQLite database per test.
|
|
"""
|
|
import os
|
|
os.environ["DATABASE_URL"] = "sqlite:///:memory:"
|
|
|
|
import json
|
|
import sys
|
|
import unittest
|
|
from datetime import datetime
|
|
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.attempt import AttemptAnswer, QuizAttempt
|
|
from app.models.question import Question
|
|
from app.models.question_category import QuestionCategory, QuestionCategoryLink
|
|
from app.models.quiz import Quiz
|
|
from app.models.quiz_category import QuizCategory
|
|
from app.models.quiz_question_link import QuizQuestionLink
|
|
from app.models.user import User
|
|
from app.models.article import Article
|
|
from app.routers import questions, study_tools
|
|
from app.utils.auth import get_current_user
|
|
|
|
with patch.dict(sys.modules, {"app.services.quiz_service": ModuleType("app.services.quiz_service")}):
|
|
from app.routers import quizzes
|
|
|
|
|
|
class SessionFeedTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.engine = create_engine("sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool)
|
|
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.mod = User(id=3, name="Mod", email="mod@example.test", hashed_password="unused", role="moderator")
|
|
self.db.add_all([self.owner, self.mod])
|
|
self.db.add(QuizCategory(id=7, name="Boards", user_id=1))
|
|
self.db.add_all([
|
|
Quiz(id=1, title="Fresh", user_id=1, mode="learning", questions_count=3, is_published=1,
|
|
created_at=datetime(2026, 1, 1)),
|
|
Quiz(id=2, title="Halfway", user_id=1, mode="timed", questions_count=4, is_published=1,
|
|
category_id=7, created_at=datetime(2026, 1, 2)),
|
|
Quiz(id=3, title="Finished", user_id=1, mode="timed", questions_count=2, is_published=1,
|
|
created_at=datetime(2026, 1, 3)),
|
|
])
|
|
self.db.flush()
|
|
self.db.add_all([
|
|
QuizAttempt(id=10, quiz_id=2, user_id=1, mode="exam", total_questions=4,
|
|
started_at=datetime(2026, 2, 1)),
|
|
QuizAttempt(id=11, quiz_id=3, user_id=1, mode="exam", total_questions=2, score=1,
|
|
started_at=datetime(2026, 1, 20), completed_at=datetime(2026, 1, 20)),
|
|
# An expired auto-submit never counts as a completed attempt.
|
|
QuizAttempt(id=12, quiz_id=1, user_id=1, mode="exam", total_questions=3, score=0, expired=1,
|
|
started_at=datetime(2026, 1, 21), completed_at=datetime(2026, 1, 21)),
|
|
])
|
|
self.db.commit()
|
|
|
|
self.user = self.owner
|
|
app = FastAPI()
|
|
app.include_router(quizzes.router, prefix="/quizzes")
|
|
app.include_router(questions.router, prefix="/questions")
|
|
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 sessions(self, progress=None):
|
|
"""Call the feed with Redis stubbed to the given attempt progress blobs."""
|
|
fake = Mock()
|
|
fake.mget.return_value = [json.dumps(progress)] if progress else [None]
|
|
redis_module = ModuleType("redis")
|
|
redis_module.from_url = lambda *a, **k: fake
|
|
with patch.dict(sys.modules, {"redis": redis_module}):
|
|
response = self.client.get("/quizzes/sessions")
|
|
self.assertEqual(response.status_code, 200, response.text)
|
|
return {row["title"]: row for row in response.json()}
|
|
|
|
def test_feed_reports_one_row_per_quiz_with_its_attempt_state(self):
|
|
rows = self.sessions(progress={"answers": {"1": "yes", "2": "no"}})
|
|
self.assertEqual(rows["Fresh"]["state"], "not_started")
|
|
self.assertEqual(rows["Fresh"]["attempts_count"], 0) # expired attempt excluded
|
|
self.assertIsNone(rows["Fresh"]["last_attempt_id"])
|
|
|
|
halfway = rows["Halfway"]
|
|
self.assertEqual(halfway["state"], "in_progress")
|
|
self.assertEqual(halfway["active_attempt_id"], 10)
|
|
self.assertEqual((halfway["answered"], halfway["total"]), (2, 4))
|
|
self.assertEqual(halfway["category_name"], "Boards")
|
|
|
|
finished = rows["Finished"]
|
|
self.assertEqual(finished["state"], "completed")
|
|
self.assertEqual(finished["last_attempt_id"], 11)
|
|
self.assertEqual(finished["last_percentage"], 50)
|
|
self.assertEqual(finished["attempts_count"], 1)
|
|
|
|
def test_feed_is_ordered_by_last_activity_and_survives_redis_failure(self):
|
|
redis_module = ModuleType("redis")
|
|
redis_module.from_url = Mock(side_effect=ConnectionError("redis down"))
|
|
with patch.dict(sys.modules, {"redis": redis_module}):
|
|
response = self.client.get("/quizzes/sessions")
|
|
self.assertEqual(response.status_code, 200, response.text)
|
|
rows = response.json()
|
|
# Newest activity first: live attempt (Feb 1) > completed (Jan 20) > created (Jan 1).
|
|
self.assertEqual([r["title"] for r in rows], ["Halfway", "Finished", "Fresh"])
|
|
self.assertEqual(rows[0]["answered"], 0) # degrades instead of failing
|
|
|
|
def test_feed_hides_another_users_unpublished_quiz(self):
|
|
self.db.add_all([
|
|
Quiz(id=5, title="Hidden of mod", user_id=3, is_published=0, is_shared=0,
|
|
questions_count=1, created_at=datetime(2026, 1, 5)),
|
|
Quiz(id=6, title="Published of mod", user_id=3, is_published=1, is_shared=0,
|
|
questions_count=1, created_at=datetime(2026, 1, 6)),
|
|
])
|
|
self.db.commit()
|
|
rows = self.sessions()
|
|
self.assertNotIn("Hidden of mod", rows)
|
|
self.assertFalse(rows["Published of mod"]["is_owner"]) # visible, but not manageable
|
|
|
|
|
|
class QuestionManagerTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.engine = create_engine("sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool)
|
|
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.mod = User(id=3, name="Mod", email="mod@example.test", hashed_password="unused", role="moderator")
|
|
self.db.add_all([self.owner, self.mod])
|
|
self.db.add_all([QuestionCategory(id=1, name="Cardiology", user_id=3),
|
|
QuestionCategory(id=2, name="Neonatology", user_id=3)])
|
|
self.db.flush()
|
|
# 1: complete · 2: no explanation · 3: uncategorized + no difficulty
|
|
for qid, category, difficulty, explanation in [
|
|
(1, 1, "hard", "Because…"),
|
|
(2, 1, "easy", ""),
|
|
(3, None, None, "Because…"),
|
|
(4, 2, "medium", "Because…"),
|
|
]:
|
|
self.db.add(Question(id=qid, question_category_id=category, difficulty=difficulty,
|
|
explanation=explanation, user_id=3,
|
|
question_text=f"Question {qid}", question_type="mcq",
|
|
options=["yes", "no"], correct_answer="yes"))
|
|
self.db.commit()
|
|
|
|
self.user = self.mod
|
|
app = FastAPI()
|
|
app.include_router(questions.router, prefix="/questions")
|
|
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 bank_ids(self, **params):
|
|
response = self.client.get("/questions/bank", params=params)
|
|
self.assertEqual(response.status_code, 200, response.text)
|
|
return {q["id"] for q in response.json()["questions"]}
|
|
|
|
def test_summary_counts_each_editorial_gap(self):
|
|
summary = self.client.get("/questions/manage/summary").json()
|
|
self.assertEqual(summary["total"], 4)
|
|
self.assertEqual(summary["uncategorized"], 1)
|
|
self.assertEqual(summary["no_explanation"], 1)
|
|
self.assertEqual(summary["no_difficulty"], 1)
|
|
|
|
def test_summary_is_moderator_only(self):
|
|
self.user = self.owner
|
|
self.assertEqual(self.client.get("/questions/manage/summary").status_code, 403)
|
|
|
|
def test_needs_filter_selects_the_matching_gap(self):
|
|
self.assertEqual(self.bank_ids(needs="category"), {3})
|
|
self.assertEqual(self.bank_ids(needs="explanation"), {2})
|
|
self.assertEqual(self.bank_ids(needs="difficulty"), {3})
|
|
self.assertEqual(self.client.get("/questions/bank", params={"needs": "nonsense"}).status_code, 422)
|
|
|
|
def test_bulk_sets_category_and_difficulty(self):
|
|
response = self.client.post("/questions/bulk",
|
|
json={"question_ids": [3], "action": "category", "category_id": 2})
|
|
self.assertEqual(response.json()["updated"], 1)
|
|
self.assertEqual(self.db.get(Question, 3).question_category_id, 2)
|
|
|
|
self.client.post("/questions/bulk", json={"question_ids": [2, 3], "action": "difficulty", "difficulty": "easy"})
|
|
self.assertEqual([self.db.get(Question, i).difficulty for i in (2, 3)], ["easy", "easy"])
|
|
|
|
# Sharing is no longer one of the actions: there is no per-question
|
|
# flag left for it to set.
|
|
self.assertEqual(self.client.post("/questions/bulk",
|
|
json={"question_ids": [4], "action": "share"}).status_code, 422)
|
|
|
|
def test_bulk_delete_removes_questions_and_their_category_links(self):
|
|
self.db.add(QuestionCategoryLink(question_id=1, category_id=2))
|
|
self.db.commit()
|
|
response = self.client.post("/questions/bulk", json={"question_ids": [1], "action": "delete"})
|
|
self.assertEqual(response.json()["updated"], 1)
|
|
self.assertIsNone(self.db.get(Question, 1))
|
|
self.assertEqual(self.db.query(QuestionCategoryLink).filter_by(question_id=1).count(), 0)
|
|
|
|
def test_bulk_rejects_bad_input_and_non_moderators(self):
|
|
self.assertEqual(self.client.post("/questions/bulk",
|
|
json={"question_ids": [], "action": "delete"}).status_code, 400)
|
|
self.assertEqual(self.client.post("/questions/bulk",
|
|
json={"question_ids": list(range(501)), "action": "delete"}).status_code, 400)
|
|
self.assertEqual(self.client.post("/questions/bulk",
|
|
json={"question_ids": [1], "action": "category", "category_id": 999}).status_code, 404)
|
|
self.assertEqual(self.client.post("/questions/bulk",
|
|
json={"question_ids": [1], "action": "banana"}).status_code, 422)
|
|
self.user = self.owner
|
|
self.assertEqual(self.client.post("/questions/bulk",
|
|
json={"question_ids": [1], "action": "delete"}).status_code, 403)
|
|
|
|
|
|
class RecommendationTests(unittest.TestCase):
|
|
"""Focus-area ranking: readiness shrinkage, relevance and roll-up through the tree."""
|
|
|
|
def setUp(self):
|
|
self.engine = create_engine("sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool)
|
|
Base.metadata.create_all(self.engine)
|
|
self.db = Session(self.engine)
|
|
|
|
self.user = User(id=1, name="Learner", email="learner@example.test", hashed_password="unused")
|
|
self.db.add_all([self.user, User(id=3, name="Mod", email="mod@example.test",
|
|
hashed_password="unused", role="moderator")])
|
|
self.db.add_all([
|
|
QuestionCategory(id=1, name="Cardiology", user_id=3),
|
|
QuestionCategory(id=2, name="Kawasaki disease", parent_id=1, user_id=3),
|
|
QuestionCategory(id=10, name="Neurology", user_id=3),
|
|
])
|
|
self.db.add(Quiz(id=1, title="Bank test", user_id=1, is_published=1, questions_count=6))
|
|
self.db.flush()
|
|
# 6 bank questions: 4 cardiology (all under the Kawasaki child), 2 neurology.
|
|
for qid, category in [(1, 2), (2, 2), (3, 2), (4, 2), (5, 10), (6, 10)]:
|
|
self.db.add(Question(id=qid, question_category_id=category, user_id=3,
|
|
question_text=f"Question {qid}", question_type="mcq",
|
|
options=["yes", "no"], correct_answer="yes"))
|
|
self.db.add(Article(id=7, slug="kawasaki", title="Kawasaki disease", sections=[],
|
|
category_id=1, user_id=3, status="published"))
|
|
self.db.commit()
|
|
|
|
app = FastAPI()
|
|
app.include_router(study_tools.router, prefix="/study-tools")
|
|
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 answer(self, question_id, correct):
|
|
attempt = self.db.query(QuizAttempt).filter_by(user_id=1, quiz_id=1).first()
|
|
if attempt is None:
|
|
attempt = QuizAttempt(quiz_id=1, user_id=1, mode="study", total_questions=6,
|
|
completed_at=datetime(2026, 3, 1))
|
|
self.db.add(attempt)
|
|
self.db.flush()
|
|
self.db.add(AttemptAnswer(attempt_id=attempt.id, question_id=question_id,
|
|
user_answer="yes" if correct else "no", is_correct=correct))
|
|
self.db.commit()
|
|
|
|
def recommend(self, **params):
|
|
response = self.client.get("/study-tools/recommendations", params=params)
|
|
self.assertEqual(response.status_code, 200, response.text)
|
|
return response.json()
|
|
|
|
def test_relevance_and_coverage_roll_up_to_the_discipline(self):
|
|
data = self.recommend()
|
|
self.assertEqual(data["group"], "disciplines")
|
|
rows = {row["name"]: row for row in data["focus_areas"]}
|
|
# Cardiology owns 4 of 6 bank questions through its child category.
|
|
self.assertEqual(rows["Cardiology"]["available"], 4)
|
|
self.assertEqual(rows["Cardiology"]["relevance"], 66.7)
|
|
self.assertEqual(rows["Neurology"]["available"], 2)
|
|
self.assertEqual(data["bank_total"], 6)
|
|
# Disciplines are the top of the tree; a condition under one is not a row.
|
|
self.assertNotIn("Kawasaki disease", rows)
|
|
|
|
def test_the_same_answers_group_three_ways(self):
|
|
# Articles: the reading to go back to, reached through its category.
|
|
articles = self.recommend(group="articles")["focus_areas"]
|
|
self.assertEqual([row["name"] for row in articles], ["Kawasaki disease"])
|
|
row = articles[0]
|
|
self.assertEqual((row["article_id"], row["category_id"]), (7, 1))
|
|
self.assertEqual(row["available"], 4)
|
|
|
|
# Systems: no tag table on a fresh database, so there is nothing to
|
|
# show — and that is reported as no rows, not as an error.
|
|
systems = self.recommend(group="systems")
|
|
self.assertEqual(systems["group"], "systems")
|
|
self.assertEqual(systems["focus_areas"], [])
|
|
|
|
def test_an_unknown_grouping_is_refused_rather_than_guessed(self):
|
|
self.assertEqual(
|
|
self.client.get("/study-tools/recommendations", params={"group": "subtopics"}).status_code,
|
|
422)
|
|
|
|
def test_readiness_stays_locked_until_enough_answers(self):
|
|
self.answer(1, False)
|
|
data = self.recommend()
|
|
self.assertFalse(data["unlocked"])
|
|
self.assertEqual(data["answers_needed"], 39)
|
|
self.assertEqual(data["overall_accuracy"], 0.0) # reported, but the client gates on `unlocked`
|
|
cardiology = next(r for r in data["focus_areas"] if r["name"] == "Cardiology")
|
|
self.assertIsNone(cardiology["readiness"])
|
|
self.assertEqual(cardiology["accuracy"], 0.0) # raw accuracy still reported
|
|
self.assertEqual(cardiology["status"], "focus")
|
|
|
|
def test_readiness_shrinks_a_small_sample_toward_overall_accuracy(self):
|
|
# 40 answers unlocks readiness: 38 correct in neurology, 2 wrong in cardiology.
|
|
for _ in range(19):
|
|
self.answer(5, True)
|
|
self.answer(6, True)
|
|
self.answer(1, False)
|
|
self.answer(2, False)
|
|
data = self.recommend()
|
|
self.assertTrue(data["unlocked"])
|
|
rows = {row["name"]: row for row in data["focus_areas"]}
|
|
cardiology = rows["Cardiology"]
|
|
self.assertEqual(cardiology["accuracy"], 0.0)
|
|
# Two wrong answers alone must not read as 0% knowledge; shrinkage pulls it up
|
|
# toward the learner's ~95% overall, but it still ranks below neurology.
|
|
self.assertGreater(cardiology["readiness"], 0.0)
|
|
self.assertLess(cardiology["readiness"], rows["Neurology"]["readiness"])
|
|
self.assertGreater(cardiology["priority"], rows["Neurology"]["priority"])
|
|
self.assertEqual(data["focus_areas"][0]["name"], "Cardiology")
|
|
|
|
def test_untouched_categories_report_no_data_and_link_their_article(self):
|
|
data = self.recommend()
|
|
cardiology = next(r for r in data["focus_areas"] if r["name"] == "Cardiology")
|
|
self.assertEqual(cardiology["status"], "no_data")
|
|
self.assertEqual(cardiology["answered"], 0)
|
|
self.assertEqual(cardiology["article_id"], 7)
|
|
self.assertEqual(cardiology["article_title"], "Kawasaki disease")
|
|
self.assertFalse(cardiology["is_focus_area"]) # focus areas need recorded answers
|
|
|
|
def test_grouping_is_validated(self):
|
|
self.assertEqual(self.client.get("/study-tools/recommendations",
|
|
params={"group": "nonsense"}).status_code, 422)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|