from __future__ import annotations from datetime import date, timedelta from sqlalchemy import select from app.core.database import SessionLocal from app.models.learning_state import ( GeneratedResource, LearningEvent, LessonProgress, RevisionItem, StudySession, ) def _payload(*, subjects: list[str] | None = None) -> dict[str, object]: return { "class_level": "SSLC / 10th", "board": "Kerala State Board", "subjects": subjects or ["Physics", "Chemistry"], "exam_date": (date.today() + timedelta(days=60)).isoformat(), "goal": "school_exam", "daily_minutes": 60, "preferred_time": "Evening", "preferences": {"language": "English", "learning_style": "Step-by-step"}, } def _signup(client, *, email: str) -> str: response = client.post( "/auth/signup", json={"name": "Learning Student", "email": email, "password": "Pass123!beta"}, ) assert response.status_code == 201, response.text return response.json()["access_token"] def _auth(token: str) -> dict[str, str]: return {"Authorization": f"Bearer {token}"} def test_empty_learning_state_never_invents_progress(client) -> None: response = client.get("/learning-state/me") assert response.status_code == 200 assert response.json() == { "profile": None, "subjects": [], "chapters": [], "tasks": [], "today_plan": None, "lesson_progress": [], "mastery": [], "quiz_attempts": [], "resources": [], "repair_items": [], "revision_items": [], "completed_lessons": 0, "completed_quiz_attempts": 0, "generated_resources": 0, "generated_notes": 0, "questions_asked": 0, "academic_state": { "class_level": None, "board": None, "subjects": [], "exam_date": None, "days_to_exam": None, "current_subject": None, "current_chapter": None, "current_topic": None, "last_meaningful_activity": None, "unfinished_lesson": None, "recent_mistakes": [], "weak_topics": [], "mastered_topics": [], "revision_due": [], "active_course": None, "active_tuition_session": None, "recent_study_chat_context": None, "suggested_notifications": [], }, "next_action": { "type": "COMPLETE_SETUP", "title": "Tell DocDoe what you are studying", "reason": ( "Class, board, subjects, exam date, and daily time are needed " "before DocDoe can choose responsibly." ), "action_label": "Set up my study plan", "subject": None, "chapter": None, "topic": None, "href": "/onboarding", "resume_payload": {}, "urgency": "normal", "estimated_minutes": 3, }, } def test_learning_state_counts_only_the_students_saved_questions(auth_client) -> None: # Real per-user auth is required here: the plain `client` fixture runs with # AUTH_ENABLED=false, so every request resolves to the same default user and # cross-user isolation cannot be observed. asker_token = _signup(auth_client, email="echo-learning-student@example.com") first = auth_client.post( "/chat/sessions", headers=_auth(asker_token), json={"subject": "Physics", "title": "Echo doubt"}, ) assert first.status_code == 201, first.text appended = auth_client.post( f"/chat/sessions/{first.json()['id']}/messages", headers=_auth(asker_token), json={ "user_content": "What is the condition for hearing an echo?", "assistant_content": "The reflected sound must return after at least 0.1 seconds.", "intent": "tutor", "evidence_label": "Kerala SCERT textbook · Sound Waves", }, ) assert appended.status_code == 201, appended.text second_token = _signup(auth_client, email="other-learning-student@example.com") second = auth_client.post( "/chat/sessions", headers=_auth(second_token), json={"subject": "Chemistry", "title": "Other student's chat"}, ) assert second.status_code == 201, second.text assert ( auth_client.post( f"/chat/sessions/{second.json()['id']}/messages", headers=_auth(second_token), json={ "user_content": "What is an acid?", "assistant_content": "An acid donates hydrogen ions in aqueous solution.", }, ).status_code == 201 ) state = auth_client.get("/learning-state/me", headers=_auth(asker_token)) assert state.status_code == 200 assert state.json()["questions_asked"] == 1 def test_onboarding_persists_profile_subjects_chapter_and_first_week(client) -> None: created = client.post("/learning-state/onboarding", json=_payload()) assert created.status_code == 200, created.text body = created.json() assert body["available_study_days"] == 60 assert len(body["tasks"]) == 7 assert body["tasks"][0]["subject"] == "Physics" assert body["tasks"][0]["chapter"] == "Sound Waves" assert body["tasks"][0]["href"].startswith("/tuition") assert body["tasks"][1]["subject"] == "Chemistry" assert body["tasks"][1]["task_type"] == "upload" fetched = client.get("/learning-state/me") assert fetched.status_code == 200 state = fetched.json() assert state["profile"]["onboarding_completed"] is True assert [item["name"] for item in state["subjects"]] == ["Physics", "Chemistry"] assert [item["title"] for item in state["chapters"]] == [ "Sound Waves", "Periodic Table and Electronic Configuration", ] assert state["mastery"] == [] # The legacy tutor context and the adaptive plan are committed together. # A fresh device uses this row for the onboarding gate and StudyChat. tutor_profile = client.get("/study-profile/me") assert tutor_profile.status_code == 200, tutor_profile.text saved_profile = tutor_profile.json() assert saved_profile["onboarding_completed"] is True assert saved_profile["grade"] == "SSLC / 10th" assert saved_profile["board"] == "Kerala State Board" assert saved_profile["subject"] == "Physics" assert saved_profile["time_left"] == _payload()["exam_date"] assert saved_profile["extra"]["subjects"] == ["Physics", "Chemistry"] assert saved_profile["extra"]["daily_minutes"] == 60 def test_onboarding_without_exam_date_still_builds_a_plan(client) -> None: # A student who does not know their exam date must not be trapped: onboarding # must complete, exam_date persists as null, and a steady default horizon is # used to build the first week. payload = _payload() payload.pop("exam_date") created = client.post("/learning-state/onboarding", json=payload) assert created.status_code == 200, created.text body = created.json() assert body["available_study_days"] == 90 # DEFAULT_UNKNOWN_HORIZON_DAYS assert len(body["tasks"]) == 7 fetched = client.get("/learning-state/me") assert fetched.status_code == 200 state = fetched.json() assert state["profile"]["onboarding_completed"] is True assert state["profile"]["exam_date"] is None def test_onboarding_rejects_a_past_exam_date(client) -> None: payload = _payload() payload["exam_date"] = (date.today() - timedelta(days=1)).isoformat() response = client.post("/learning-state/onboarding", json=payload) assert response.status_code == 422 assert response.json()["detail"]["code"] == "EXAM_DATE_NOT_FUTURE" assert client.get("/study-profile/me").status_code == 404 def test_generated_resource_is_readable_and_deletable_by_its_owner(client) -> None: with SessionLocal() as db: resource = GeneratedResource( user_id="usr_demo_student", resource_type="notes", title="Stored Sound Waves note", status="ready", resource_data={"key_points": ["f = 1 / T", "Sound needs a medium"]}, ) db.add(resource) db.commit() db.refresh(resource) resource_id = resource.id state = client.get("/learning-state/me") assert state.status_code == 200 assert state.json()["resources"][0]["title"] == "Stored Sound Waves note" assert state.json()["generated_notes"] == 1 deleted = client.delete(f"/learning-state/resources/{resource_id}") assert deleted.status_code == 204 refreshed = client.get("/learning-state/me").json() assert refreshed["resources"] == [] assert client.delete(f"/learning-state/resources/{resource_id}").status_code == 404 def test_new_event_ledger_links_quiz_lesson_and_revision_records(client) -> None: """Reusable reconciliation check for Phase 1's critical completion path.""" client.post("/learning-state/onboarding", json=_payload(subjects=["Physics"])) state = client.get("/learning-state/me").json() task = next(item for item in state["tasks"] if item["task_type"] == "lesson") lesson = client.post( "/learning-state/lessons/progress", json={ "chapter_catalog_id": "phy-p1-c1", "mission_id": task["mission_id"], "title": task["title"], "daily_task_id": task["id"], "duration_minutes": 20, "current_step": 11, "progress_percent": 100, "status": "completed", }, ) assert lesson.status_code == 200, lesson.text revision = client.post( "/learning-state/revision-items", json={ "client_item_id": f"reconcile:phy-p1-c1:{task['mission_id']}", "title": "Review: How sound is produced", "chapter_catalog_id": "phy-p1-c1", "mission_id": task["mission_id"], "source_kind": "lesson_recap", }, ) assert revision.status_code == 200, revision.text with SessionLocal() as db: events = db.scalars(select(LearningEvent)).all() by_type = {event.event_type: event for event in events} assert by_type["LESSON_COMPLETED"].entity_type == "lesson_progress" assert by_type["REVISION_ITEM_CREATED"].entity_id == revision.json()["item"]["id"] assert db.get(RevisionItem, by_type["REVISION_ITEM_CREATED"].entity_id) is not None def test_missed_task_recalculates_the_upcoming_schedule(client) -> None: plan = client.post("/learning-state/onboarding", json=_payload(subjects=["Physics"])).json() first_task = plan["tasks"][0] response = client.patch( f"/learning-state/tasks/{first_task['id']}/status", json={"status": "missed", "reason": "School event"}, ) assert response.status_code == 200, response.text body = response.json() assert body["task"]["status"] == "missed" assert body["recalculated_tasks"] replacement = next( item for item in body["recalculated_tasks"] if item["rescheduled_from_id"] == first_task["id"] ) assert replacement["status"] == "pending" assert replacement["priority"] > first_task["priority"] assert "recalculated" in body["message"].lower() def test_completed_tuition_class_updates_lesson_task_and_study_session_once(client) -> None: plan = client.post( "/learning-state/onboarding", json=_payload(subjects=["Physics"]), ).json() lesson_task = next(item for item in plan["tasks"] if item["task_type"] == "lesson") assert lesson_task["mission_id"] payload = { "chapter_catalog_id": "phy-p1-c1", "mission_id": lesson_task["mission_id"], "title": "How sound is produced", "daily_task_id": lesson_task["id"], "duration_minutes": 31, "current_step": 11, "progress_percent": 100, "status": "completed", } completed = client.post("/learning-state/lessons/progress", json=payload) assert completed.status_code == 200, completed.text assert completed.json()["completed_lessons"] == 1 assert completed.json()["completed_task_id"] == lesson_task["id"] repeated = client.post("/learning-state/lessons/progress", json=payload) assert repeated.status_code == 200, repeated.text assert repeated.json()["completed_lessons"] == 1 refreshed = client.get("/learning-state/me").json() assert refreshed["completed_lessons"] == 1 assert refreshed["lesson_progress"] == [ { "id": refreshed["lesson_progress"][0]["id"], "chapter_id": refreshed["chapters"][0]["id"], "chapter_catalog_id": "phy-p1-c1", "chapter": "Sound Waves", "subject_id": refreshed["subjects"][0]["id"], "mission_id": lesson_task["mission_id"], "status": "completed", "progress_percent": 100, "current_step": 11, "completed_at": refreshed["lesson_progress"][0]["completed_at"], "last_seen_at": refreshed["lesson_progress"][0]["last_seen_at"], } ] saved_task = next(item for item in refreshed["tasks"] if item["id"] == lesson_task["id"]) assert saved_task["status"] == "completed" with SessionLocal() as db: assert len(db.scalars(select(LessonProgress)).all()) == 1 sessions = db.scalars(select(StudySession)).all() assert len(sessions) == 1 assert sessions[0].duration_minutes == 31 events = db.scalars( select(LearningEvent).where(LearningEvent.event_type == "LESSON_COMPLETED") ).all() assert len(events) == 1 assert events[0].entity_id == refreshed["lesson_progress"][0]["id"] def test_manual_revision_save_is_authoritative_and_idempotent(client) -> None: client.post("/learning-state/onboarding", json=_payload(subjects=["Physics"])) payload = { "client_item_id": "tuition-method:phy-p1-c1:M1", "title": "Method: Oscillation, amplitude, period and frequency", "chapter_catalog_id": "phy-p1-c1", "mission_id": "M1", "topic_key": "phy-p1-c1:M1", "source_kind": "manual_save", "source_ref": "method", "due_in_days": 1, } created = client.post("/learning-state/revision-items", json=payload) assert created.status_code == 200, created.text assert created.json()["replayed"] is False replayed = client.post("/learning-state/revision-items", json=payload) assert replayed.status_code == 200, replayed.text assert replayed.json()["replayed"] is True assert replayed.json()["item"]["id"] == created.json()["item"]["id"] state = client.get("/learning-state/me").json() assert [item["id"] for item in state["revision_items"]] == [created.json()["item"]["id"]] assert state["revision_items"][0]["source_kind"] == "manual_save" with SessionLocal() as db: assert len(db.scalars(select(RevisionItem)).all()) == 1 events = db.scalars( select(LearningEvent).where(LearningEvent.event_type == "REVISION_ITEM_CREATED") ).all() assert len(events) == 1 assert events[0].entity_id == created.json()["item"]["id"] def test_wrong_answer_changes_mastery_and_adds_revision_task(client) -> None: client.post("/learning-state/onboarding", json=_payload(subjects=["Physics"])) state = client.get("/learning-state/me").json() subject_id = state["subjects"][0]["id"] chapter_id = state["chapters"][0]["id"] test_task = next( item for item in state["tasks"] if item["task_type"] in {"test", "practice", "lesson", "revision"} ) assessment = { "client_attempt_id": "quiz-sound-waves:attempt-1", "title": "Sound Waves chapter test", "topic_key": "sound-frequency", "topic_label": "Frequency and period", "score": 2, "max_score": 10, "subject_id": subject_id, "chapter_id": chapter_id, "daily_task_id": test_task["id"], "mission_id": "M6", "answers": [ { "question_id": "frequency-definition", "question": "Define frequency", "answer": "speed", "correct": False, } ], "corrections": [{"answer": "number of oscillations per second"}], "missing_keywords": ["oscillations per second", "hertz"], "misconceptions": ["frequency is wave speed"], } response = client.post("/learning-state/assessments", json=assessment) assert response.status_code == 200, response.text consequence = response.json() assert consequence["mastery_after"] < 20 assert consequence["revision_task"]["task_type"] == "revision" assert consequence["revision_task"]["mission_id"] == "M6" assert consequence["next_recommended_task_id"] == consequence["revision_task"]["id"] replayed = client.post("/learning-state/assessments", json=assessment) assert replayed.status_code == 200, replayed.text replayed_consequence = replayed.json() assert replayed_consequence["attempt_id"] == consequence["attempt_id"] assert replayed_consequence["revision_task"]["id"] == consequence["revision_task"]["id"] assert "already saved" in replayed_consequence["message"].lower() refreshed = client.get("/learning-state/me").json() assert refreshed["completed_quiz_attempts"] == 1 assert refreshed["quiz_attempts"][0]["title"] == "Sound Waves assessment" assert refreshed["quiz_attempts"][0]["score"] == 2 # The centralized mastery engine reports evidence-derived states # (not the legacy secure/needs_revision pair). assert refreshed["mastery"][0]["last_result"] == "developing" assert refreshed["mastery"][0]["state"] == "developing" assert any(item["title"] == "Frequency and period" for item in refreshed["revision_items"]) today_titles = [task["title"] for task in (refreshed.get("today_plan") or {}).get("tasks") or []] assert any(title.startswith("Revise — Frequency and period") for title in today_titles) with SessionLocal() as db: quiz_events = db.scalars( select(LearningEvent).where(LearningEvent.event_type == "QUIZ_COMPLETED") ).all() assert len(quiz_events) == 1 assert quiz_events[0].entity_id == consequence["attempt_id"] assert quiz_events[0].topic_key == "sound-frequency" def test_assessment_rejects_a_score_above_the_maximum(client) -> None: response = client.post( "/learning-state/assessments", json={ "client_attempt_id": "quiz-invalid-score:attempt-1", "title": "Invalid score", "topic_key": "invalid-score", "topic_label": "Invalid score", "score": 6, "max_score": 5, }, ) assert response.status_code == 422 assert client.get("/learning-state/me").json()["completed_quiz_attempts"] == 0 def test_assistant_plan_change_only_schedules_verified_chapters_and_reschedules_work(client) -> None: client.post("/learning-state/onboarding", json=_payload()) response = client.post( "/learning-state/plan-adjustments", json={ "target_subject": "Physics", "replace_subject": "Chemistry", "reason": "Physics exam tomorrow", "chapters": ["1", "Quantum Gravity"], }, ) assert response.status_code == 200, response.text body = response.json() physics_tasks = [task for task in body["tasks"] if task["subject"] == "Physics"] assert physics_tasks assert physics_tasks[0]["chapter"] == "Sound Waves" assert "EXPLICIT_TODAY_FOCUS" in (physics_tasks[0].get("reason_codes") or []) assert body["unavailable_chapters"] == ["Quantum Gravity"] assert "Plan updated" in body["message"] assert "Quantum Gravity" in body["message"] repeated = client.post( "/learning-state/plan-adjustments", json={ "target_subject": "Physics", "replace_subject": "Chemistry", "reason": "Physics exam tomorrow", "chapters": ["1", "Quantum Gravity"], }, ) assert repeated.status_code == 200, repeated.text assert [task["logical_key"] for task in repeated.json()["tasks"]] == [ task["logical_key"] for task in body["tasks"] ] refreshed = client.get("/learning-state/me").json() today = refreshed["today_plan"] assert today is not None assert today["tasks"][0]["chapter"] == "Sound Waves" assert today["generation"] >= 1 def test_assistant_rejects_unavailable_chapters_without_changing_the_plan(client) -> None: client.post("/learning-state/onboarding", json=_payload()) before = client.get("/learning-state/me").json() response = client.post( "/learning-state/plan-adjustments", json={ "target_subject": "Physics", "replace_subject": "Chemistry", "reason": "Physics exam tomorrow", "chapters": ["Quantum Gravity", "Imaginary Optics"], }, ) assert response.status_code == 422, response.text assert response.json()["detail"]["code"] == "CHAPTER_NOT_AVAILABLE" assert "Quantum Gravity" in response.json()["detail"]["message"] after = client.get("/learning-state/me").json() assert [ (task["id"], task["status"], task["scheduled_for"]) for task in after["tasks"] ] == [ (task["id"], task["status"], task["scheduled_for"]) for task in before["tasks"] ] def test_learning_state_is_owned_and_isolated(auth_client) -> None: alice = _signup(auth_client, email="learning-alice@docdoe.test") bob = _signup(auth_client, email="learning-bob@docdoe.test") assert auth_client.post( "/learning-state/onboarding", headers=_auth(alice), json=_payload(subjects=["Physics"]) ).status_code == 200 bob_state = auth_client.get("/learning-state/me", headers=_auth(bob)) assert bob_state.status_code == 200 assert bob_state.json()["profile"] is None assert bob_state.json()["tasks"] == [] assert auth_client.get("/learning-state/me").status_code in (401, 403) def test_revision_items_are_owned_and_hidden_from_other_students(auth_client) -> None: alice = _signup(auth_client, email="revision-alice@docdoe.test") bob = _signup(auth_client, email="revision-bob@docdoe.test") alice_headers = _auth(alice) bob_headers = _auth(bob) assert auth_client.post( "/learning-state/onboarding", headers=alice_headers, json=_payload(subjects=["Physics"]) ).status_code == 200 assert auth_client.post( "/learning-state/onboarding", headers=bob_headers, json=_payload(subjects=["Physics"]) ).status_code == 200 created = auth_client.post( "/learning-state/revision-items", headers=alice_headers, json={ "client_item_id": "alice:revision:phy-p1-c1:M1", "title": "Review: Oscillation", "chapter_catalog_id": "phy-p1-c1", "mission_id": "M1", "source_kind": "manual_save", }, ) assert created.status_code == 200, created.text # The same client key is only idempotent inside one account; it cannot # address Alice's row or reveal it to Bob. bob_save = auth_client.post( "/learning-state/revision-items", headers=bob_headers, json={ "client_item_id": "alice:revision:phy-p1-c1:M1", "title": "Review: Oscillation", "chapter_catalog_id": "phy-p1-c1", "mission_id": "M1", "source_kind": "manual_save", }, ) assert bob_save.status_code == 200, bob_save.text assert bob_save.json()["item"]["id"] != created.json()["item"]["id"] bob_state = auth_client.get("/learning-state/me", headers=bob_headers) assert [item["id"] for item in bob_state.json()["revision_items"]] == [ bob_save.json()["item"]["id"] ]