import json from pathlib import Path import sys import tempfile import unittest from unittest.mock import patch sys.path.insert(0, str(Path(__file__).resolve().parent)) import run_freecad_web_scheduler as scheduler from scheduler_context import compact_packet, source_snapshot, tracked_edit_paths, SourceChangeTracker from prepare_module_implementation_packet import compact_excerpt, instruction_context class ContextTests(unittest.TestCase): def test_business_recovery_method_survives_context_compaction_without_full_report(self): context = instruction_context([]) self.assertEqual(context["geometryBoundary"], "docs/OCCT_GEOMETRY_BOUNDARY.md") policy = context["businessRecovery"] self.assertEqual(policy["workedExample"], "docs/current/part-mesh-business-unblock.md") self.assertEqual(len(policy["steps"]), 6) self.assertLess(len(json.dumps(policy).encode()), 1500) batch = {"batchId": "batch", "selected": {"module": "Mesh"}, "records": [["Mesh:save"]]} packet = {"repositoryContext": context, "editableExcerpts": [{"path": "source.cpp", "excerpt": "x" * 40000}]} result = compact_packet(batch, packet, {}, 12000, "packet.json") payload = json.loads(result.split("\n", 2)[-1]) self.assertEqual(payload["repositoryContext"]["businessRecovery"], policy) self.assertEqual(payload["repositoryContext"]["geometryBoundary"], context["geometryBoundary"]) self.assertLessEqual(len(result.encode()), 12000) def test_compaction_keeps_identity_contract_and_whole_excerpts(self): batch = {"batchId": "batch", "selected": {"module": "Part"}, "records": [["Part:save"]]} small = {"path": "save.cpp", "line": 8, "excerpt": "完整行();"} large = {"path": "huge.cpp", "line": 1, "excerpt": "x" * 40000} packet = {"acceptanceSetup": {"missingIds": ["Part:save"]}, "editableExcerpts": [large, small]} result = compact_packet(batch, packet, {}, 12000, "packet.json") payload = json.loads(result.split("\n", 2)[-1]) self.assertLessEqual(len(result.encode()), 12000) self.assertEqual(payload["recordIds"], ["Part:save"]) self.assertEqual(payload["acceptanceSetup"]["missingIds"], ["Part:save"]) self.assertEqual(payload["editableExcerpts"], [small]) self.assertEqual(payload["omittedEditableExcerpts"], 1) def test_overflow_build_uses_structured_packet(self): batch = {"batchId": "batch", "selected": {"module": "Part"}, "records": [["Part:save"]]} packet = {"batchId": "batch"} with patch.object(scheduler, "_build_prompt_unbounded", return_value="x" * 50000), patch.object(scheduler, "read_progress", side_effect=[batch, packet, {}]): result = scheduler.build_prompt("active") self.assertEqual(json.loads(result.split("\n", 2)[-1])["recordIds"], ["Part:save"]) def test_legacy_bound_does_not_split_single_line_json(self): huge = json.dumps({"source": "中" * 20000}, ensure_ascii=False) result = scheduler.bound_prompt_context("rules\n" + huge + "\nbatchId=current\n", 12000) self.assertIn("batchId=current", result) self.assertNotIn('"source"', result) def test_source_excerpt_preserves_lines(self): self.assertEqual(compact_excerpt("first();\n" + "x" * 200, 30), "first();") self.assertEqual(compact_excerpt("x" * 200, 30), "") def test_measurement_separates_aggregate_from_request_context(self): with tempfile.TemporaryDirectory() as directory, patch.object(scheduler, "SESSION_CACHE_PATH", Path(directory) / "session.json"), patch.object(scheduler, "read_progress", return_value={}), patch.object(scheduler, "SESSION_INPUT_TOKEN_LIMIT", 200000): scheduler.save_session_cache("Part", "batch", {"threadId": "t", "usage": {"input_tokens": 1530000}, "requestUsage": {"complete": True, "lastInputTokens": 80000, "peakInputTokens": 80000}}, None, True) session = scheduler.load_session_cache() self.assertEqual(session["lastInputTokens"], 80000) self.assertEqual(session["aggregateInputTokens"], 1530000) self.assertEqual(session["contextMeasurement"], "per-request") self.assertEqual(session["predictedNextInputTokens"], 160000) self.assertIsNotNone(scheduler.resumable_session("Part")) def test_missing_request_usage_uses_conservative_fallback(self): with tempfile.TemporaryDirectory() as directory, patch.object(scheduler, "SESSION_CACHE_PATH", Path(directory) / "session.json"), patch.object(scheduler, "read_progress", return_value={}): scheduler.save_session_cache("Part", "batch", {"threadId": "t", "usage": {"input_tokens": 1000000}, "requestUsage": {"complete": False, "lastInputTokens": 20000}}, None, True) self.assertEqual(scheduler.load_session_cache()["contextMeasurement"], "aggregate-fallback") self.assertIsNone(scheduler.resumable_session("Part")) def test_post_edit_snapshot_invalidates_external_changes(self): with tempfile.TemporaryDirectory() as directory: root = Path(directory) source = root / "native.cpp" source.write_text("after own edit") packet = {"batchId": "batch", "module": "Part", "supportFiles": [{"path": "native.cpp"}]} with patch.object(scheduler, "ROOT", root), patch.object(scheduler, "SESSION_CACHE_PATH", root / "session.json"), patch.object(scheduler, "read_progress", return_value=packet): scheduler.save_session_cache("Part", "batch", {"threadId": "t"}, None, True) self.assertIsNotNone(scheduler.resumable_session("Part", "batch", same_batch_only=True)) source.write_text("external edit") self.assertIsNone(scheduler.resumable_session("Part", "batch", same_batch_only=True)) def test_snapshot_checks_official_source_and_rejects_escape(self): with tempfile.TemporaryDirectory() as directory: root = Path(directory) (root / "source.cpp").write_text("source") result = source_snapshot(root, {"records": [{"source": "source.cpp"}], "supportFiles": [{"path": "../outside"}]}) self.assertEqual(list(result), ["source.cpp"]) def test_tracker_ignores_touch_but_detects_content_creation_and_deletion(self): with tempfile.TemporaryDirectory() as directory: source = Path(directory) / "source.cpp" source.write_text("before") tracker = SourceChangeTracker([source, source]) self.assertEqual(tracker.poll(), []) source.touch() self.assertEqual(tracker.poll(), []) source.write_text("after!") self.assertEqual(tracker.poll(), [str(source)]) self.assertEqual(tracker.poll(), []) source.unlink() self.assertEqual(tracker.poll(), [str(source)]) source.write_text("new") self.assertEqual(tracker.poll(), [str(source)]) self.assertGreater(tracker.seconds, 0) def test_tracker_scope_excludes_generated_evidence_and_outside_paths(self): with tempfile.TemporaryDirectory() as directory: root = Path(directory) names = ["native.cpp", "CMakeLists.txt", "web/src/facade.ts", "docs/business-acceptance/Part.json", "docs/current/status.json", "kernel.wasm", "../outside.cpp"] packet = {"supportFiles": [{"path": name} for name in names]} self.assertEqual(set(tracked_edit_paths(root, packet)), {root / name for name in names[:4]}) if __name__ == "__main__": unittest.main()