File size: 11,859 Bytes
7c6ffa6 d5ee82b 7c6ffa6 d5ee82b 7c6ffa6 d5ee82b 7c6ffa6 d5ee82b 7c6ffa6 d5ee82b 7c6ffa6 d5ee82b 7c6ffa6 d5ee82b 7c6ffa6 d5ee82b 7c6ffa6 d5ee82b 7c6ffa6 d5ee82b 7c6ffa6 d5ee82b 7c6ffa6 d5ee82b 7c6ffa6 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 | """API-level tests for checkpoints, the repair queue, and the daily planner."""
from __future__ import annotations
from datetime import date, datetime, timedelta, timezone
from sqlalchemy import select
from app.core.database import SessionLocal
from app.models.learning_state import RepairItem, RevisionItem, TopicMastery
def _payload(*, subjects: list[str] | None = None, daily_minutes: int = 60) -> dict[str, object]:
return {
"class_level": "SSLC / 10th",
"board": "Kerala State Board",
"subjects": subjects or ["Physics"],
"exam_date": (date.today() + timedelta(days=60)).isoformat(),
"goal": "school_exam",
"daily_minutes": daily_minutes,
"preferred_time": "Evening",
"preferences": {"language": "English"},
}
def _checkpoint(
*,
event_id: str,
correct: bool,
attempt_index: int = 1,
kind: str = "checkpoint_mcq",
question_id: str = "m1-q1",
student_answer: str | None = None,
) -> dict[str, object]:
return {
"client_event_id": event_id,
"concept_key": "phy-p1-c1:M1",
"concept_label": "Oscillation, Amplitude, Period and Frequency",
"chapter_catalog_id": "phy-p1-c1",
"mission_id": "M1",
"kind": kind,
"question_id": question_id,
"question_type": "mcq",
"prompt": "The SI unit of frequency is:",
"student_answer": student_answer if student_answer is not None else ("hertz" if correct else "metre"),
"correct_answer": "hertz",
"client_correct": correct,
"attempt_index": attempt_index,
"hint_used": False,
"expected_keywords": ["hertz"],
}
def test_wrong_checkpoint_creates_repair_and_mastery_consequence(client) -> None:
client.post("/learning-state/onboarding", json=_payload())
response = client.post(
"/learning-state/checkpoints",
json=_checkpoint(event_id="evt-wrong-0001", correct=False),
)
assert response.status_code == 200, response.text
body = response.json()
assert body["correct"] is False
assert body["error_category"] == "concept_misunderstanding"
assert body["repair_status"] == "created"
assert body["repair_item"]["status"] == "open"
assert body["mastery"]["after_state"] == "needs_repair"
assert body["diagnosis"]
state = client.get("/learning-state/me").json()
assert len(state["repair_items"]) == 1
assert state["repair_items"][0]["error_category"] == "concept_misunderstanding"
assert state["mastery"][0]["state"] == "needs_repair"
def test_checkpoint_replay_is_idempotent(client) -> None:
client.post("/learning-state/onboarding", json=_payload())
first = client.post(
"/learning-state/checkpoints",
json=_checkpoint(event_id="evt-replay-01", correct=False),
).json()
replay = client.post(
"/learning-state/checkpoints",
json=_checkpoint(event_id="evt-replay-01", correct=False),
).json()
assert replay["replayed"] is True
assert replay["checkpoint_id"] == first["checkpoint_id"]
with SessionLocal() as db:
assert len(db.scalars(select(RepairItem)).all()) == 1
mastery = db.scalar(select(TopicMastery))
assert mastery is not None and mastery.attempts_count == 1
def test_repeated_same_error_escalates_one_repair_item(client) -> None:
client.post("/learning-state/onboarding", json=_payload())
client.post("/learning-state/checkpoints", json=_checkpoint(event_id="evt-esc-1", correct=False))
second = client.post(
"/learning-state/checkpoints",
json=_checkpoint(event_id="evt-esc-2", correct=False, question_id="m1-q1", attempt_index=2),
).json()
assert second["repair_status"] == "escalated"
assert second["repair_item"]["support_level"] == 2
assert second["repair_item"]["failed_attempts"] == 2
state = client.get("/learning-state/me").json()
open_items = [
item for item in state["repair_items"] if item["status"] in {"open", "escalated"}
]
assert len(open_items) == 1
def test_successful_retry_resolves_repair_idempotently(client) -> None:
client.post("/learning-state/onboarding", json=_payload())
client.post("/learning-state/checkpoints", json=_checkpoint(event_id="evt-fix-1", correct=False))
fixed = client.post(
"/learning-state/checkpoints",
json=_checkpoint(event_id="evt-fix-2", correct=True, attempt_index=2, question_id="m1-q2"),
).json()
assert fixed["repair_status"] == "resolved"
assert fixed["repair_item"]["retry_result"] == "recovered"
state = client.get("/learning-state/me").json()
assert all(item["status"] == "resolved" for item in state["repair_items"])
assert state["mastery"][0]["state"] != "needs_repair"
# Re-sending the successful retry does not resurrect or duplicate anything.
replay = client.post(
"/learning-state/checkpoints",
json=_checkpoint(event_id="evt-fix-2", correct=True, attempt_index=2, question_id="m1-q2"),
).json()
assert replay["replayed"] is True
def test_server_verifies_mcq_answers_itself(client) -> None:
client.post("/learning-state/onboarding", json=_payload())
response = client.post(
"/learning-state/checkpoints",
json=_checkpoint(
event_id="evt-lie-1",
correct=True, # client CLAIMS correct
student_answer="metre", # but the answer is wrong
),
).json()
assert response["correct"] is False
assert response["repair_status"] == "created"
def test_plan_today_adds_repair_task_with_reason_and_is_deterministic(client) -> None:
client.post("/learning-state/onboarding", json=_payload())
client.post("/learning-state/checkpoints", json=_checkpoint(event_id="evt-plan-1", correct=False))
first = client.post("/learning-state/plan/today")
assert first.status_code == 200, first.text
body = first.json()
repair_tasks = [task for task in body["tasks"] if task["task_type"] == "mistake_repair"]
assert len(repair_tasks) == 1
assert repair_tasks[0]["repair_item_id"]
assert "repair" in (repair_tasks[0]["reason"] or "").lower()
assert body["planned_minutes"] <= body["daily_minutes"]
# Deterministic: a second ensure is read-only and does not mint another plan.
second = client.post("/learning-state/plan/today").json()
assert second["created_task_ids"] == []
assert second["plan_id"] == body["plan_id"]
repair_tasks_after = [task for task in second["tasks"] if task["task_type"] == "mistake_repair"]
assert len(repair_tasks_after) == 1
def test_plan_today_respects_time_budget(client) -> None:
client.post("/learning-state/onboarding", json=_payload(daily_minutes=15))
client.post("/learning-state/checkpoints", json=_checkpoint(event_id="evt-budget-1", correct=False))
body = client.post("/learning-state/plan/today").json()
today_new_minutes = sum(
task["duration_minutes"]
for task in body["tasks"]
if task["id"] in body["created_task_ids"]
)
assert today_new_minutes <= 15 or len(body["created_task_ids"]) <= 1
def test_plan_today_schedules_due_revision_and_cancels_stale_ones(client) -> None:
client.post("/learning-state/onboarding", json=_payload())
client.post(
"/learning-state/checkpoints",
json=_checkpoint(event_id="evt-rev-1", correct=True, question_id="m1-q9"),
)
due_at = datetime.now(timezone.utc) - timedelta(hours=2)
with SessionLocal() as db:
mastery = db.scalar(select(TopicMastery))
assert mastery is not None
mastery.next_review_at = due_at
revision = db.scalar(select(RevisionItem))
assert revision is not None
revision.due_at = due_at
revision.status = "due"
db.commit()
body = client.post(
"/learning-state/today-plan/replan",
json={"idempotency_key": "evt-rev-due", "reason": "meaningful_evidence"},
).json()
revision_tasks = [
task
for task in body["tasks"]
if task["task_type"] == "revision" and task.get("revision_concept_key")
]
assert len(revision_tasks) == 1
assert "revise" in revision_tasks[0]["title"].lower()
assert revision_tasks[0]["reason"]
# The student revises (mastery engine pushes next_review_at forward).
client.post(
"/learning-state/checkpoints",
json=_checkpoint(event_id="evt-rev-2", correct=True, kind="revision_recall", question_id="m1-q10"),
)
cleaned = client.get("/learning-state/today-plan").json()
assert all(
task["id"] != revision_tasks[0]["id"] or task["status"] != "pending"
for task in cleaned["tasks"]
)
def test_checkpoint_ownership_is_isolated(auth_client) -> None:
def signup(email: str) -> str:
response = auth_client.post(
"/auth/signup",
json={"name": "Student", "email": email, "password": "Pass123!beta"},
)
assert response.status_code == 201, response.text
return response.json()["access_token"]
alice = signup("adaptive-alice@docdoe.test")
bob = signup("adaptive-bob@docdoe.test")
headers_alice = {"Authorization": f"Bearer {alice}"}
headers_bob = {"Authorization": f"Bearer {bob}"}
auth_client.post("/learning-state/onboarding", headers=headers_alice, json=_payload())
auth_client.post(
"/learning-state/checkpoints",
headers=headers_alice,
json=_checkpoint(event_id="evt-iso-1", correct=False),
)
bob_state = auth_client.get("/learning-state/me", headers=headers_bob).json()
assert bob_state["repair_items"] == []
assert bob_state["mastery"] == []
def test_empty_account_plan_today_requires_setup(client) -> None:
response = client.post("/learning-state/plan/today")
assert response.status_code == 409
assert response.json()["detail"]["code"] == "STUDY_PLAN_REQUIRED"
def test_plan_today_cancels_coverage_for_completed_mission(client) -> None:
client.post("/learning-state/onboarding", json=_payload())
# Onboarding already creates the curated M1 lesson task. The planner must
# NOT add a coverage twin for the same mission — the student would see the
# same class listed twice.
first = client.post("/learning-state/plan/today").json()
m1_lessons = [
task
for task in first["tasks"]
if task["task_type"] in {"lesson", "new_lesson", "continue_lesson"}
and task["mission_id"] == "M1"
and task["status"] == "pending"
]
assert len(m1_lessons) == 1, [task["title"] for task in first["tasks"]]
assert m1_lessons[0]["source"] != "adaptive_planner_v2"
# The student completes the M1 lesson (which completes ONE linked task).
done = client.post(
"/learning-state/lessons/progress",
json={
"chapter_catalog_id": "phy-p1-c1",
"mission_id": "M1",
"title": "Oscillation, Amplitude, Period and Frequency",
"duration_minutes": 20,
"current_step": 11,
"progress_percent": 100,
"status": "completed",
},
)
assert done.status_code == 200, done.text
# Reconcile again: no pending M1 lesson may linger as fake remaining work,
# and the planner may now schedule the NEXT prerequisite-safe concept.
second = client.post("/learning-state/plan/today").json()
remaining_m1_lessons = [
task
for task in second["tasks"]
if task["task_type"] == "lesson" and task["mission_id"] == "M1"
]
assert remaining_m1_lessons == []
next_lessons = [
task
for task in second["tasks"]
if task["task_type"] == "lesson" and task["source"] == "adaptive_planner_v2"
]
assert all(task["mission_id"] != "M1" for task in next_lessons)
|