chatkit/backend/tests/test_database.py

"""Tests for the SQLAlchemy data-access layer (run against a temp SQLite file)."""

from __future__ import annotations

import pytest

from app import config as config_module
from app import database
from app import db as db_module
from app.answer_grid import from_dict as grid_from_dict, score_student


@pytest.fixture
async def fresh_db(tmp_path, monkeypatch):
    db_file = tmp_path / "t.db"
    cached = config_module.get_settings()
    monkeypatch.setattr(cached, "database_provider", "sqlite", raising=False)
    monkeypatch.setattr(cached, "supabase_db_url", "", raising=False)
    monkeypatch.setattr(cached, "database_path", str(db_file), raising=False)
    monkeypatch.setattr(cached, "admin_emails", "", raising=False)
    db_module.get_engine.cache_clear()
    db_module.get_sessionmaker.cache_clear()
    await database.init_database()
    yield
    db_module.get_engine.cache_clear()
    db_module.get_sessionmaker.cache_clear()


@pytest.mark.unit
async def test_teacher_profile_roundtrip(fresh_db):
    uid = await database.create_user(
        "t@example.com", "hash", full_name="Wang", school="First High"
    )
    user = await database.get_user_by_email("t@example.com")
    assert user["id"] == uid
    assert user["full_name"] == "Wang"
    assert user["school"] == "First High"
    assert user["is_admin"] is False


@pytest.mark.unit
async def test_student_matched_by_normalized_name(fresh_db):
    uid = await database.create_user("t@example.com", "h")
    a = await database.match_or_create_student(uid, "Alice Chen")
    b = await database.match_or_create_student(uid, "  alice   chen ")
    c = await database.match_or_create_student(uid, "Bob")
    assert a == b  # whitespace/case-insensitive match
    assert a != c
    students = await database.get_students(uid)
    assert len(students) == 2


@pytest.mark.unit
async def test_student_timeline_across_quizzes(fresh_db):
    uid = await database.create_user("t@example.com", "h")
    sid = await database.match_or_create_student(uid, "Alice")

    q1 = await database.create_session(uid, "Quiz 1")
    await database.update_session_status(q1, "saved")
    await database.save_student_result(q1, sid, 60.0)
    q2 = await database.create_session(uid, "Quiz 2")
    await database.update_session_status(q2, "saved")
    await database.save_student_result(q2, sid, 100.0)

    timeline = await database.get_student_timeline(sid)
    assert [t["score"] for t in timeline] == [60.0, 100.0]
    assert timeline[0]["quiz_title"] == "Quiz 1"


@pytest.mark.unit
async def test_save_student_result_is_upsert(fresh_db):
    uid = await database.create_user("t@example.com", "h")
    sid = await database.match_or_create_student(uid, "Alice")
    q = await database.create_session(uid, "Quiz")
    await database.update_session_status(q, "saved")
    await database.save_student_result(q, sid, 40.0)
    await database.save_student_result(q, sid, 80.0)
    timeline = await database.get_student_timeline(sid)
    assert len(timeline) == 1  # same quiz+student -> updated, not duplicated
    assert timeline[0]["score"] == 80.0


@pytest.mark.unit
async def test_stats_counts(fresh_db):
    uid = await database.create_user("t@example.com", "h")
    sid = await database.match_or_create_student(uid, "Alice")
    q = await database.create_session(uid, "Quiz")
    # get_stats' "quizzes" count matches admin_list_sessions (the 測驗 tab):
    # status="saved" only, not just any non-draft status — a "processed"
    # (uploaded but never saved) quiz doesn't count here either.
    await database.update_session_status(q, "processed")
    await database.save_student_result(q, sid, 100.0)
    stats = await database.get_stats()
    assert stats["teachers"] == 1
    assert stats["students"] == 1
    assert stats["quizzes"] == 0
    assert stats["student_results"] == 1

    await database.update_session_status(q, "saved")
    stats = await database.get_stats()
    assert stats["quizzes"] == 1


@pytest.mark.unit
async def test_stats_and_lists_scope_to_teacher(fresh_db):
    # Two teachers, each with their own quiz + student + result.
    t1 = await database.create_user("t1@example.com", "h")
    t2 = await database.create_user("t2@example.com", "h")
    s1 = await database.match_or_create_student(t1, "Alice")
    s2 = await database.match_or_create_student(t2, "Bob")
    q1 = await database.create_session(t1, "T1 Quiz")
    q2 = await database.create_session(t2, "T2 Quiz")
    await database.save_student_result(q1, s1, 100.0)
    await database.save_student_result(q2, s2, 33.3)

    # Global (admin) sees everything.
    g = await database.get_stats()
    assert g["teachers"] == 2 and g["students"] == 2 and g["quizzes"] == 0 and g["student_results"] == 2
    # (quizzes == 0 above: get_stats' "quizzes" count matches admin_list_sessions
    # — status="saved" only — and these two sessions are still drafts;
    # save_student_result doesn't touch status.)

    # Scoped to t1 sees only t1's data.
    sc = await database.get_stats(t1)
    assert sc["students"] == 1 and sc["quizzes"] == 0 and sc["student_results"] == 1

    # admin_list_sessions only surfaces status="saved" quizzes (a quiz becomes
    # a record there only once the teacher finishes and explicitly saves).
    assert await database.admin_list_sessions(t1) == []
    await database.update_session_status(q1, "saved")
    await database.update_session_status(q2, "saved")

    assert {x["title"] for x in await database.admin_list_sessions(t1)} == {"T1 Quiz"}
    assert {x["name"] for x in await database.admin_list_students(t1)} == {"Alice"}
    assert len(await database.admin_list_sessions()) == 2  # admin sees both


