Spaces:
Running
Running
| 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() | |