"""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)