Blame for tests/test_classroom.py

import asyncio
import io
import json
import tempfile
import unittest
import zipfile
from pathlib import Path
from types import SimpleNamespace
from unittest.mock import patch

import classroom_service as classroom


class ClassroomServiceTests(unittest.TestCase):
    def setUp(self):
        self.temp_dir = tempfile.TemporaryDirectory()
        classroom.RUNTIME = Path(self.temp_dir.name)
        classroom.DB = classroom.RUNTIME / "classroom.sqlite3"
        classroom.live.clear()
        classroom._student_connection_windows.clear()
        classroom._student_message_windows.clear()
        classroom.init_db()
        with classroom.lock, classroom.sqlite3.connect(classroom.DB) as db:
            db.execute("INSERT OR IGNORE INTO classroom_courses VALUES(?,?,?)", ("L6A", "2026 Fall 進階專題", 0))

    def tearDown(self):
        self.temp_dir.cleanup()

    def test_normalizes_multifile_project_and_rejects_unsafe_paths(self):
        files = classroom.norm_files(
            {
                "main.py": "from clues import answer\nprint(answer)\n",
                "clues.py": "answer = 42\n",
                "data/case.json": json.dumps({"place": "library"}),
                "../secret.py": "nope",
                "image.png": "not text",
            }
        )
        self.assertEqual(set(files), {"main.py", "clues.py", "data/case.json"})

    def test_publish_uses_latest_submission_while_history_keeps_every_version(self):
        state = classroom.empty("L6A", "student-1", "Alex", "week02")
        state["project_name"] = "校園謎案"
        state["files"] = {"main.py": 'print("first")\n'}
        state["final_files"] = dict(state["files"])
        state["submitted_at"] = 100
        classroom.record_version(state, "submit")
        classroom.save(state)

        state["files"] = {"main.py": 'print("latest")\n', "clues.py": "CLUE = 3\n"}
        state["final_files"] = dict(state["files"])
        state["submitted_at"] = 200
        classroom.record_version(state, "submit")
        classroom.save(state)

        restored = classroom.load("L6A", "student-1", "Alex", "week02")
        self.assertEqual(restored["final_files"], state["files"])
        self.assertNotIn("first", restored["final_files"]["main.py"])
        history = classroom.version_history("L6A", "student-1")
        self.assertEqual(len(history), 2)
        self.assertIn("latest", history[0]["files"]["main.py"])
        self.assertIn("first", history[1]["files"]["main.py"])

    def test_keeps_latest_weekly_checkpoint_separate_from_final(self):
        state = classroom.empty("L6A", "student-1", "Alex", "week02")
        state["project_name"] = "校園謎案"
        state["files"] = {"main.py": 'print("checkpoint")\n'}
        state["checkpoint_files"] = dict(state["files"])
        state["checkpoint_at"] = 100
        classroom.record_version(state, "checkpoint")
        classroom.save(state)

        restored = classroom.load("L6A", "student-1", "Alex", "week02")
        self.assertEqual(restored["checkpoint_files"], state["files"])
        self.assertIsNone(restored["final_files"])
        with zipfile.ZipFile(io.BytesIO(classroom.make_zip("L6A", "week02"))) as archive:
            self.assertEqual(archive.namelist(), [])

    def test_bootstrap_preserves_checkpoint_separately_from_later_repair(self):
        classroom.init_db()
        loaded = classroom.load("L6Sat", "李合盛", "李合盛", "week02")

        repaired = (classroom.REPAIRS / "L6Sat" / "week02" / "李合盛.py").read_text(encoding="utf-8")
        self.assertEqual(loaded["files"]["main.py"], repaired)
        self.assertNotEqual(loaded["files"], loaded["checkpoint_files"])
        matching = [
            item
            for item in classroom.version_history("L6Sat", "李合盛")
            if item["source"] == "checkpoint"
            and item["submitted_at"] == loaded["checkpoint_at"]
        ]
        self.assertEqual(len(matching), 1)
        self.assertEqual(matching[0]["files"], loaded["checkpoint_files"])

    def test_fresh_bootstrap_does_not_replace_new_submission_with_legacy_repair(self):
        data = Path(self.temp_dir.name) / "seed"
        data.mkdir()
        (data / "classes.json").write_text("[]", encoding="utf-8")
        versions = data / "versions"
        repair = data / "repairs" / "L6A" / "week02" / "student.py"
        repair.parent.mkdir(parents=True)
        repair.write_text('print("old repair")\n', encoding="utf-8")
        files = {"main.py": 'print("latest student story")\n'}
        timestamp = classroom.LEGACY_REPAIR_CUTOFF_MS + 1
        record = {"class_code": "L6A", "student_id": "student", "display_name": "Alex",
                  "week_id": "week02", "version_id": "teacher-submit", "kind": "submit",
                  "project_name": "case", "created_at": timestamp, "files": files}
        version = versions / "L6A" / "student" / "week02" / "teacher-submit.json"
        version.parent.mkdir(parents=True)
        version.write_text(json.dumps(record), encoding="utf-8")
        with patch.multiple(classroom, DATA=data, REPAIRS=data / "repairs",
                            VERSIONS=versions, CHECKPOINTS=data / "checkpoints",
                            DB=Path(self.temp_dir.name) / "fresh.sqlite3"):
            classroom.init_db()
            restored = classroom.load("L6A", "student", "Alex", "week02")
            self.assertEqual(restored["files"], files)
            self.assertEqual(restored["final_files"], files)
            self.assertEqual(restored["submitted_at"], timestamp)

    def test_progress_picker_lists_every_checkpoint_and_submission_newest_first(self):
        state = classroom.empty("L6A", "student-1", "Alex", "week02")
        state["project_name"] = "校園謎案"
        state["final_files"] = {"main.py": 'print("formal")\n'}
        state["submitted_at"] = 100
        classroom.record_version(state, "submit")
        state["files"] = {"main.py": 'print("newer checkpoint")\n'}
        state["checkpoint_files"] = dict(state["files"])
        state["checkpoint_at"] = 200
        classroom.record_version(state, "checkpoint")
        classroom.save(state)

        old_published = classroom.PUBLISHED
        try:
            classroom.PUBLISHED = Path(self.temp_dir.name) / "empty-projects"
            result = classroom.progress("L6A", "student-1", "Alex")
        finally:
            classroom.PUBLISHED = old_published
        self.assertEqual(len(result["progress"]), 2)
        self.assertEqual(result["progress"][0]["source"], "checkpoint")
        self.assertIn("newer checkpoint", result["progress"][0]["files"]["main.py"])
        self.assertEqual(result["progress"][1]["source"], "submit")

        teacher_state = classroom.snapshot("L6A", "week02")[0]
        self.assertEqual(teacher_state["version_counts"], {"checkpoint": 1, "submit": 1})
        self.assertEqual(teacher_state["checkpoint_at"], 200)
        self.assertEqual(teacher_state["submitted_at"], 100)

    def test_class_zip_contains_each_students_final_project_folder(self):
        state = classroom.empty("L6A", "student-1", "Alex", "week02")
        state["project_name"] = "校園謎案"
        state["files"] = {"main.py": 'print("ok")\n', "data/case.json": "{}"}
        state["final_files"] = dict(state["files"])
        classroom.save(state)

        with zipfile.ZipFile(io.BytesIO(classroom.make_zip("L6A", "week02"))) as archive:
            names = set(archive.namelist())
        self.assertIn("2026Fall進階專題/Alex/week02_校園謎案/main.py", names)
        self.assertIn("2026Fall進階專題/Alex/week02_校園謎案/data/case.json", names)

    def test_course_and_week_templates_are_bundled(self):
        course = json.loads((classroom.DATA / "course.json").read_text(encoding="utf-8"))
        self.assertEqual(len(course["weeks"]), 18)
        self.assertEqual([item["lesson"] for item in course["weeks"]], list(range(1, 19)))
        self.assertEqual(set(classroom.template_files("week02")), {"main.py", "case_data.json"})
        lesson3 = next(item for item in course["weeks"] if item["id"] == "week03")
        lesson4 = next(item for item in course["weeks"] if item["id"] == "week04")
        self.assertEqual(lesson3["workspace_id"], "week02")
        self.assertEqual(lesson4["workspace_id"], "week02")
        self.assertFalse(lesson3["template"])
        with self.assertRaises(Exception):
            classroom.template_files("week03")

    def test_catalog_lists_courses_and_existing_students(self):
        state = classroom.empty("L6A", "student-1", "Alex", "week02")
        classroom.save(state)
        catalog = classroom.catalog()
        course = next(item for item in catalog["courses"] if item["class_code"] == "L6A")
        self.assertEqual(course["course_name"], "2026 Fall 進階專題")
        self.assertEqual(course["students"], [{"student_id": "student-1", "display_name": "Alex"}])

    def test_snapshot_includes_member_without_current_week_project(self):
        state = classroom.empty("L6A", "student-1", "Alex", "week01")
        classroom.save(state)
        students = classroom.snapshot("L6A", "week02")
        self.assertEqual(len(students), 1)
        self.assertEqual(students[0]["display_name"], "Alex")
        self.assertFalse(students[0]["connected"])

    def test_teacher_can_delete_student_and_all_classroom_progress(self):
        old_password = classroom.PASSWORD
        try:
            classroom.PASSWORD = "teacher-test"
            for week in ("week01", "week02"):
                classroom.save(classroom.empty("L6A", "student-1", "Alex", week))
            request = classroom.StudentDelete(
                class_code="L6A", student_id="student-1", password="teacher-test"
            )
            result = asyncio.run(classroom.delete_student(request))
            self.assertTrue(result["ok"])
            self.assertIn("warning", result)
        finally:
            classroom.PASSWORD = old_password
        course = next(item for item in classroom.catalog()["courses"] if item["class_code"] == "L6A")
        self.assertEqual(course["students"], [])
        self.assertEqual(classroom.snapshot("L6A", "week01"), [])
        self.assertEqual(classroom.snapshot("L6A", "week02"), [])

    def test_teacher_can_rename_student_without_changing_identity_or_work(self):
        old_password, old_token = classroom.PASSWORD, classroom.TOKEN
        try:
            classroom.PASSWORD = "teacher-test"
            classroom.TOKEN = ""
            state = classroom.empty("L6A", "student-1", "舊姓名", "week02")
            state["project_name"] = "校園謎案"
            state["files"] = {"main.py": 'print("目前進度")\n'}
            state["checkpoint_files"] = dict(state["files"])
            state["checkpoint_at"] = 100
            state["final_files"] = {"main.py": 'print("正式提交")\n'}
            state["submitted_at"] = 200
            classroom.record_version(state, "checkpoint")
            classroom.record_version(state, "submit")
            classroom.save(state)

            request = classroom.StudentUpdate(
                class_code="L6A", student_id="student-1",
                display_name="新姓名", password="teacher-test",
            )
            result = asyncio.run(classroom.update_student(request))
            self.assertTrue(result["ok"])
            self.assertEqual(result["student_id"], "student-1")

            member = classroom.catalog()["courses"]
            course = next(item for item in member if item["class_code"] == "L6A")
            self.assertEqual(course["students"], [{"student_id": "student-1", "display_name": "新姓名"}])
            restored = classroom.load("L6A", "student-1", "新姓名", "week02")
            self.assertEqual(restored["files"], state["files"])
            self.assertEqual(restored["final_files"], state["final_files"])
            submitted = next(item for item in classroom.submission_catalog() if item["class_code"] == "L6A")
            self.assertEqual(submitted["display_name"], "新姓名")
            with classroom.sqlite3.connect(classroom.DB) as db:
                names = db.execute(
                    "SELECT DISTINCT display_name FROM classroom_versions WHERE class_code=? AND student_id=?",
                    ("L6A", "student-1"),
                ).fetchall()
            self.assertEqual(names, [("新姓名",)])
            classroom.init_db()
            submitted = next(item for item in classroom.submission_catalog() if item["class_code"] == "L6A")
            self.assertEqual(submitted["display_name"], "新姓名")
        finally:
            classroom.PASSWORD, classroom.TOKEN = old_password, old_token

    def test_teacher_cannot_rename_student_to_an_existing_name(self):
        old_password, old_token = classroom.PASSWORD, classroom.TOKEN
        try:
            classroom.PASSWORD = "teacher-test"
            classroom.TOKEN = ""
            classroom.save(classroom.empty("L6A", "student-1", "Alex", "week02"))
            classroom.save(classroom.empty("L6A", "student-2", "Sam", "week02"))
            request = classroom.StudentUpdate(
                class_code="L6A", student_id="student-1",
                display_name=" sam ", password="teacher-test",
            )
            with self.assertRaises(classroom.HTTPException) as error:
                asyncio.run(classroom.update_student(request))
            self.assertEqual(error.exception.status_code, 409)
            self.assertEqual(
                classroom.find_member("L6A", "Alex", "student-1")["display_name"],
                "Alex",
            )
        finally:
            classroom.PASSWORD, classroom.TOKEN = old_password, old_token

    def test_current_roster_name_wins_over_older_snapshot_names_on_startup(self):
        old_paths = classroom.DATA, classroom.TEMPLATES, classroom.CHECKPOINTS, classroom.VERSIONS, classroom.REPAIRS
        data = Path(self.temp_dir.name) / "seeded-classroom-data"
        submission_root = data / "submissions" / "C1" / "student-1"
        version_root = data / "versions" / "C1" / "student-1" / "week02"
        submission_root.mkdir(parents=True)
        version_root.mkdir(parents=True)
        (data / "classes.json").write_text(json.dumps([{
            "code": "C1", "name": "測試班",
            "students": [{"student_id": "student-1", "display_name": "新姓名"}],
        }], ensure_ascii=False), encoding="utf-8")
        files = {"main.py": 'print("保留作品")\n'}
        old_submission = {
            "class_code": "C1", "student_id": "student-1", "display_name": "舊姓名",
            "week_id": "week02", "project_name": "校園謎案", "submitted_at": 100, "files": files,
        }
        old_version = {
            "class_code": "C1", "student_id": "student-1", "display_name": "舊姓名",
            "week_id": "week02", "version_id": "100-submit-one", "kind": "submit",
            "project_name": "校園謎案", "created_at": 100, "files": files,
        }
        (submission_root / "week02.json").write_text(json.dumps(old_submission, ensure_ascii=False), encoding="utf-8")
        (version_root / "100-submit-one.json").write_text(json.dumps(old_version, ensure_ascii=False), encoding="utf-8")
        try:
            classroom.DATA = data
            classroom.TEMPLATES = data / "templates"
            classroom.CHECKPOINTS = data / "checkpoints"
            classroom.VERSIONS = data / "versions"
            classroom.REPAIRS = data / "repairs"
            classroom.init_db()

            course = next(item for item in classroom.catalog()["courses"] if item["class_code"] == "C1")
            self.assertEqual(course["students"], [{"student_id": "student-1", "display_name": "新姓名"}])
            current_submission = next(
                item for item in classroom.submission_catalog()
                if item["class_code"] == "C1"
            )
            self.assertEqual(current_submission["display_name"], "新姓名")
            with classroom.sqlite3.connect(classroom.DB) as db:
                version_name = db.execute(
                    "SELECT display_name FROM classroom_versions WHERE class_code='C1' AND student_id='student-1'"
                ).fetchone()[0]
            self.assertEqual(version_name, "新姓名")
            restored = classroom.load("C1", "student-1", "新姓名", "week02")
            self.assertEqual(restored["final_files"], files)
        finally:
            classroom.DATA, classroom.TEMPLATES, classroom.CHECKPOINTS, classroom.VERSIONS, classroom.REPAIRS = old_paths

    def test_course_is_available_even_when_repo_persistence_is_unavailable(self):
        old_password, old_token = classroom.PASSWORD, classroom.TOKEN
        try:
            classroom.PASSWORD = "teacher-test"
            classroom.TOKEN = ""
            request = classroom.CourseCreate(
                class_code="L6B", course_name="週六進階班", password="teacher-test"
            )
            created = classroom.create_class(request)
            repeated = classroom.create_class(request)
        finally:
            classroom.PASSWORD, classroom.TOKEN = old_password, old_token
        self.assertTrue(created["ok"])
        self.assertIn("HF_TOKEN", created["warning"])
        self.assertTrue(repeated["already_exists"])
        course = next(item for item in classroom.catalog()["courses"] if item["class_code"] == "L6B")
        self.assertEqual(course["course_name"], "週六進階班")

    def test_teacher_creates_roster_and_student_types_matching_name(self):
        old_password, old_token = classroom.PASSWORD, classroom.TOKEN
        try:
            classroom.PASSWORD = "teacher-test"
            classroom.TOKEN = ""
            request = classroom.StudentCreate(
                class_code="L6A",
                display_names=["12 Alex", "王小明", "12 Alex"],
                password="teacher-test",
            )
            result = asyncio.run(classroom.create_students(request))
        finally:
            classroom.PASSWORD, classroom.TOKEN = old_password, old_token

        self.assertEqual([item["display_name"] for item in result["created"]], ["12 Alex", "王小明"])
        matched = classroom.student_login(
            classroom.StudentLogin(class_code="L6A", display_name="12 alex")
        )
        self.assertEqual(matched["display_name"], "12 Alex")
        with self.assertRaises(Exception):
            classroom.student_login(
                classroom.StudentLogin(class_code="L6A", display_name="不在名單")
            )

    def test_admin_can_rename_and_delete_classroom(self):
        old_password, old_token = classroom.PASSWORD, classroom.TOKEN
        try:
            classroom.PASSWORD = "teacher-test"
            classroom.TOKEN = ""
            classroom.create_class(classroom.CourseCreate(class_code="L6Z", course_name="舊班名", password="teacher-test"))
            updated = classroom.update_class(classroom.CourseUpdate(class_code="L6Z", course_name="新班名", password="teacher-test"))
            self.assertEqual(updated["course_name"], "新班名")
            deleted = asyncio.run(classroom.delete_class(classroom.CourseDelete(class_code="L6Z", password="teacher-test")))
            self.assertTrue(deleted["ok"])
        finally:
            classroom.PASSWORD, classroom.TOKEN = old_password, old_token
        self.assertFalse(any(item["class_code"] == "L6Z" for item in classroom.catalog()["courses"]))

    def test_public_course_catalog_does_not_expose_student_names(self):
        state = classroom.empty("L6A", "student-1", "Alex", "week02")
        classroom.save(state)
        public = classroom.public_catalog()
        self.assertNotIn("students", public["courses"][0])
        self.assertNotIn("Alex", json.dumps(public, ensure_ascii=False))

    def test_bundled_l6sat_backup_restores_students_and_submissions(self):
        course = next(item for item in classroom.catalog()["courses"] if item["class_code"] == "L6Sat")
        submissions = [item for item in classroom.submission_catalog() if item["class_code"] == "L6Sat"]
        self.assertEqual(len(course["students"]), 8)
        self.assertEqual(len(submissions), 8)
        restored = classroom.get_submission("L6Sat", "胡惟媗", "week02")
        self.assertEqual(restored["project_name"], "校園謎案調查局")
        self.assertEqual(set(restored["final_files"]), {"main.py", "case_data.json"})
        for submission in submissions:
            project = classroom.get_submission("L6Sat", submission["student_id"], "week02")
            compile(project["final_files"]["main.py"], submission["display_name"], "exec")

    def test_hf_failure_message_distinguishes_authentication_and_permission(self):
        unauthorized = RuntimeError("token hf_should_never_leak was rejected")
        unauthorized.response = SimpleNamespace(status_code=401)
        forbidden = RuntimeError("Forbidden")
        forbidden.response = SimpleNamespace(status_code=403)

        unauthorized_message = classroom.hf_failure_message(unauthorized)
        forbidden_message = classroom.hf_failure_message(forbidden)

        self.assertIn("HF_TOKEN 未被 Space 正確載入", unauthorized_message)
        self.assertNotIn("hf_should_never_leak", unauthorized_message)
        self.assertIn(classroom.REPO, forbidden_message)
        self.assertIn("寫入驗證被拒絕", forbidden_message)

    def test_published_final_can_be_loaded_after_runtime_restart(self):
        old_published = classroom.PUBLISHED
        try:
            classroom.PUBLISHED = Path(self.temp_dir.name) / "projects"
            project = classroom.PUBLISHED / classroom.DEMO_CLASS / "Alex" / "week02_校園謎案"
            project.mkdir(parents=True)
            (project / "main.py").write_text('print("已發布")\n', encoding="utf-8")
            progress = classroom.published_progress("L6A", "Alex")
        finally:
            classroom.PUBLISHED = old_published
        self.assertEqual(progress[0]["week_id"], "week02")
        self.assertEqual(progress[0]["source"], "published")
        self.assertIn("已發布", progress[0]["files"]["main.py"])

    def test_version_write_rate_limit_is_persisted_per_student_and_lesson(self):
        old_limit, old_cooldown = classroom.CHECKPOINTS_PER_HOUR, classroom.VERSION_WRITE_COOLDOWN_MS
        try:
            classroom.CHECKPOINTS_PER_HOUR = 1
            classroom.VERSION_WRITE_COOLDOWN_MS = 0
            classroom.consume_version_write("L6A", "student-1", "week02", "checkpoint")
            with self.assertRaises(classroom.VersionLimitError):
                classroom.consume_version_write("L6A", "student-1", "week02", "checkpoint")
            classroom.consume_version_write("L6A", "student-2", "week02", "checkpoint")
        finally:
            classroom.CHECKPOINTS_PER_HOUR, classroom.VERSION_WRITE_COOLDOWN_MS = old_limit, old_cooldown

    def test_version_retention_prunes_only_oldest_versions_of_same_kind(self):
        old_limit = classroom.MAX_VERSIONS_PER_KIND
        try:
            classroom.MAX_VERSIONS_PER_KIND = 2
            state = classroom.empty("L6A", "student-1", "Alex", "week02")
            for stamp in (100, 200, 300):
                state["files"] = {"main.py": f"print({stamp})\n"}
                state["checkpoint_files"] = dict(state["files"])
                state["checkpoint_at"] = stamp
                classroom.record_version(state, "checkpoint")
            state["final_files"] = {"main.py": "print('submission')\n"}
            state["submitted_at"] = 400
            classroom.record_version(state, "submit")
        finally:
            classroom.MAX_VERSIONS_PER_KIND = old_limit

        history = classroom.version_history("L6A", "student-1")
        checkpoints = [item for item in history if item["source"] == "checkpoint"]
        submissions = [item for item in history if item["source"] == "submit"]
        self.assertEqual([item["submitted_at"] for item in checkpoints], [300, 200])
        self.assertEqual(len(submissions), 1)

    def test_course_quota_rejects_before_inserting_version(self):
        old_quota = classroom.COURSE_STORAGE_QUOTA_BYTES
        try:
            classroom.COURSE_STORAGE_QUOTA_BYTES = 1
            state = classroom.empty("L6A", "student-1", "Alex", "week02")
            state["files"] = {"main.py": "print('too large')\n"}
            state["checkpoint_files"] = dict(state["files"])
            state["checkpoint_at"] = 123
            with self.assertRaises(classroom.VersionLimitError):
                classroom.record_version(state, "checkpoint")
        finally:
            classroom.COURSE_STORAGE_QUOTA_BYTES = old_quota
        self.assertEqual(classroom.version_history("L6A", "student-1"), [])

    def test_websocket_connection_and_message_budgets_are_shared_across_reconnects(self):
        old_connections, old_messages = classroom.STUDENT_WS_CONNECTIONS_PER_MINUTE, classroom.STUDENT_WS_MESSAGES_PER_MINUTE
        try:
            classroom.STUDENT_WS_CONNECTIONS_PER_MINUTE = 2
            classroom.STUDENT_WS_MESSAGES_PER_MINUTE = 2
            self.assertTrue(classroom.allow_student_connection("198.51.100.7"))
            self.assertTrue(classroom.allow_student_connection("198.51.100.7"))
            self.assertFalse(classroom.allow_student_connection("198.51.100.7"))
            key = ("L6A", "student-1", "week02")
            self.assertTrue(classroom.allow_student_message(key))
            self.assertTrue(classroom.allow_student_message(key))
            self.assertFalse(classroom.allow_student_message(key))
        finally:
            classroom.STUDENT_WS_CONNECTIONS_PER_MINUTE, classroom.STUDENT_WS_MESSAGES_PER_MINUTE = old_connections, old_messages


if __name__ == "__main__":
    unittest.main()