"""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.course import Course 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(Course(id=1, title="Course", user_id=3, status="published")) 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)), # Course quizzes stay out of the main management feed. Quiz(id=4, title="Course quiz", user_id=1, course_id=1, is_published=1, questions_count=2), ]) 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.assertNotIn("Course quiz", rows) 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()