@pytest.mark.unit
async def test_teacher_retention_windows(fresh_db):
    from datetime import datetime, timedelta, timezone

    from app.db import get_sessionmaker
    from app.models import LoginEvent, Teacher

    now = datetime.now(timezone.utc)
    async with get_sessionmaker()() as db:
        old_returning = Teacher(
            email="returning@example.com",
            password_hash="h",
            created_at=now - timedelta(days=40),
        )
        old_one_time = Teacher(
            email="onetime@example.com",
            password_hash="h",
            created_at=now - timedelta(days=40),
        )
        too_new = Teacher(
            email="new@example.com", password_hash="h", created_at=now - timedelta(days=3)
        )
        db.add_all([old_returning, old_one_time, too_new])
        await db.flush()
        # Returning teacher logged in again well after both the week and
        # month cutoffs from their signup.
        db.add(
            LoginEvent(
                role="teacher", user_id=old_returning.id, created_at=now - timedelta(days=2)
            )
        )
        await db.commit()

    retention = await database.get_teacher_retention()
    # Only the two 40-day-old accounts are eligible for either window — the
    # 3-day-old one hasn't existed long enough to judge yet.
    assert retention["week"]["eligible"] == 2
    assert retention["week"]["retained"] == 1
    assert retention["week"]["pct"] == 50.0
    assert retention["month"]["eligible"] == 2
    assert retention["month"]["retained"] == 1


@pytest.mark.unit
async def test_feedback_loop_funnel_stages(fresh_db):
    from datetime import datetime, timedelta, timezone

    from app.db import get_sessionmaker
    from app.models import PracticeQuiz, Quiz, StudentResult

    uid = await database.create_user("t@example.com", "h")
    s1 = await database.match_or_create_student(uid, "Alice")
    s2 = await database.match_or_create_student(uid, "Bob")

    now = datetime.now(timezone.utc)
    t0 = now - timedelta(days=10)
    async with get_sessionmaker()() as db:
        quiz1 = Quiz(teacher_id=uid, title="Exam 1", status="saved", created_at=t0)
        db.add(quiz1)
        await db.flush()
        db.add_all(
            [
                # Alice: weakness identified, practice generated + completed,
                # then reassessed with a later real exam — completes all 5 stages.
                StudentResult(
                    quiz_id=quiz1.id, student_id=s1, score=60.0, weaknesses="verb tense"
                ),
                # Bob: no weakness identified — stops at stage 1.
                StudentResult(quiz_id=quiz1.id, student_id=s2, score=95.0, weaknesses=""),
            ]
        )
        db.add(
            PracticeQuiz(
                teacher_id=uid,
                student_id=s1,
                share_token="tok1",
                status="completed",
                created_at=t0 + timedelta(days=1),
                completed_at=t0 + timedelta(days=2),
            )
        )
        await db.flush()
        quiz2 = Quiz(
            teacher_id=uid, title="Exam 2", status="saved", created_at=t0 + timedelta(days=3)
        )
        db.add(quiz2)
        await db.flush()
        db.add(StudentResult(quiz_id=quiz2.id, student_id=s1, score=80.0))
        await db.commit()

    funnel = await database.get_feedback_loop_funnel()
    # Alice+Bob's results on quiz1, plus Alice's reassessment result on quiz2.
    assert funnel["uploaded"] == 3
    assert funnel["weakness_identified"] == 1  # Alice only
    assert funnel["practice_generated"] == 1
    assert funnel["practice_completed"] == 1
    assert funnel["reassessed"] == 1  # Alice's quiz2 came after her practice completed


@pytest.mark.unit
async def test_issue_report_lifecycle(fresh_db):
    admin_id = await database.create_user("admin@example.com", "h")
    await database.set_user_admin(admin_id, True)
    await database.create_user("teacher@example.com", "h")

    assert await database.get_admin_emails() == ["admin@example.com"]

    created = await database.create_issue_report(
        "teacher@example.com", "bug", "Upload fails", "Steps to reproduce...", "/dashboard"
    )
    assert created["status"] == "open"

    open_issues = await database.list_open_issue_reports()
    assert len(open_issues) == 1
    assert open_issues[0]["subject"] == "Upload fails"

    assert await database.dismiss_issue_report(created["id"]) is True
    assert await database.list_open_issue_reports() == []
    # Already dismissed — dismissing again finds nothing to update.
    assert await database.dismiss_issue_report(created["id"]) is False


@pytest.mark.unit
def test_score_student_grades_only_concrete_keys():
    grid = grid_from_dict(
        {
            "total_questions": 4,
            # Q4 official is blank -> not graded
            "official_answers": ["A", "B", "C", ""],
            "students": [{"name": "Alice", "answers": ["A", "X", "=", ""]}],
            "questions": [],
        }
    )
    graded, correct, score = score_student(grid, 0)
    # Graded = Q1..Q3 (3). Correct: Q1 (A), Q3 ("=") -> 2 correct, Q2 wrong.
    assert graded == 3
    assert correct == 2
    assert score == pytest.approx(66.7, abs=0.1)