Spaces:
Sleeping
Sleeping
taboola
Revert "Revert "add teacher retention, feedback-loop, and issue-report analytics to admin backend""
701a99e | """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 | |
| 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() | |
| 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 | |
| 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 | |
| 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" | |
| 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 | |
| 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 | |
| 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 | |
| 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 | |
| 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 | |
| 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 | |
| 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) | |