import json from pathlib import Path import tempfile import unittest from wbv2_preparation import prepare_module class CandidatePreparationTests(unittest.TestCase): def setUp(self): self.temporary = tempfile.TemporaryDirectory() self.addCleanup(self.temporary.cleanup) self.root = Path(self.temporary.name) self.directory = self.root / "tests/golden/WBV2/autopilot" self.job = {"id": "AUDIT", "commands": ["setAuditFixture"], "ownedPaths": [], "module": "audit"} self.candidate = self.root / "web/tests/e2e/audit.spec.ts" self.candidate.parent.mkdir(parents=True) self.candidate.write_text("setAuditFixture before") self.unrelated = self.candidate.with_name("unrelated.spec.ts") self.unrelated.write_text("unrelated before") inventory = self.root / "tests/golden/WBV2/dynamic-inventory.json" inventory.parent.mkdir(parents=True) inventory.write_text('{"units": []}') def prepare(self): relative = prepare_module(self.root, self.directory, self.job) path = self.root / relative return path, json.loads(path.read_text()) def test_existing_candidate_content_invalidates_cached_preparation(self): import os from unittest.mock import patch import subprocess path, first = self.prepare() self.assertIn("web/tests/e2e/audit.spec.ts", first["candidateTests"]) directory_time = self.candidate.parent.stat().st_mtime_ns previous_time = self.candidate.stat().st_mtime_ns self.candidate.write_text("setAuditFixture after!") os.utime(self.candidate, ns=(previous_time + 1000000000, previous_time + 1000000000)) self.assertEqual(directory_time, self.candidate.parent.stat().st_mtime_ns) with patch("wbv2_preparation.subprocess.run", wraps=subprocess.run) as search: _, changed = self.prepare() self.assertEqual(search.call_count, 1) self.assertNotEqual(first["fingerprint"], changed["fingerprint"]) saved_time = path.stat().st_mtime_ns with patch("wbv2_preparation.subprocess.run", side_effect=AssertionError("unchanged inputs were rescanned")): _, reused = self.prepare() self.assertEqual(changed, reused) self.assertEqual(saved_time, path.stat().st_mtime_ns) def test_unrelated_file_edit_keeps_preparation_reusable(self): from unittest.mock import patch path, first = self.prepare() saved_time = path.stat().st_mtime_ns self.unrelated.write_text("unrelated after!") with patch("wbv2_preparation.subprocess.run", side_effect=AssertionError("unrelated input was scanned")): _, reused = self.prepare() self.assertEqual(first, reused) self.assertEqual(saved_time, path.stat().st_mtime_ns) def test_candidate_paths_reject_escape_symlinks_and_non_test_files(self): from wbv2_preparation import candidate_test_paths outside = self.root / "outside.spec.ts" outside.write_text("setAuditFixture") linked = self.candidate.with_name("linked.spec.ts") linked.symlink_to(outside) linked_directory = self.candidate.parent / "linked-directory" linked_directory.symlink_to(self.root, target_is_directory=True) candidates = ["web/tests/e2e/audit.spec.ts", "../outside.spec.ts", str(outside), "outside.spec.ts", "web/tests/e2e/linked.spec.ts", "web/tests/e2e/linked-directory/outside.spec.ts", "web/tests/e2e", "web/tests/e2e/../audit.spec.ts", None, "web/tests/e2e/invalid\x00.spec.ts"] self.assertEqual(candidate_test_paths(self.root, candidates), ("web/tests/e2e/audit.spec.ts",)) self.assertEqual(candidate_test_paths(self.root, [None] * 12 + candidates), ()) class BusinessBlockerTests(unittest.TestCase): setUp = CandidatePreparationTests.setUp prepare = CandidatePreparationTests.prepare def test_blocker_classification_tracks_source_test_and_contract_gaps(self): _, missing_source = self.prepare() self.assertEqual(missing_source["businessBlocker"]["category"], "source_binding_missing") self.assertIsNone(missing_source["businessBlocker"]["sourceSearch"]) inventory = self.root / "tests/golden/WBV2/dynamic-inventory.json" inventory.write_text(json.dumps({"units": [{"id": "web-command:setAuditFixture", "locations": []}]})) self.candidate.unlink() _, missing_test = self.prepare() self.assertEqual(missing_test["businessBlocker"]["category"], "behavioral_test_missing") self.candidate.write_text("setAuditFixture restored") _, missing_contract = self.prepare() blocker = missing_contract["businessBlocker"] self.assertEqual(blocker["category"], "workflow_contract_missing") self.assertEqual(missing_contract["state"], "needs_implementation") self.assertEqual(blocker["contractPath"], "my-tools/local-workflows/AUDIT.json") self.assertEqual(blocker["candidateTests"], ["web/tests/e2e/audit.spec.ts"]) self.assertEqual(blocker["sourceSearch"]["argv"], [ "node", "tools/web/search-wbv2-batch-source.mjs", "--unit", "web-command:setAuditFixture", "--pattern", "setAuditFixture", "--max-hits", "8"]) actions = missing_contract["nextActions"] self.assertEqual(actions[0]["kind"], "inspect_source") self.assertEqual(actions[-1]["path"], blocker["contractPath"]) def test_input_timestamps_stay_stable_until_fingerprint_changes(self): from unittest.mock import patch with patch("wbv2_autopilot.now", return_value="2026-09-10T01:00:00Z"): path, first = self.prepare() timestamp = path.stat().st_mtime_ns with patch("wbv2_autopilot.now", side_effect=AssertionError("unchanged preparation updated time")): _, repeated = self.prepare() self.assertEqual(first, repeated) self.assertEqual(timestamp, path.stat().st_mtime_ns) self.candidate.write_text("setAuditFixture changed") with patch("wbv2_autopilot.now", return_value="2026-09-10T02:00:00Z"): _, changed = self.prepare() blocker = changed["businessBlocker"] self.assertEqual(blocker["firstSeenAt"], "2026-09-10T01:00:00Z") self.assertEqual(blocker["inputChangedAt"], "2026-09-10T02:00:00Z") self.assertNotEqual(first["fingerprint"], blocker["inputFingerprint"]) self.assertNotIn("lastProgressAt", blocker) def test_blocker_commands_and_retry_paths_are_bounded_and_local(self): from wbv2_preparation import _business_blocker outside = self.root / "outside.cpp" outside.write_text("local") linked = self.candidate.with_name("linked.spec.ts") linked.symlink_to(outside) report = {"job": "AUDIT", "fingerprint": "fixture", "commands": ["setAuditFixture", "bad;command"] + ["command" + str(index) for index in range(60)], "sourceUnits": [{"id": "bad unit;command"}, {"id": "web-command:setAuditFixture", "locations": [ {"path": "../outside.cpp"}, {"path": "web/tests/e2e/linked.spec.ts"}, {"path": "blender-5.2.0/source/blender/web_engine/local.cpp"}]}], "candidateTests": ["web/tests/e2e/audit.spec.ts", "../outside.cpp", "web/tests/e2e/linked.spec.ts"], "missingSourceBindings": [], "ownedInputs": {"web/app/src/input" + str(index) + ".ts": "fixture" for index in range(80)}} report["ownedInputs"]["../outside.cpp"] = "fixture" _business_blocker(self.root, report, {}) blocker = report["businessBlocker"] self.assertLessEqual(len(blocker["retryOn"]), 64) self.assertIn("my-tools/local-workflows/AUDIT.json", blocker["retryOn"]) self.assertIn("web/tests/e2e/audit.spec.ts", blocker["retryOn"]) self.assertNotIn("../outside.cpp", blocker["retryOn"]) self.assertNotIn("web/tests/e2e/linked.spec.ts", blocker["retryOn"]) self.assertEqual(blocker["sourceSearch"]["argv"][3], "web-command:setAuditFixture") for action in report["nextActions"]: self.assertLessEqual(len(action.get("commands", [])), 32) self.assertNotIn("bad;command", action.get("commands", [])) def test_invalid_job_path_is_rejected_before_preparation(self): from unittest.mock import patch for identifier in ("../escape", "/tmp/escape", "A" * 129): with self.subTest(identifier=identifier): with patch("wbv2_preparation.subprocess.run", side_effect=AssertionError("invalid job reached search")): with self.assertRaises(ValueError): prepare_module(self.root, self.directory, {**self.job, "id": identifier}) class PreparationTests(unittest.TestCase): def setUp(self): self.temporary = tempfile.TemporaryDirectory() self.addCleanup(self.temporary.cleanup) self.root = Path(self.temporary.name) self.directory = self.root / "tests/golden/WBV2/autopilot" (self.root / "source.cpp").write_text("// source fixture\n") self.job = {"id": "LOCAL-MODULE", "commands": ["readExample"], "ownedPaths": ["source.cpp"], "module": "EXAMPLE"} def test_preparation_is_not_a_completion_claim(self): path = prepare_module(self.root, self.directory, self.job) report = json.loads((self.root / path).read_text()) self.assertEqual("needs_implementation", report["state"]) self.assertEqual(["readExample"], report["missingSourceBindings"]) self.assertNotIn("result", report) def test_unchanged_sources_do_not_repeat_preparation(self): path = self.root / prepare_module(self.root, self.directory, self.job) before = path.stat().st_mtime_ns prepare_module(self.root, self.directory, self.job) self.assertEqual(before, path.stat().st_mtime_ns) (self.root / "source.cpp").write_text("// updated source fixture\n") prepare_module(self.root, self.directory, self.job) self.assertNotEqual(before, path.stat().st_mtime_ns) def test_owned_path_escape_is_rejected(self): self.job["ownedPaths"] = ["../outside.cpp"] with self.assertRaises(ValueError): prepare_module(self.root, self.directory, self.job)