from __future__ import annotations import json from pathlib import Path import unittest ROOT = Path(__file__).resolve().parents[2] FIXTURE = ROOT / "tests/fixtures/ignored-workspace-v1" def load(name: str) -> dict: with (FIXTURE / name).open("rb") as source: return json.load(source) class IgnoredWorkspaceContractTests(unittest.TestCase): def test_contract_dispatch_and_selected_mechanism_are_exact(self) -> None: contract = load("contract.json") self.assertEqual(contract["schema"], "lazaret-ignored-workspace-contract-v1") self.assertEqual(contract["mechanism"], "rootless-host-fuse-router") self.assertEqual(contract["route_classes"], ["host", "isolated"]) def test_cross_route_namespace_operations_refuse_before_mutation(self) -> None: transitions = { transition["id"]: transition for transition in load("contract.json")["transition_contracts"] } for transition_id in ( "cross-route-rename", "rename-whiteout", "cross-route-hard-link", ): with self.subTest(transition_id=transition_id): self.assertEqual(transitions[transition_id]["mutation"], "none") self.assertEqual(transitions["cross-route-rename"]["result"], "EXDEV") self.assertEqual(transitions["cross-route-hard-link"]["result"], "EXDEV") self.assertEqual(transitions["rename-whiteout"]["result"], "EOPNOTSUPP") def test_transitions_never_copy_or_rebind_open_handles(self) -> None: transitions = load("contract.json")["transition_contracts"] copying = [ transition["id"] for transition in transitions if "copy" in transition and transition["copy"] != "none" ] self.assertEqual(copying, []) handle_transitions = { transition["id"]: transition["open_handles"] for transition in transitions if "open_handles" in transition } self.assertEqual( handle_transitions, { "ignored-to-host-policy": "remain-isolated", "host-to-ignored-policy": "remain-host", "isolated-to-tracked-host-established": "remain-isolated", }, ) def test_isolated_only_force_add_is_explicitly_refused(self) -> None: contract = load("contract.json") transitions = { transition["id"]: transition for transition in contract["transition_contracts"] } transition = transitions["isolated-to-tracked-guest-publication"] self.assertEqual(transition["result"], "EBUSY") self.assertEqual(transition["index_mutation"], "none") compatibility = { operation["operation"]: operation["support"] for operation in contract["git_compatibility"] } self.assertEqual( compatibility["force-add-isolated-only"], "EBUSY-before-index-publication", ) def test_operation_matrix_covers_required_surface(self) -> None: operations = { operation["operation"]: operation for operation in load("contract.json")["filesystem_operations"] } required = { "lookup-getattr-readdir", "create-mkdir-open-read-write-truncate-fallocate-fsync", "rename", "rename-noreplace", "rename-exchange", "rename-whiteout", "hard-link", "symlink-readlink", "user-xattrs", "posix-acls", "sparse-seek-hole-data-punch-hole", "copy-file-range", "utimens-nanoseconds", "mmap-shared-private", "flock-posix-record-lock", "fifo-unix-socket", "character-block-device-create", } self.assertTrue(required.issubset(operations)) self.assertEqual(operations["rename"]["cross_route"], "EXDEV") self.assertEqual(operations["hard-link"]["cross_route"], "EXDEV") self.assertEqual(operations["copy-file-range"]["cross_route"], "EXDEV") self.assertEqual(operations["character-block-device-create"]["host"], "EPERM") def test_corpus_ids_and_byte_paths_are_canonical(self) -> None: cases = load("corpus.json")["cases"] ids = [case["id"] for case in cases] self.assertEqual(len(ids), len(set(ids))) for case in cases: with self.subTest(case=case["id"]): path_hex = case["path_hex"] self.assertEqual(path_hex, path_hex.lower()) self.assertEqual(bytes.fromhex(path_hex).hex(), path_hex) self.assertNotIn(b"\x00", bytes.fromhex(path_hex)) non_utf8 = next(case for case in cases if case["id"] == "non-utf8-ignore-match") with self.assertRaises(UnicodeDecodeError): bytes.fromhex(non_utf8["path_hex"]).decode("utf-8") def test_corpus_covers_every_required_behavior(self) -> None: cases = load("corpus.json")["cases"] tags = {tag for case in cases for tag in case["tags"]} required = { "non-utf8-names", "ignore-precedence", "core-ignore-case", "worktrees", "nested-repositories", "mmap", "locks", "xattrs", "acls", "sparse", "copy-range", "rename", "link", "symlink", "timestamps", "special-files", "concurrent-mutation", "cross-policy-rename", "ignored-to-tracked-transition", "open-handle-transition", "no-silent-loss", "no-copy-back", } self.assertTrue(required.issubset(tags), sorted(required - tags)) def test_corpus_asserts_backing_state_for_refusal_and_loss_cases(self) -> None: cases = {case["id"]: case for case in load("corpus.json")["cases"]} self.assertEqual( cases["cross-route-file-rename"]["expect"], ["EXDEV", "source-unchanged", "destination-absent"], ) self.assertIn("isolated-object-recoverable", cases["unmediated-inconsistent-index"]["expect"]) self.assertIn("access-error-not-host-fallback", cases["router-loss-fails-closed"]["expect"]) self.assertIn("host-sentinel-unchanged", cases["router-loss-fails-closed"]["expect"]) def test_benchmark_reference_conditions_are_complete(self) -> None: reference = load("benchmark.json")["reference_environment"] required = { "architecture", "kernel", "cpu", "cpu_set", "cpu_governor", "memory_bytes", "swap", "storage", "storage_stack", "filesystem", "filesystem_block_bytes", "filesystem_mount_options", "fuse_userspace", "runsc", "git", "uv", "rust", } self.assertEqual(set(reference), required) self.assertEqual(reference["architecture"], "x86_64") self.assertEqual(reference["filesystem"], "XFS") self.assertGreaterEqual(reference["memory_bytes"], 16 * 1024**3) def test_sampling_protocol_fixes_statistics_and_variance(self) -> None: sampling = load("benchmark.json")["sampling"] self.assertEqual(sampling["independent_runs"], 3) self.assertEqual(sampling["warmup_samples_per_cell"], 2) self.assertEqual(sampling["measured_samples_per_cell"], 15) self.assertEqual(sampling["statistic"], "median") self.assertEqual( sampling["relative_statistic"], "median of paired routed/direct elapsed-time ratios", ) self.assertEqual(sampling["outlier_policy"], "retain every completed measured sample") self.assertLessEqual(sampling["within_run_relative_mad_limit"], 0.10) self.assertLessEqual(sampling["between_run_median_spread_limit"], 0.10) self.assertEqual(sampling["budget_tolerance"], 0.0) def test_each_workload_has_absolute_and_relative_budgets(self) -> None: benchmark = load("benchmark.json") workloads = benchmark["workloads"] self.assertEqual( {workload["category"] for workload in workloads}, {"metadata", "git", "package-manager", "build"}, ) for workload in workloads: with self.subTest(workload=workload["id"]): self.assertGreater(workload["absolute_median_ms"], 0) self.assertGreater(workload["relative_median_ratio"], 1.0) self.assertEqual(workload["preparation_states"], ["unprimed", "primed"]) self.assertEqual(len(workload["fixture_seed"]), 64) self.assertTrue(workload["fixture_algorithm"]) self.assertTrue(workload["command_sequence"]) self.assertTrue(all(command for command in workload["command_sequence"])) self.assertTrue(workload["result_oracle"]) def test_specification_names_the_committed_contract(self) -> None: specification = (ROOT / "docs/specification.md").read_text() self.assertIn("#### Ignored Workspace State routing contract", specification) self.assertIn("`lazaret-ignored-workspace-contract-v1`", specification) self.assertIn("rootless host FUSE filesystem", specification) if __name__ == "__main__": unittest.main()