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