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