6abe3cca11
The card offered Review answers and Resume session at once on a session still in progress, which is the muddle: there is nothing to review yet and nothing to resume once it is done. It is one or the other now, and what decides it is whether anything is left to answer — not whether it was an exam or a study session, which have the same two states as each other. A study session keeps going until every question is answered and becomes the review at that point, without waiting to be handed in. Repeat is offered either way. The questions worth sitting again are worth sitting again now. "Skipped" meant gone past, and was shown for questions in a session still running that had not been reached. Those read "not yet answered". And a timed block is now ninety seconds a question, set from the count rather than asked for. Choosing a limit is a decision nobody has the information to make — the pace belongs to the exam being rehearsed, not to a preference — and a block sat at the wrong pace teaches the wrong pace. Forty questions is an hour. An explicit limit is still honoured. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01TqXevQJhxFrM7jJg82cgZN
364 lines
19 KiB
Python
364 lines
19 KiB
Python
"""What happens to a session when its clock runs out, when it is deleted, and
|
|
when a learner resets everything.
|
|
|
|
Disposable SQLite; Redis is a Mock. The rule under test for expiry: an exam
|
|
closed without being suspended keeps running, and when time is up it is
|
|
submitted with what was answered and the score counts.
|
|
"""
|
|
import json
|
|
import sys
|
|
import unittest
|
|
from datetime import datetime
|
|
from types import ModuleType
|
|
from unittest.mock import Mock, patch
|
|
|
|
import test_quiz_builder as fixtures
|
|
from app.models.attempt import AttemptAnswer, QuizAttempt
|
|
from app.models.favorite import Favorite
|
|
from app.models.quiz import Quiz
|
|
from app.models.study_plan import StudyPlan, StudyPlanBlock, StudyPlanBlockProgress
|
|
from app.models.user_note import QuestionNote
|
|
from app.routers import attempts, study_plans
|
|
from app.services.attempt_expiry import seconds_remaining
|
|
|
|
with patch.dict(sys.modules, {"app.services.quiz_service": ModuleType("app.services.quiz_service")}):
|
|
from app.routers import quizzes
|
|
|
|
LONG_AGO = "2000-01-01T00:00:00+00:00"
|
|
|
|
|
|
def fake_redis(store):
|
|
"""Just enough of a Redis client: get / setex / delete / mget / scan_iter."""
|
|
client = Mock()
|
|
client.get.side_effect = lambda k: store.get(k)
|
|
client.setex.side_effect = lambda k, ttl, v: store.__setitem__(k, v)
|
|
client.mget.side_effect = lambda keys: [store.get(k) for k in keys]
|
|
client.scan_iter.side_effect = lambda pattern: [k for k in list(store) if k.startswith(pattern.rstrip("*"))]
|
|
|
|
def delete(*keys):
|
|
n = sum(1 for k in keys if store.pop(k, None) is not None)
|
|
return n
|
|
client.delete.side_effect = delete
|
|
module = Mock()
|
|
module.from_url.return_value = client
|
|
return module
|
|
|
|
|
|
class SessionLifecycleTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.bank = fixtures.BuilderTests()
|
|
self.bank.setUp()
|
|
self.client = self.bank.client
|
|
self.db = self.bank.db
|
|
for router, prefix in [(quizzes.router, "/quizzes"), (attempts.router, "/attempts"),
|
|
(study_plans.router, "/study-plans")]:
|
|
self.client.app.include_router(router, prefix=prefix)
|
|
self.bank.user = self.bank.owner
|
|
self.store = {}
|
|
self.redis = fake_redis(self.store)
|
|
|
|
def tearDown(self):
|
|
self.bank.tearDown()
|
|
|
|
def timed_quiz(self):
|
|
quiz_id = self.bank.generate(is_shared=True, category_ids=[1], count=2, mode="timed").json()["id"]
|
|
with patch.dict(sys.modules, {"redis": self.redis}):
|
|
aid = self.client.post(f"/attempts/start?quiz_id={quiz_id}&fresh=true").json()["id"]
|
|
return quiz_id, aid
|
|
|
|
def save(self, uid, aid, **extra):
|
|
blob = {"answers": {"1": "yes"}, "total_time": 600, "started_at": LONG_AGO, **extra}
|
|
self.store[f"quiz_progress:{uid}:{aid}"] = json.dumps(blob)
|
|
|
|
# ── expiry ────────────────────────────────────────────────────────────────
|
|
|
|
def test_a_suspended_exam_never_expires(self):
|
|
self.assertIsNone(seconds_remaining({"total_time": 1, "started_at": LONG_AGO, "suspended": True}))
|
|
self.assertIsNone(seconds_remaining({"answers": {}}))
|
|
self.assertLess(seconds_remaining({"total_time": 1, "started_at": LONG_AGO}), 0)
|
|
|
|
def test_a_closed_exam_spends_no_time_while_it_is_closed(self):
|
|
"""The crux: what is left is what the player last saved.
|
|
|
|
It used to be computed from `started_at` and `total_time`, so an
|
|
afternoon with the tab closed spent an afternoon of the exam on
|
|
questions nobody was shown. An exam not on screen is not being sat.
|
|
"""
|
|
# Started long ago by the wall clock, but with eight minutes still on
|
|
# it — because the clock only ran while it was open.
|
|
self.assertEqual(
|
|
seconds_remaining({"total_time": 600, "started_at": LONG_AGO, "time_left": 480}),
|
|
480.0)
|
|
|
|
def test_opening_and_closing_still_spends_the_time_it_is_open_for(self):
|
|
"""And eventually runs out, which is the other half of the rule."""
|
|
left = 600.0
|
|
for spent in (200, 200, 200):
|
|
# Each sitting saves what is left when it ends.
|
|
left = max(0.0, left - spent)
|
|
self.assertEqual(
|
|
seconds_remaining({"total_time": 600, "started_at": LONG_AGO, "time_left": left}),
|
|
left)
|
|
self.assertEqual(left, 0.0)
|
|
|
|
def test_an_exam_closed_at_zero_is_submitted_on_the_next_look(self):
|
|
"""If the tab goes before the submit lands, the next page settles it."""
|
|
quiz_id, aid = self.timed_quiz()
|
|
self.save(self.bank.owner.id, aid, time_left=0)
|
|
with patch.dict(sys.modules, {"redis": self.redis}):
|
|
rows = {row["quiz_id"]: row for row in self.client.get("/quizzes/sessions").json()}
|
|
self.assertEqual(rows[quiz_id]["state"], "completed")
|
|
self.assertIsNotNone(self.db.get(QuizAttempt, aid).completed_at)
|
|
|
|
def test_an_exam_with_time_on_it_is_left_alone(self):
|
|
quiz_id, aid = self.timed_quiz()
|
|
self.save(self.bank.owner.id, aid, time_left=90)
|
|
with patch.dict(sys.modules, {"redis": self.redis}):
|
|
rows = {row["quiz_id"]: row for row in self.client.get("/quizzes/sessions").json()}
|
|
# Still in progress, however long ago it was started.
|
|
self.assertNotEqual(rows[quiz_id]["state"], "completed")
|
|
self.assertIsNone(self.db.get(QuizAttempt, aid).completed_at)
|
|
|
|
def test_an_unsuspended_exam_that_ran_out_is_a_finished_exam_in_the_session_list(self):
|
|
quiz_id, aid = self.timed_quiz()
|
|
self.save(self.bank.owner.id, aid)
|
|
# A second one, also out of time: settling the first must not disturb
|
|
# the pass over the rest (it once mutated the dict being iterated).
|
|
other_quiz, other_aid = self.timed_quiz()
|
|
self.save(self.bank.owner.id, other_aid)
|
|
with patch.dict(sys.modules, {"redis": self.redis}):
|
|
rows = {row["quiz_id"]: row for row in self.client.get("/quizzes/sessions").json()}
|
|
self.assertEqual(rows[other_quiz]["state"], "completed")
|
|
row = rows[quiz_id]
|
|
self.assertEqual(row["state"], "completed")
|
|
self.assertEqual(row["last_attempt_id"], aid)
|
|
self.assertEqual(row["attempts_count"], 1)
|
|
attempt = self.db.get(QuizAttempt, aid)
|
|
self.assertIsNotNone(attempt.completed_at)
|
|
self.assertFalse(attempt.expired)
|
|
# Graded on what was answered; the unanswered question counts against.
|
|
self.assertEqual(attempt.total_questions, 2)
|
|
self.assertNotIn(f"quiz_progress:{self.bank.owner.id}:{aid}", self.store)
|
|
|
|
def test_resuming_an_expired_exam_reports_the_submission(self):
|
|
quiz_id, aid = self.timed_quiz()
|
|
self.save(self.bank.owner.id, aid)
|
|
with patch.dict(sys.modules, {"redis": self.redis}):
|
|
response = self.client.get(f"/attempts/progress?quiz_id={quiz_id}")
|
|
self.assertEqual(response.json(), {"expired_submitted": True, "attempt_id": aid, "quiz_id": quiz_id})
|
|
# And the analysis exists for it straight away.
|
|
self.assertEqual(self.client.get(f"/attempts/{aid}/analysis").status_code, 200)
|
|
|
|
def test_a_suspended_exam_resumes_with_its_clock_held(self):
|
|
quiz_id, aid = self.timed_quiz()
|
|
self.save(self.bank.owner.id, aid, suspended=True, time_left=120)
|
|
with patch.dict(sys.modules, {"redis": self.redis}):
|
|
saved = self.client.get(f"/attempts/progress?quiz_id={quiz_id}").json()
|
|
self.assertEqual(saved["total_time"], 120)
|
|
self.assertFalse(saved["suspended"])
|
|
self.assertIsNone(self.db.get(QuizAttempt, aid).completed_at)
|
|
|
|
# ── delete ────────────────────────────────────────────────────────────────
|
|
|
|
def test_deleting_the_attempt_that_completed_a_block_makes_the_block_unfinished(self):
|
|
self.db.add(StudyPlan(id=1, slug="p", name="Plan", kind="set", is_published=1))
|
|
self.db.flush()
|
|
self.db.add(StudyPlanBlock(id=10, plan_id=1, position=1, title="Block 1", question_ids=[1, 2]))
|
|
self.db.commit()
|
|
quiz_id = self.client.post("/study-plans/blocks/10/start", params={"mode": "timed"}).json()["id"]
|
|
self.db.add(QuizAttempt(id=800, quiz_id=quiz_id, user_id=self.bank.owner.id, mode="exam",
|
|
total_questions=2, score=2, started_at=datetime(2026, 1, 1),
|
|
completed_at=datetime(2026, 1, 1, 1)))
|
|
self.db.add(AttemptAnswer(attempt_id=800, question_id=1, user_answer="A", is_correct=True))
|
|
progress = self.db.query(StudyPlanBlockProgress).filter_by(quiz_id=quiz_id).one()
|
|
progress.completed_at = datetime(2026, 1, 1, 1)
|
|
self.db.commit()
|
|
self.store[f"quiz_progress:{self.bank.owner.id}:800"] = "{}"
|
|
|
|
with patch.dict(sys.modules, {"redis": self.redis}):
|
|
self.assertEqual(self.client.delete("/attempts/800").status_code, 204)
|
|
self.db.expire_all()
|
|
self.assertIsNone(self.db.get(QuizAttempt, 800))
|
|
self.assertEqual(self.db.query(AttemptAnswer).filter_by(attempt_id=800).count(), 0)
|
|
self.assertIsNone(self.db.get(StudyPlanBlockProgress, progress.id).completed_at)
|
|
self.assertEqual(self.db.get(StudyPlanBlockProgress, progress.id).quiz_id, quiz_id)
|
|
self.assertNotIn(f"quiz_progress:{self.bank.owner.id}:800", self.store)
|
|
|
|
def test_someone_elses_attempt_cannot_be_deleted(self):
|
|
quiz_id, aid = self.timed_quiz()
|
|
self.bank.user = self.bank.peer
|
|
with patch.dict(sys.modules, {"redis": self.redis}):
|
|
self.assertEqual(self.client.delete(f"/attempts/{aid}").status_code, 404)
|
|
|
|
# ── reset ─────────────────────────────────────────────────────────────────
|
|
|
|
def test_reset_removes_what_was_practised_and_keeps_what_was_made(self):
|
|
quiz_id, aid = self.timed_quiz()
|
|
uid = self.bank.owner.id
|
|
self.db.add_all([
|
|
AttemptAnswer(attempt_id=aid, question_id=1, user_answer="A", is_correct=True),
|
|
Favorite(user_id=uid, question_id=1),
|
|
QuestionNote(user_id=uid, question_id=1, content="remember this"),
|
|
])
|
|
# Another learner's data is untouched.
|
|
self.db.add(QuizAttempt(id=901, quiz_id=quiz_id, user_id=self.bank.peer.id, mode="exam",
|
|
total_questions=2, score=1, started_at=datetime(2026, 1, 1),
|
|
completed_at=datetime(2026, 1, 1)))
|
|
self.db.commit()
|
|
self.store[f"quiz_progress:{uid}:{aid}"] = "{}"
|
|
self.store[f"quiz_active:{uid}:{aid}"] = "dev"
|
|
self.store[f"quiz_progress:{self.bank.peer.id}:901"] = "{}"
|
|
|
|
with patch.dict(sys.modules, {"redis": self.redis}):
|
|
refused = self.client.post("/attempts/reset-all", json={"confirm": "yes"})
|
|
self.assertEqual(refused.status_code, 400)
|
|
response = self.client.post("/attempts/reset-all", json={"confirm": "reset"})
|
|
self.assertEqual(response.status_code, 200, response.text)
|
|
removed = response.json()["removed"]
|
|
self.assertEqual((removed["attempts"], removed["answers"], removed["saved_questions"],
|
|
removed["question_notes"], removed["in_progress"]), (1, 1, 1, 1, 2))
|
|
|
|
self.db.expire_all()
|
|
self.assertEqual(self.db.query(QuizAttempt).filter_by(user_id=uid).count(), 0)
|
|
self.assertIsNotNone(self.db.get(QuizAttempt, 901))
|
|
self.assertIn(f"quiz_progress:{self.bank.peer.id}:901", self.store)
|
|
# The test they built is still theirs.
|
|
self.assertIsNotNone(self.db.get(Quiz, quiz_id))
|
|
self.assertEqual(self.db.query(Favorite).filter_by(user_id=uid).count(), 0)
|
|
self.assertEqual(self.db.query(QuestionNote).filter_by(user_id=uid).count(), 0)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|
|
|
|
|
|
class QuizAnalysisTests(unittest.TestCase):
|
|
"""A session addressed by quiz: the same page at zero."""
|
|
|
|
def setUp(self):
|
|
self.bank = fixtures.BuilderTests()
|
|
self.bank.setUp()
|
|
self.client = self.bank.client
|
|
self.db = self.bank.db
|
|
self.client.app.include_router(attempts.router, prefix="/attempts")
|
|
self.bank.user = self.bank.owner
|
|
|
|
def tearDown(self):
|
|
self.bank.tearDown()
|
|
|
|
def test_a_quiz_nobody_has_sat_answers_with_zeroes_and_every_question_skipped(self):
|
|
quiz_id = self.bank.generate(is_shared=True, category_ids=[1], count=2).json()["id"]
|
|
body = self.client.get(f"/attempts/quiz/{quiz_id}/analysis").json()
|
|
self.assertTrue(body["not_started"])
|
|
self.assertIsNone(body["attempt_id"])
|
|
self.assertEqual((body["answered"], body["score"], body["percent"]), (0, 0, 0))
|
|
self.assertEqual(body["total"], len(body["questions"]))
|
|
self.assertTrue(body["questions"])
|
|
self.assertEqual({q["status"] for q in body["questions"]}, {"skipped"})
|
|
self.assertEqual([q["position"] for q in body["questions"]],
|
|
list(range(1, len(body["questions"]) + 1)))
|
|
self.assertEqual(body["recommendations"], [])
|
|
|
|
def test_once_sat_the_quiz_address_answers_with_the_real_analysis(self):
|
|
quiz_id = self.bank.generate(is_shared=True, category_ids=[1], count=2).json()["id"]
|
|
self.db.add(QuizAttempt(id=910, quiz_id=quiz_id, user_id=self.bank.owner.id, mode="learning",
|
|
total_questions=1, score=1, started_at=datetime(2026, 5, 1),
|
|
completed_at=datetime(2026, 5, 1, 1)))
|
|
self.db.add(AttemptAnswer(attempt_id=910, question_id=1, user_answer="A", is_correct=True))
|
|
self.db.commit()
|
|
body = self.client.get(f"/attempts/quiz/{quiz_id}/analysis").json()
|
|
self.assertFalse(body["not_started"])
|
|
self.assertEqual(body["attempt_id"], 910)
|
|
self.assertEqual(body["percent"], 100)
|
|
|
|
def test_a_quiz_the_learner_cannot_see_is_refused(self):
|
|
quiz_id = self.bank.generate(is_shared=False, category_ids=[1], count=2).json()["id"]
|
|
self.bank.user = self.bank.peer
|
|
self.assertEqual(self.client.get(f"/attempts/quiz/{quiz_id}/analysis").status_code, 403)
|
|
|
|
|
|
class LiveAnalysisTests(unittest.TestCase):
|
|
"""A session can be analysed while it is still being sat."""
|
|
|
|
def setUp(self):
|
|
self.bank = fixtures.BuilderTests()
|
|
self.bank.setUp()
|
|
self.client = self.bank.client
|
|
self.db = self.bank.db
|
|
self.client.app.include_router(attempts.router, prefix="/attempts")
|
|
self.bank.user = self.bank.owner
|
|
self.store = {}
|
|
self.redis = fake_redis(self.store)
|
|
|
|
def tearDown(self):
|
|
self.bank.tearDown()
|
|
|
|
def test_answers_given_but_not_submitted_are_analysed(self):
|
|
quiz_id = self.bank.generate(is_shared=True, category_ids=[1], count=2).json()["id"]
|
|
with patch.dict(sys.modules, {"redis": self.redis}):
|
|
aid = self.client.post(f"/attempts/start?quiz_id={quiz_id}&mode=study").json()["id"]
|
|
# One answered, saved to progress; nothing submitted.
|
|
self.store[f"quiz_progress:{self.bank.owner.id}:{aid}"] = json.dumps({"answers": {"1": "yes"}})
|
|
|
|
with patch.dict(sys.modules, {"redis": self.redis}):
|
|
body = self.client.get(f"/attempts/{aid}/analysis").json()
|
|
|
|
# It used to report 0/0 with an empty table while the session list,
|
|
# which reads the same saved progress, said one had been answered.
|
|
self.assertEqual(body["answered"], 1)
|
|
self.assertGreater(body["total"], 1)
|
|
self.assertEqual(body["score"], 1)
|
|
statuses = {q["status"] for q in body["questions"]}
|
|
self.assertIn("correct", statuses)
|
|
self.assertIn("unanswered", statuses)
|
|
# Nothing was written: submitting is what records answers.
|
|
self.assertEqual(self.db.query(AttemptAnswer).filter_by(attempt_id=aid).count(), 0)
|
|
|
|
def test_an_exam_still_running_is_not_marked(self):
|
|
"""The integrity rule: a running exam reports progress, never a score.
|
|
|
|
Grading it live would let a learner answer, open the analysis to see
|
|
whether it was right, and go back and change it — the exam defeated
|
|
rather than analysed.
|
|
"""
|
|
quiz_id = self.bank.generate(is_shared=True, category_ids=[1], count=2).json()["id"]
|
|
with patch.dict(sys.modules, {"redis": self.redis}):
|
|
aid = self.client.post(f"/attempts/start?quiz_id={quiz_id}&mode=exam").json()["id"]
|
|
self.store[f"quiz_progress:{self.bank.owner.id}:{aid}"] = json.dumps({"answers": {"1": "yes"}})
|
|
|
|
with patch.dict(sys.modules, {"redis": self.redis}):
|
|
body = self.client.get(f"/attempts/{aid}/analysis").json()
|
|
|
|
self.assertFalse(body["graded"])
|
|
self.assertIsNone(body["score"])
|
|
self.assertIsNone(body["percent"])
|
|
# How far through, which is not a leak, and nothing about rightness.
|
|
self.assertEqual(body["answered"], 1)
|
|
statuses = {q["status"] for q in body["questions"]}
|
|
# "Skipped" would mean gone past; in a session still running it has
|
|
# not been reached.
|
|
self.assertEqual(statuses, {"answered", "unanswered"})
|
|
self.assertNotIn("correct", statuses)
|
|
self.assertNotIn("incorrect", statuses)
|
|
# And nothing to go back to yet: a recommendation is a verdict.
|
|
self.assertEqual(body["recommendations"], [])
|
|
|
|
def test_a_finished_exam_is_marked(self):
|
|
quiz_id = self.bank.generate(is_shared=True, category_ids=[1], count=2).json()["id"]
|
|
with patch.dict(sys.modules, {"redis": self.redis}):
|
|
aid = self.client.post(f"/attempts/start?quiz_id={quiz_id}&mode=exam").json()["id"]
|
|
self.client.post(f"/attempts/{aid}/submit",
|
|
json={"answers": [{"question_id": 1, "user_answer": "yes"}]})
|
|
body = self.client.get(f"/attempts/{aid}/analysis").json()
|
|
# Once it is over the withholding stops: this is the "converts to
|
|
# study" moment, and it is what the clock running out does too.
|
|
self.assertTrue(body["graded"])
|
|
self.assertIsNotNone(body["percent"])
|
|
self.assertIn("correct", {q["status"] for q in body["questions"]})
|
|
|
|
def test_a_live_attempt_with_nothing_answered_reads_as_nothing_answered(self):
|
|
quiz_id = self.bank.generate(is_shared=True, category_ids=[1], count=2).json()["id"]
|
|
with patch.dict(sys.modules, {"redis": self.redis}):
|
|
aid = self.client.post(f"/attempts/start?quiz_id={quiz_id}&mode=study").json()["id"]
|
|
body = self.client.get(f"/attempts/{aid}/analysis").json()
|
|
self.assertEqual((body["answered"], body["score"]), (0, 0))
|