147 lines
6.7 KiB
Python
147 lines
6.7 KiB
Python
"""Flashcard-Übungspool: Leitner-Schritte, Deck-Bau (Level-Gate, fällig/neu), Persistenz."""
|
||
|
||
import json
|
||
from datetime import datetime, timedelta, timezone
|
||
|
||
from learning import LEITNER_MAX_BOX, PRACTICE_NEW_PER_SESSION, leitner_step
|
||
|
||
TOPIC = "t"
|
||
|
||
|
||
def _iso(days: float = 0) -> str:
|
||
return (datetime.now(timezone.utc) + timedelta(days=days)).isoformat()
|
||
|
||
|
||
async def _card(db, bn, sn, sub_title="Sub", q="Q?", block="Block"):
|
||
await db.put_sub_artifact(TOPIC, bn, sn, "flashcard",
|
||
json.dumps({"question": q, "answer": "A"}), block, sub_title)
|
||
|
||
|
||
# ── Leitner rein funktional ──────────────────────────────────────────────────────────
|
||
|
||
def test_leitner_step_transitions():
|
||
assert leitner_step(None, True) == (2, 1) # neue Karte gewusst → Box 2, morgen
|
||
assert leitner_step(None, False) == (1, 0) # neue Karte falsch → Box 1, sofort
|
||
assert leitner_step(2, True) == (3, 3)
|
||
assert leitner_step(LEITNER_MAX_BOX, True) == (LEITNER_MAX_BOX, 21) # Cap
|
||
assert leitner_step(4, False) == (1, 0) # falsch → zurück auf Anfang
|
||
|
||
|
||
# ── Persistenz ───────────────────────────────────────────────────────────────────────
|
||
|
||
async def test_progress_upsert_roundtrip(testdb):
|
||
db = testdb
|
||
await db.upsert_practice_progress(TOPIC, "b", "s", 2, _iso(1))
|
||
await db.upsert_practice_progress(TOPIC, "b", "s", 3, _iso(3))
|
||
rows = await db.get_practice_progress(TOPIC)
|
||
assert len(rows) == 1 and rows[0]["box"] == 3
|
||
|
||
|
||
async def test_progress_survives_artefakte_wipe(testdb):
|
||
db = testdb
|
||
await _card(db, "b", "s")
|
||
await db.upsert_practice_progress(TOPIC, "b", "s", 4, _iso(7))
|
||
await db.delete_sub_artefakte(TOPIC) # Regenerations-Wipe
|
||
assert (await db.get_practice_progress(TOPIC))[0]["box"] == 4
|
||
|
||
|
||
async def test_delete_topic_pipeline_clears_progress(testdb):
|
||
db = testdb
|
||
await db.upsert_practice_progress(TOPIC, "b", "s", 2, _iso(1))
|
||
await db.delete_topic_pipeline(TOPIC)
|
||
assert await db.get_practice_progress(TOPIC) == []
|
||
|
||
|
||
async def test_sub_levels_norm_and_counts(testdb):
|
||
db = testdb
|
||
await db.put_subblock(TOPIC, "b", "s1", "Block", "S1", level="beginner")
|
||
await db.put_subblock(TOPIC, "b", "s2", "Block", "S2", level="expert")
|
||
await db.put_subblock(TOPIC, "b", "s3", "Block", "S3", level="beginner", relevance="peripheral")
|
||
await db.put_subblock(TOPIC, "b", "s4", "Block", "S4", level="beginner", status="variant")
|
||
levels = await db.sub_levels_norm(TOPIC)
|
||
assert levels[("b", "s1")] == 1 and levels[("b", "s2")] == 3 and levels[("b", "s3")] == 4
|
||
assert ("b", "s4") not in levels # non-consensus ausgeschlossen
|
||
counts = await db.subs_per_level_norm(TOPIC)
|
||
assert counts["b"] == {1: 1, 2: 0, 3: 1, 4: 1}
|
||
|
||
|
||
# ── Deck-Bau ─────────────────────────────────────────────────────────────────────────
|
||
|
||
async def test_deck_level_gate_and_unlock(testdb):
|
||
from routes import build_practice_deck
|
||
db = testdb
|
||
# block_norm muss _norm_title(Roh-Titel) sein — so entsteht er auch in der Pipeline
|
||
await db.put_subblock(TOPIC, "block", "s1", "Block", "S1", level="beginner")
|
||
await db.put_subblock(TOPIC, "block", "s2", "Block", "S2", level="expert")
|
||
await _card(db, "block", "s1", "S1")
|
||
await _card(db, "block", "s2", "S2")
|
||
deck = await build_practice_deck(TOPIC)
|
||
assert [c["sub_norm"] for c in deck["cards"]] == ["s1"] # expert gesperrt
|
||
assert deck["counts"]["gesperrt"] == 1
|
||
# Score über S1+S2-Schwelle (2 Subs × 25 = 50) → expert (Level 3) frei
|
||
await db.set_block_score_and_streak(TOPIC, "Block", 50, 0)
|
||
deck = await build_practice_deck(TOPIC)
|
||
assert {c["sub_norm"] for c in deck["cards"]} == {"s1", "s2"}
|
||
|
||
|
||
async def test_deck_due_before_new_oldest_first(testdb):
|
||
from routes import build_practice_deck
|
||
db = testdb
|
||
for sn in ("s1", "s2", "s3"):
|
||
await db.put_subblock(TOPIC, "b", sn, "Block", sn.upper(), level="beginner")
|
||
await _card(db, "b", sn, sn.upper())
|
||
await db.upsert_practice_progress(TOPIC, "b", "s2", 2, _iso(-1))
|
||
await db.upsert_practice_progress(TOPIC, "b", "s3", 2, _iso(-5))
|
||
deck = await build_practice_deck(TOPIC)
|
||
assert [c["sub_norm"] for c in deck["cards"]] == ["s3", "s2", "s1"] # älteste fällige zuerst
|
||
assert [c["status"] for c in deck["cards"]] == ["due", "due", "new"]
|
||
assert deck["counts"] == {"due": 2, "new": 1, "new_total": 1, "gesperrt": 0}
|
||
|
||
|
||
async def test_deck_caps_new_and_reports_total(testdb):
|
||
from routes import build_practice_deck
|
||
db = testdb
|
||
for i in range(PRACTICE_NEW_PER_SESSION + 5):
|
||
sn = f"s{i:02d}"
|
||
await db.put_subblock(TOPIC, "b", sn, "Block", sn, level="beginner")
|
||
await _card(db, "b", sn, sn)
|
||
deck = await build_practice_deck(TOPIC)
|
||
assert deck["counts"]["new"] == PRACTICE_NEW_PER_SESSION
|
||
assert deck["counts"]["new_total"] == PRACTICE_NEW_PER_SESSION + 5
|
||
|
||
|
||
async def test_deck_future_due_sets_next_due_at(testdb):
|
||
from routes import build_practice_deck
|
||
db = testdb
|
||
await db.put_subblock(TOPIC, "b", "s1", "Block", "S1", level="beginner")
|
||
await _card(db, "b", "s1", "S1")
|
||
await db.upsert_practice_progress(TOPIC, "b", "s1", 3, _iso(3))
|
||
deck = await build_practice_deck(TOPIC)
|
||
assert deck["cards"] == [] and deck["counts"]["due"] == 0
|
||
assert deck["next_due_at"] is not None
|
||
|
||
|
||
async def test_deck_orphan_progress_and_legacy_block(testdb):
|
||
from routes import build_practice_deck
|
||
db = testdb
|
||
# Orphan: Progress ohne Karte → unschädlich, taucht nicht auf
|
||
await db.upsert_practice_progress(TOPIC, "weg", "s0", 2, _iso(-1))
|
||
# Legacy: Karte ohne subblocks-Zeilen → ungefiltert durchlassen
|
||
await _card(db, "leg", "sx", "SX")
|
||
deck = await build_practice_deck(TOPIC)
|
||
assert [c["block_norm"] for c in deck["cards"]] == ["leg"]
|
||
|
||
|
||
async def test_answer_books_without_card(testdb):
|
||
"""Antwort während Regeneration: bucht immer, kein Fehlerpfad."""
|
||
from models import PracticeAnswerRequest
|
||
from routes import practice_answer
|
||
db = testdb
|
||
res = await practice_answer(PracticeAnswerRequest(
|
||
topic=TOPIC, block_norm="b", sub_norm="s", correct=True))
|
||
assert res["box"] == 2
|
||
res = await practice_answer(PracticeAnswerRequest(
|
||
topic=TOPIC, block_norm="b", sub_norm="s", correct=False))
|
||
assert res["box"] == 1
|
||
assert (await db.get_practice_progress(TOPIC))[0]["box"] == 1
|