"""The prepared session: the arithmetic, and whether the plan tells the truth. The test that matters here is the last one. A plan that does not describe the session it starts is worse than no plan, so the account handed to the learner is checked against the questions actually put in the quiz — not against the ranking that produced them, which would only prove the code agrees with itself. """ import os os.environ.setdefault("DATABASE_URL", "sqlite:///:memory:") import sys import unittest from datetime import datetime, timedelta from decimal import Decimal from types import ModuleType from unittest.mock import 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 import test_quiz_builder # noqa: F401 — imports every model, so the metadata resolves from app.database import Base, get_db from app.models.attempt import AttemptAnswer, QuizAttempt from app.models.exam import BlueprintCategoryLink, Exam, ExamBlueprint, QuestionExamLink 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.user import User from app.services import prepared_session from app.services.quiz_builder import (DUE_RECALL, EVIDENCE_HALF_LIFE_DAYS, MAX_REVIEW_SHARE, NEUTRAL_RECALL, CandidateRanking, recall_probability, recency_weight) 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 questions #: Cardiology is filed a level down, which is where the plan's grouping has to #: reach up from; the other two disciplines hold their questions directly. CARDIOLOGY, CONGENITAL, RHEUMATOLOGY, DERMATOLOGY = 1, 2, 3, 4 class DecayTests(unittest.TestCase): """No database: the curve on its own.""" def test_evidence_halves_once_a_half_life(self): self.assertAlmostEqual(recency_weight(0), 1.0) self.assertAlmostEqual(recency_weight(EVIDENCE_HALF_LIFE_DAYS), 0.5) self.assertAlmostEqual(recency_weight(3 * EVIDENCE_HALF_LIFE_DAYS), 0.125) # An answer dated in the future is a clock disagreement, not evidence # worth more than a fresh one. self.assertAlmostEqual(recency_weight(-40), 1.0) def test_both_outcomes_decay_towards_a_coin_flip_rather_than_past_it(self): fresh_hit, old_hit = recall_probability(True, 0), recall_probability(True, 365) fresh_miss, old_miss = recall_probability(False, 0), recall_probability(False, 365) self.assertGreater(fresh_hit, old_hit) self.assertLess(fresh_miss, old_miss) # Time turns a right answer into "no idea", never into a wrong one. self.assertAlmostEqual(old_hit, NEUTRAL_RECALL, places=2) self.assertAlmostEqual(old_miss, NEUTRAL_RECALL, places=2) self.assertGreater(old_hit, NEUTRAL_RECALL) self.assertLess(old_miss, NEUTRAL_RECALL) def test_a_recent_miss_outranks_an_older_one(self): # The ranking is worth (1 − recall), so lower recall is picked first. self.assertLess(recall_probability(False, 7), recall_probability(False, 90)) def test_a_correct_answer_comes_back_round_once_it_is_old_enough(self): self.assertGreater(recall_probability(True, 3), DUE_RECALL) self.assertLess(recall_probability(True, 90), DUE_RECALL) # And a wrong answer is due from the moment it is given. self.assertLess(recall_probability(False, 0), DUE_RECALL) class Bank(unittest.TestCase): """Thirty questions over three disciplines, weighted 60 / 10 / 30.""" NOW = datetime(2026, 6, 1) 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.db.add(Exam(id=1, slug="boards", name="Pediatrics Boards", is_active=1)) self.user = User(id=1, name="Learner", email="l@example.test", hashed_password="unused", active_exam_id=1) self.db.add(self.user) self.db.add_all([ QuestionCategory(id=CARDIOLOGY, name="Cardiology", user_id=1), QuestionCategory(id=CONGENITAL, name="Congenital", parent_id=CARDIOLOGY, user_id=1), QuestionCategory(id=RHEUMATOLOGY, name="Rheumatology", user_id=1), QuestionCategory(id=DERMATOLOGY, name="Dermatology", user_id=1), ]) self.db.add(Quiz(id=1, title="Origin", user_id=1, is_published=1)) self.db.flush() self.of_category = {} for qid in range(1, 31): category = CONGENITAL if qid <= 12 else RHEUMATOLOGY if qid <= 21 else DERMATOLOGY self.of_category[qid] = category self.db.add(Question(id=qid, question_category_id=category, user_id=1, question_text=f"Q{qid}", question_type="mcq", options=["a", "b"], correct_answer="a")) self.db.add(QuestionExamLink(question_id=qid, exam_id=1)) self.db.add_all([ ExamBlueprint(id=1, exam_id=1, code="1", title="Cardiology", weight=Decimal("60")), ExamBlueprint(id=2, exam_id=1, code="2", title="Rheumatology", weight=Decimal("10")), ExamBlueprint(id=3, exam_id=1, code="3", title="Dermatology", weight=Decimal("30")), ]) self.db.flush() self.db.add_all([BlueprintCategoryLink(blueprint_id=1, category_id=CARDIOLOGY), BlueprintCategoryLink(blueprint_id=2, category_id=RHEUMATOLOGY), BlueprintCategoryLink(blueprint_id=3, category_id=DERMATOLOGY)]) self.db.commit() 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 sitting(self, marks, days_ago=1, when=None): """One finished attempt. `marks` is {question id: was it right}.""" finished = (when or self.NOW) - timedelta(days=days_ago) attempt = QuizAttempt(user_id=1, quiz_id=1, completed_at=finished, expired=0, total_questions=len(marks), score=sum(1 for ok in marks.values() if ok)) self.db.add(attempt) self.db.flush() for question_id, was_correct in marks.items(): self.db.add(AttemptAnswer(attempt_id=attempt.id, question_id=question_id, is_correct=was_correct, user_answer="a" if was_correct else "b")) self.db.commit() return attempt def ranking(self, now=None): return CandidateRanking(self.db, self.user, now=now or self.NOW) def plan(self, count=None, now=None): return prepared_session.prepare_session(self.db, self.user, count, now or self.NOW) class SelectionTests(Bank): def test_a_miss_last_week_is_recycled_before_a_miss_in_the_spring(self): # Same discipline, so the topic damping and the blueprint weight are # identical and only the dates can separate them. self.sitting({1: False}, days_ago=120) self.sitting({2: False}, days_ago=7) # Three questions leaves one review slot; the recent miss should take it. picked = self.ranking().select(3) self.assertIn(2, picked) self.assertNotIn(1, picked) def test_a_topic_answered_right_once_does_not_count_as_known_forever(self): self.sitting({qid: True for qid in range(1, 13)}, days_ago=1) fresh = self.ranking() # Answered, recently, and correctly: nothing to review here yet. self.assertEqual([row for row in fresh.due if row[0] <= 12], []) stale = self.ranking(now=self.NOW + timedelta(days=180)) self.assertEqual(len(stale.due), 12) # And the topic's accuracy has drifted back towards unknown rather than # staying at the hundred percent one good day bought it. self.assertGreater(fresh.accuracy(CONGENITAL), 0.8) self.assertLess(stale.accuracy(CONGENITAL), 0.6) def test_review_takes_its_share_and_no_more(self): self.sitting({qid: False for qid in range(1, 13)}, days_ago=3) picked = self.ranking().select(10) seen = [qid for qid in picked if qid <= 12] self.assertEqual(len(seen), round(MAX_REVIEW_SHARE * 10)) self.assertEqual(len(picked), 10) def test_nothing_left_unseen_still_fills_the_session(self): self.sitting({qid: qid % 2 == 0 for qid in range(1, 31)}, days_ago=3) picked = self.ranking().select(12) self.assertEqual(len(picked), 12) self.assertEqual(len(set(picked)), 12) def test_selection_is_reproducible(self): self.sitting({1: False, 13: True, 25: False}, days_ago=10) self.assertEqual(self.ranking().select(15), self.ranking().select(15)) class LengthTests(Bank): def test_the_default_holds_until_enough_sessions_are_finished(self): self.sitting({qid: True for qid in range(1, 6)}, days_ago=20) self.sitting({qid: True for qid in range(6, 11)}, days_ago=15) self.assertEqual(prepared_session.habitual_length(self.db, self.user), (None, 2)) self.assertEqual(self.plan()["count"], prepared_session.DEFAULT_SESSION_LENGTH) def test_length_becomes_the_learner_s_own_median_finished_session(self): # Eight, eight, thirty: the median is eight and the mean is fifteen, # which is a length this learner has never once sat. self.sitting({qid: True for qid in range(1, 9)}, days_ago=30) self.sitting({qid: True for qid in range(9, 17)}, days_ago=20) self.sitting({qid: True for qid in range(1, 31)}, days_ago=10) self.assertEqual(prepared_session.habitual_length(self.db, self.user), (8, 3)) plan = self.plan() self.assertEqual(plan["count"], 8) self.assertIn("You usually finish 8 questions", plan["length_reason"]) def test_an_unfinished_or_expired_sitting_is_not_a_length(self): attempt = self.sitting({qid: True for qid in range(1, 4)}, days_ago=5) attempt.expired = 1 self.db.commit() self.assertEqual(prepared_session.habitual_length(self.db, self.user), (None, 0)) def test_the_offered_length_never_exceeds_what_is_left(self): for qid in range(6, 31): self.db.delete(self.db.get(Question, qid)) self.db.commit() plan = self.plan() self.assertEqual(plan["count"], 5) self.assertIn("everything left in your bank", plan["length_reason"]) def test_a_length_the_learner_asks_for_is_honored_and_said_to_be_theirs(self): plan = self.plan(count=7) self.assertEqual(plan["count"], 7) self.assertEqual(plan["length_reason"], "7 questions, your choice.") class PlanTests(Bank): def rows(self, plan): return {row["name"]: row for row in plan["topics"]} def test_a_learner_with_no_history_is_told_so_and_gets_the_blueprint_spread(self): plan = self.plan(count=10) self.assertEqual(plan["basis"], "cold_start") self.assertIn("haven't finished a session yet", plan["summary"]) self.assertIn("exam blueprint", plan["summary"]) self.assertEqual(plan["review_count"], 0) rows = self.rows(plan) # 60 / 10 / 30 of ten questions, which is what the board publishes. self.assertEqual(rows["Cardiology"]["count"], 6) self.assertEqual(rows["Rheumatology"]["count"], 1) self.assertEqual(rows["Dermatology"]["count"], 3) # And the reason says the blueprint, not a weakness nobody has measured. self.assertEqual(rows["Cardiology"]["reason"], "Worth 60% of the exam") for row in plan["topics"]: self.assertIsNone(row["accuracy"]) def test_a_learner_with_one_topic_answered_is_told_which_and_how_badly(self): self.sitting({qid: False for qid in range(1, 9)}, days_ago=4) plan = self.plan(count=10) self.assertEqual(plan["basis"], "personalized") rows = self.rows(plan) self.assertRegex(rows["Cardiology"]["reason"], r"^Weak area — \d+% correct so far$") self.assertLess(rows["Cardiology"]["accuracy"], 30) # Cardiology is both the weakest and the heaviest, so it leads — but the # damping keeps the other two disciplines in the session. self.assertEqual(plan["topics"][0]["name"], "Cardiology") self.assertGreater(len(plan["topics"]), 1) # Nothing has been measured about dermatology, and the plan says so # rather than inventing a figure for it. self.assertIsNone(rows["Dermatology"]["accuracy"]) self.assertEqual(rows["Dermatology"]["reason"], "Not attempted yet, worth 30% of the exam") def test_a_topic_last_seen_months_ago_is_named_as_due_not_as_a_weakness(self): # Everything right, but long enough ago that it is worth checking. The # prior would drag the accuracy under the weak threshold; the evidence # gate is what stops "you got all of these right" reading as "weak". self.sitting({qid: True for qid in range(13, 22)}, days_ago=150) self.sitting({qid: True for qid in range(22, 25)}, days_ago=2) self.sitting({qid: True for qid in range(25, 28)}, days_ago=1) rows = self.rows(self.plan(count=12)) self.assertIn("Rheumatology", rows) self.assertRegex(rows["Rheumatology"]["reason"], r"^Due for review — last answered \d+ months ago$") def test_the_counts_in_the_plan_add_up_to_the_session(self): self.sitting({qid: qid % 3 == 0 for qid in range(1, 16)}, days_ago=45) plan = self.plan(count=14) self.assertEqual(sum(row["count"] for row in plan["topics"]), plan["count"]) self.assertEqual(plan["new_count"] + plan["review_count"], plan["count"]) for row in plan["topics"]: self.assertEqual(row["new_count"] + row["review_count"], row["count"]) class CommittedPlanTests(Bank): """The plan has to describe the session it starts.""" def start(self, **body): response = self.client.post("/questions/builder/prepared", json=body) self.assertEqual(response.status_code, 200, response.text) return response.json() def questions_in(self, quiz_id): return [row[0] for row in self.db.query(QuizQuestionLink.question_id).filter( QuizQuestionLink.quiz_id == quiz_id).order_by(QuizQuestionLink.position).all()] def test_the_plan_describes_the_questions_actually_put_in_the_session(self): self.sitting({qid: qid % 4 == 0 for qid in range(1, 19)}, days_ago=6) self.sitting({qid: True for qid in range(19, 25)}, days_ago=200) self.sitting({25: False, 26: False}, days_ago=2) created = self.start(count=16, title="Prepared") plan = created["plan"] asked = self.questions_in(created["id"]) self.assertEqual(len(asked), 16) self.assertEqual(created["questions_count"], 16) self.assertEqual(plan["count"], 16) # Every claim the plan makes, recounted from the questions themselves. by_discipline = {} for question_id in asked: leaf = self.of_category[question_id] top = CARDIOLOGY if leaf == CONGENITAL else leaf by_discipline[top] = by_discipline.get(top, 0) + 1 named = {CARDIOLOGY: "Cardiology", RHEUMATOLOGY: "Rheumatology", DERMATOLOGY: "Dermatology"} self.assertEqual({row["name"]: row["count"] for row in plan["topics"]}, {named[key]: value for key, value in by_discipline.items()}) seen_before = {row[0] for row in self.db.query(AttemptAnswer.question_id).all()} self.assertEqual(plan["review_count"], len([q for q in asked if q in seen_before])) self.assertEqual(plan["new_count"], len([q for q in asked if q not in seen_before])) for row in plan["topics"]: top = next(key for key, name in named.items() if name == row["name"]) here = [q for q in asked if (CARDIOLOGY if self.of_category[q] == CONGENITAL else self.of_category[q]) == top] self.assertEqual(row["review_count"], len([q for q in here if q in seen_before])) self.assertEqual(row["new_count"], len([q for q in here if q not in seen_before])) # And every topic named gives its one reason. for row in plan["topics"]: self.assertTrue(row["reason"]) def test_the_preview_says_what_starting_it_will_give(self): self.sitting({qid: qid % 2 == 0 for qid in range(1, 20)}, days_ago=9) preview = self.client.get("/questions/builder/prepared", params={"count": 12}) self.assertEqual(preview.status_code, 200, preview.text) forecast = preview.json() # The preview is for reading; it must not hand out the stems' ids. self.assertNotIn("question_ids", forecast) started = self.start(count=12)["plan"] self.assertEqual(forecast, started) def test_the_preview_writes_nothing(self): before = self.db.query(Quiz).count() self.client.get("/questions/builder/prepared") self.assertEqual(self.db.query(Quiz).count(), before) def test_changing_the_length_replans_rather_than_padding_the_old_plan(self): self.sitting({qid: False for qid in range(1, 10)}, days_ago=5) short, long = self.start(count=6)["plan"], self.start(count=24)["plan"] self.assertEqual(short["count"], 6) self.assertEqual(long["count"], 24) self.assertEqual(sum(row["count"] for row in long["topics"]), 24) def test_without_a_study_objective_it_refuses_rather_than_guessing(self): self.user.active_exam_id = None self.db.commit() for call in (self.client.get("/questions/builder/prepared"), self.client.post("/questions/builder/prepared", json={})): self.assertEqual(call.status_code, 400, call.text) self.assertIn("studying for", call.json()["detail"]) if __name__ == "__main__": unittest.main()