1f52c1217f
1,881 files: a Capacitor shell for Android and iOS, a Forgejo workflow that
signed and uploaded an APK, and `/api/mobile` — the API that shell talked to.
Nothing in the web app called it, and the PWA (manifest, service worker, icon)
is untouched: installing PedsHub on a phone works exactly as before.
`/api/mobile` was worth removing on its own account. `GET /api/mobile/sync`
returned, for every quiz a learner could see, every question in it *with its
correct answer and explanation* — "bulk sync with answer keys for offline
study". It is the same hole that was just closed on the question bank, still
open through a second door, and `GET /api/mobile/quizzes/{id}` was a third. It
also carried a second password-login endpoint with its own rate limiter, which
is twice the sign-in surface for an app that no longer exists.
The tests that reached those endpoints were asserting quiz visibility and
grading, not the transport. They assert the same things through /quizzes and
/attempts now, so nothing is less covered than it was.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01TqXevQJhxFrM7jJg82cgZN
628 lines
34 KiB
Python
628 lines
34 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.services.quiz_builder import category_descendants
|
||
from app.utils.auth import get_current_user
|
||
from app.routers import questions, question_categories, attempts
|
||
# 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.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="Second origin", user_id=3, is_published=1, is_shared=1)])
|
||
self.db.flush()
|
||
# Per-question sharing is gone: every question in the bank is in the
|
||
# bank, and who may manage one is the grant tree's business.
|
||
for qid, category, owner, source in [
|
||
(1, 1, 3, 1), (2, 2, 3, 1), (3, 3, 1, None),
|
||
(4, 2, 2, None), (5, 2, 3, 2), (6, None, 3, None),
|
||
]:
|
||
self.db.add(Question(id=qid, question_category_id=category, user_id=owner,
|
||
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.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)]:
|
||
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 saved_test(self, ids, **overrides):
|
||
"""A test built from named questions, for the tests that care which."""
|
||
response = self.client.post("/questions/from-bank",
|
||
json={"title": "Named", "question_ids": ids, **overrides})
|
||
self.assertEqual(response.status_code, 200, response.text)
|
||
return response.json()["id"]
|
||
|
||
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_adaptive_selection_prefers_unanswered_then_recycles_weakest(self):
|
||
from app.services.quiz_builder import adaptive_select
|
||
self.answer(1, False, day=0) # Question 1 (cat 1): incorrect — recycled first.
|
||
self.answer(2, True, day=1) # Question 2 (cat 2): correct.
|
||
ids = adaptive_select(self.db, self.owner, 5, [], 'all', None)
|
||
self.assertEqual(set(ids[:3]), {3, 4, 6}) # Unanswered questions preferred.
|
||
self.assertEqual(ids[3], 1) # Older incorrect resurfaces first.
|
||
self.assertEqual(ids[4], 2)
|
||
|
||
def test_difficulty_tags_filter_and_validate(self):
|
||
self.user = self.mod
|
||
self.assertEqual(self.client.patch('/questions/1', json={'difficulty': 'banana'}).status_code, 422)
|
||
response = self.client.patch('/questions/1', json={'difficulty': 'hard'})
|
||
self.assertEqual(response.status_code, 200, response.text)
|
||
bank = self.client.get('/questions/bank', params={'q': 'Question 1'}).json()
|
||
self.assertEqual(bank['questions'][0]['difficulty'], 'hard')
|
||
self.user = self.owner
|
||
self.assertEqual(self.client.get('/questions/builder/count', params={'difficulty': 'hard', 'state': 'all'}).json()['count'], 1)
|
||
self.assertEqual(self.client.get('/questions/builder/count', params={'difficulty': 'easy', 'state': 'all'}).json()['count'], 0)
|
||
|
||
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]), 5)
|
||
self.assertEqual(self.count(), 6)
|
||
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"], 5)
|
||
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,999").status_code, 400)
|
||
bank = self.client.get("/questions/bank", params={"category_ids": "1,2", "search_mode": "keyword"}).json()
|
||
self.assertEqual(bank["total"], 5)
|
||
self.assertEqual(set(self.client.get("/questions/bank/ids").json()), {1, 2, 3, 4, 5, 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": 7}, 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, 4, 5, 6})
|
||
result = self.generate(count=6, expected_count=6, 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, 4, 5, 6})
|
||
self.assertEqual(len(quiz.questions), 6)
|
||
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, 4, 5, 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)
|
||
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"), 3)
|
||
self.assertEqual(self.count(state="incorrect"), 2)
|
||
self.assertEqual(self.count(state="bookmarked"), 2)
|
||
res = self.generate(state="incorrect", count=2).json()
|
||
self.assertEqual(sorted(q.id for q in self.db.get(Quiz, res["id"]).questions), [2, 3])
|
||
self.answer(2, True, day=3)
|
||
self.answer(3, True, day=3)
|
||
self.assertEqual(self.count(state="incorrect"), 0)
|
||
|
||
def test_explicit_ids_atomic_permissions_order_and_category_creator(self):
|
||
# One id that cannot be used spoils the whole request, whether it is
|
||
# missing or removed — the session is never made half-way.
|
||
self.db.get(Question, 5).deleted_at = datetime(2026, 1, 1)
|
||
self.db.commit()
|
||
for ids in ([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(self):
|
||
private_id = self.generate(category_ids=[3], count=1).json()["id"]
|
||
# A deleted question is out of everyone's reach, so a session holding
|
||
# one cannot be handed on. Per-question sharing used to be the other
|
||
# reason; there is no such flag now.
|
||
self.db.get(Question, 3).deleted_at = datetime(2026, 1, 1)
|
||
self.db.commit()
|
||
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.saved_test([1, 2], is_shared=True)
|
||
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.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("/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(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_and_legacy_publication(self):
|
||
saved_id = self.saved_test([1, 2], is_shared=True)
|
||
self.user = self.peer
|
||
self.assertEqual(self.client.get("/quizzes/1").status_code, 200) # published legacy quiz
|
||
# Deleting a question revokes every session that carried it. That is
|
||
# the whole of per-question revocation now: the flag an author could
|
||
# set is gone, and a question either exists or it does not.
|
||
self.db.get(Question, 1).deleted_at = datetime(2026, 1, 1)
|
||
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.user = self.mod
|
||
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.saved_test([1, 2], is_shared=True)
|
||
self.db.get(Question, 1).deleted_at = datetime(2026, 1, 1)
|
||
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.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.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.saved_test([1, 2], is_shared=True)
|
||
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"), 4)
|
||
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"), 4)
|
||
|
||
def test_submission_rejects_duplicates_and_out_of_pool_atomically(self):
|
||
saved = self.saved_test([1, 2], is_shared=True)
|
||
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_an_expired_attempt_is_graded_over_the_questions_it_selected(self):
|
||
import json
|
||
saved = self.saved_test([1, 2], is_shared=True)
|
||
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)
|
||
# Out of time, and handed back unmarked: resuming shows the player its
|
||
# own saved answers so it can open on Time's Up. Nothing is graded by a
|
||
# request the learner did not make.
|
||
self.assertNotIn("expired_submitted", response.json())
|
||
self.assertIsNone(self.db.get(QuizAttempt, aid).completed_at)
|
||
# And when the learner does submit, it is still graded over the
|
||
# selected questions only — the parity this test is named for.
|
||
submitted = self.client.post(f"/attempts/{aid}/submit", json={"answers": []})
|
||
self.assertEqual(submitted.status_code, 200, submitted.text)
|
||
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()
|
||
|
||
|
||
class AdaptiveSelectionTests(unittest.TestCase):
|
||
"""What the adaptive session actually picks, and in what order."""
|
||
|
||
def setUp(self):
|
||
self.bank = BuilderTests()
|
||
self.bank.setUp()
|
||
self.db = self.bank.db
|
||
self.user = self.bank.owner
|
||
|
||
def tearDown(self):
|
||
self.bank.tearDown()
|
||
|
||
def test_unseen_material_leads_and_review_takes_only_its_share(self):
|
||
from app.services.quiz_builder import MAX_REVIEW_SHARE, adaptive_select
|
||
# Answer one question in category 1 wrongly, one in category 2 rightly.
|
||
self.bank.answer(1, correct=False)
|
||
self.bank.answer(5, correct=True)
|
||
picked = adaptive_select(self.db, self.user, 2, [], "all", None)
|
||
self.assertEqual(len(picked), 2)
|
||
# Unseen material still leads, but "unseen first, always" is not what
|
||
# this does any more: it meant a learner with three thousand unanswered
|
||
# questions never saw a repeat, which is no spaced repetition at all.
|
||
# Review gets its share of the session and no more — one slot in two.
|
||
seen = [qid for qid in picked if qid in (1, 5)]
|
||
self.assertEqual(len(seen), round(MAX_REVIEW_SHARE * 2))
|
||
# And the slot goes to the miss, not to the one that went well.
|
||
self.assertEqual(seen, [1])
|
||
|
||
def test_one_wrong_answer_does_not_settle_a_topic(self):
|
||
"""The shrinkage, pinned.
|
||
|
||
A topic with a single miss used to sort first: raw accuracy reads 0%,
|
||
and 0% is the strongest signal there is. It is also the least evidence
|
||
there is. `accuracy()` pulls a topic towards NEUTRAL_RECALL by
|
||
PRIOR_ANSWERS, so one answer moves it a little and twenty move it a
|
||
lot — which is the whole reason the recommendations page shows
|
||
readiness rather than accuracy, and this is the same arithmetic.
|
||
"""
|
||
from app.services.quiz_builder import (
|
||
CandidateRanking, NEUTRAL_RECALL, PRIOR_ANSWERS)
|
||
|
||
# Read as of the day the answer was given, so recency decay is not part
|
||
# of what is being measured here — the fixture dates its attempts to
|
||
# 2026-01-01, and eight months of decay would flatten the answer to
|
||
# almost nothing before the shrinkage ever saw it.
|
||
as_of = datetime(2026, 1, 1)
|
||
ranking = lambda: CandidateRanking(self.db, self.user, now=as_of)
|
||
|
||
self.assertAlmostEqual(ranking().accuracy(1), NEUTRAL_RECALL, places=6)
|
||
|
||
self.bank.answer(1, correct=False)
|
||
after_one = ranking().accuracy(1)
|
||
# Moved, but nowhere near the 0.0 that raw accuracy would report.
|
||
self.assertLess(after_one, NEUTRAL_RECALL)
|
||
self.assertGreater(after_one, 0.25)
|
||
# And exactly where the formula says: (0 + prior × neutral) / (1 + prior).
|
||
self.assertAlmostEqual(
|
||
after_one, (PRIOR_ANSWERS * NEUTRAL_RECALL) / (1 + PRIOR_ANSWERS), places=6)
|
||
|
||
# Impact follows accuracy, so one miss cannot send a topic to the top of
|
||
# the queue the way a raw 0% would.
|
||
self.assertLess(ranking().impact(1), ranking().weight(1))
|
||
|
||
def test_the_level_follows_the_learner_topic_by_topic(self):
|
||
"""The half of "adaptive" that was missing until the bank had labels.
|
||
|
||
Difficulty was a filter a learner could set and nothing the session did
|
||
on its own, so somebody at 30% on a topic and somebody at 90% were
|
||
asked the same questions in the same order.
|
||
"""
|
||
from app.services.quiz_builder import (
|
||
CandidateRanking, DIFFICULTY_FIT, UNRATED_FIT, target_difficulty)
|
||
|
||
# Knowing nothing about somebody is the middle, not the bottom: the
|
||
# easiest questions are a poor way to find out what they know.
|
||
self.assertEqual(target_difficulty(None), 'medium')
|
||
self.assertEqual(target_difficulty(0.30), 'easy')
|
||
self.assertEqual(target_difficulty(0.65), 'medium')
|
||
self.assertEqual(target_difficulty(0.95), 'hard')
|
||
|
||
for qid, level in [(1, 'easy'), (2, 'medium'), (4, 'hard')]:
|
||
self.db.get(Question, qid).difficulty = level
|
||
self.db.commit()
|
||
|
||
as_of = datetime(2026, 1, 1)
|
||
ranking = CandidateRanking(self.db, self.user, now=as_of)
|
||
# Nothing answered anywhere, so every topic is at neutral readiness and
|
||
# every topic wants medium.
|
||
self.assertEqual(ranking.difficulty_fit((2, 2)), DIFFICULTY_FIT[0])
|
||
self.assertEqual(ranking.difficulty_fit((1, 1)), DIFFICULTY_FIT[1])
|
||
self.assertEqual(ranking.difficulty_fit((4, 2)), DIFFICULTY_FIT[1])
|
||
# An unlabelled question is unknown, not wrong.
|
||
self.assertEqual(ranking.difficulty_fit((3, 3)), UNRATED_FIT)
|
||
|
||
# It is a preference, never a filter: a bank thinned to one level is a
|
||
# bank three times smaller, and on a narrow topic that is the same
|
||
# eight questions every time.
|
||
self.assertGreater(min(DIFFICULTY_FIT), 0)
|
||
picked = self.bank.client.post('/questions/builder', json={
|
||
'title': 'Adaptive', 'count': 5, 'state': 'all', 'mode': 'learning',
|
||
'algorithm': 'adaptive', 'category_ids': [], 'expected_count': 5,
|
||
})
|
||
self.assertEqual(picked.status_code, 200, picked.text)
|
||
|
||
def test_it_can_see_the_whole_bank_not_the_first_page_of_it(self):
|
||
from app.services.quiz_builder import adaptive_select, bank_query
|
||
# The bank this learner may actually see — not every row in the table.
|
||
every = {row.id for row in bank_query(self.db, self.user).all()}
|
||
picked = set(adaptive_select(self.db, self.user, len(every), [], "all", None))
|
||
self.assertEqual(picked, every)
|
||
|
||
def test_a_question_answered_wrongly_returns_before_one_answered_rightly(self):
|
||
from app.services.quiz_builder import adaptive_select
|
||
for qid in [row.id for row in self.db.query(Question).all()]:
|
||
self.bank.answer(qid, correct=(qid != 1))
|
||
# Everything has been seen, so the whole selection is recycled.
|
||
picked = adaptive_select(self.db, self.user, 1, [], "all", None)
|
||
self.assertEqual(picked, [1])
|
||
|
||
|
||
class ExamClockTests(unittest.TestCase):
|
||
"""A timed block's length is a property of the exam, not a preference."""
|
||
|
||
def test_ninety_seconds_a_question_rounded_up_to_the_minute(self):
|
||
from app.services.quiz_builder import exam_minutes
|
||
|
||
class Ask:
|
||
mode = "timed"
|
||
time_limit_minutes = None
|
||
|
||
# Forty questions is an hour, which is the pace a real paper is sat at.
|
||
self.assertEqual(exam_minutes(Ask(), 40), 60)
|
||
self.assertEqual(exam_minutes(Ask(), 20), 30)
|
||
# Five is seven and a half minutes, and a block never gets less than a
|
||
# minute however short it is.
|
||
self.assertEqual(exam_minutes(Ask(), 5), 8)
|
||
self.assertEqual(exam_minutes(Ask(), 0), 1)
|
||
|
||
def test_a_study_block_has_no_clock(self):
|
||
from app.services.quiz_builder import exam_minutes
|
||
|
||
class Ask:
|
||
mode = "study"
|
||
time_limit_minutes = 30
|
||
|
||
self.assertIsNone(exam_minutes(Ask(), 40))
|
||
|
||
def test_an_explicit_limit_is_still_honoured(self):
|
||
from app.services.quiz_builder import exam_minutes
|
||
|
||
class Ask:
|
||
mode = "timed"
|
||
time_limit_minutes = 15
|
||
|
||
self.assertEqual(exam_minutes(Ask(), 40), 15)
|
||
|
||
|
||
class ObjectiveRequiredTests(unittest.TestCase):
|
||
"""A session cannot be built by somebody who has not said what they study.
|
||
|
||
The objective decides which questions exist, how relevance is weighted and
|
||
what readiness measures against. The interface asks; this is the same rule
|
||
where it cannot be walked past.
|
||
"""
|
||
|
||
def setUp(self):
|
||
self.bank = BuilderTests()
|
||
self.bank.setUp()
|
||
self.client = self.bank.client
|
||
self.db = self.bank.db
|
||
|
||
def tearDown(self):
|
||
self.bank.tearDown()
|
||
|
||
def offer(self):
|
||
from app.models.exam import Exam
|
||
self.db.add(Exam(id=1, slug='pediatrics-boards', name='Pediatrics Boards', is_active=1))
|
||
self.db.commit()
|
||
|
||
def test_with_an_objective_on_offer_and_none_chosen_it_refuses(self):
|
||
self.offer()
|
||
response = self.bank.generate(is_shared=True, category_ids=[1])
|
||
self.assertEqual(response.status_code, 400, response.text)
|
||
self.assertIn('studying for', response.json()['detail'])
|
||
|
||
def test_choosing_one_lets_it_through(self):
|
||
self.offer()
|
||
self.bank.owner.active_exam_id = 1
|
||
self.db.commit()
|
||
self.assertEqual(self.bank.generate(is_shared=True, category_ids=[1]).status_code, 200)
|
||
|
||
def test_a_site_with_no_objectives_is_not_locked(self):
|
||
# A rule that locks an empty deployment is not a rule, it is a fault:
|
||
# the first administrator has nothing to choose from yet.
|
||
self.assertEqual(self.bank.generate(is_shared=True, category_ids=[1]).status_code, 200)
|
||
|
||
|
||
class ExamScopeTests(unittest.TestCase):
|
||
"""The pool a session is drawn from is the pool the bank shows.
|
||
|
||
These disagreed, and in the worst direction: a learner whose exam had no
|
||
content linked to it browsed an empty bank and was then handed a session
|
||
built from every question in it.
|
||
"""
|
||
|
||
def setUp(self):
|
||
self.bank = BuilderTests()
|
||
self.bank.setUp()
|
||
self.client = self.bank.client
|
||
self.db = self.bank.db
|
||
self.bank.user = self.bank.owner
|
||
|
||
def tearDown(self):
|
||
self.bank.tearDown()
|
||
|
||
def link(self, exam_id, *question_ids):
|
||
from app.models.exam import Exam, QuestionExamLink
|
||
if not self.db.get(Exam, exam_id):
|
||
self.db.add(Exam(id=exam_id, name=f"Exam {exam_id}", slug=f"exam-{exam_id}"))
|
||
self.db.flush()
|
||
for qid in question_ids:
|
||
self.db.add(QuestionExamLink(exam_id=exam_id, question_id=qid))
|
||
self.db.commit()
|
||
|
||
def test_an_exam_with_no_content_yields_no_session_rather_than_all_of_it(self):
|
||
# Every question filed against exam 1, as the live bank is, so nothing
|
||
# is unclassified and nothing falls through to exam 2 that way.
|
||
self.link(1, 1, 2, 3, 4, 5, 6)
|
||
self.link(2) # exists, nothing linked to it
|
||
self.bank.owner.active_exam_id = 2
|
||
self.db.commit()
|
||
self.assertEqual(self.bank.count(), 0)
|
||
self.assertEqual(self.bank.generate(count=2).status_code, 400)
|
||
|
||
def test_the_count_and_the_session_agree_on_the_pool(self):
|
||
self.link(1, 1, 2)
|
||
self.link(2, 3, 4, 5, 6)
|
||
self.bank.owner.active_exam_id = 1
|
||
self.db.commit()
|
||
available = self.bank.count()
|
||
self.assertEqual(available, 2)
|
||
result = self.bank.generate(count=available, expected_count=available)
|
||
self.assertEqual(result.status_code, 200, result.text)
|
||
quiz = self.db.get(Quiz, result.json()["id"])
|
||
self.assertEqual({q.id for q in quiz.questions}, {1, 2})
|
||
|
||
def test_unclassified_content_still_reaches_everyone(self):
|
||
# A question linked to no exam is unclassified, not excluded — that is
|
||
# the rule that keeps new content visible before anybody files it.
|
||
self.link(1, 1)
|
||
self.bank.owner.active_exam_id = 1
|
||
self.db.commit()
|
||
self.assertGreater(self.bank.count(), 1)
|