| 1 | #!/usr/bin/env python3 |
| 2 | """Hermetic tests for the FEAT-015 command migration manifest gate. |
| 3 | |
| 4 | Covers the Deep-Dive schema and frontier rules: |
| 5 | |
| 6 | - valid root frontier; valid parent-to-all-children replacement; valid removal |
| 7 | - rejected partial split; rejected arbitrary addition |
| 8 | - duplicate/unsorted entry; unknown schema/tag/field; unsupported syntax |
| 9 | - missing/overlapping selector; const normalization |
| 10 | - integer/character/byte/path const atoms; out-of-range byte diagnostics |
| 11 | """ |
| 12 | |
| 13 | from __future__ import annotations |
| 14 | |
| 15 | import importlib.util |
| 16 | import json |
| 17 | import subprocess |
| 18 | import sys |
| 19 | import tempfile |
| 20 | import unittest |
| 21 | from pathlib import Path |
| 22 | |
| 23 | ROOT = Path(__file__).resolve().parents[1] |
| 24 | SCRIPT = ROOT / "scripts" / "check-command-migration-manifest.py" |
| 25 | SPEC = importlib.util.spec_from_file_location("command_migration", SCRIPT) |
| 26 | assert SPEC and SPEC.loader |
| 27 | mod = importlib.util.module_from_spec(SPEC) |
| 28 | sys.modules[SPEC.name] = mod |
| 29 | SPEC.loader.exec_module(mod) |
| 30 | |
| 31 | |
| 32 | def sample_topology() -> dict: |
| 33 | """A minimal two-group topology with one documented slice per group.""" |
| 34 | return { |
| 35 | "schema_version": 1, |
| 36 | "topology": { |
| 37 | "utility": { |
| 38 | "kind": "group", |
| 39 | "scope": ["crates/tui/src/commands/groups/utility/mod.rs"], |
| 40 | "slices": [], |
| 41 | }, |
| 42 | "session": { |
| 43 | "kind": "group", |
| 44 | "scope": ["crates/tui/src/commands/groups/session/mod.rs"], |
| 45 | "slices": [ |
| 46 | { |
| 47 | "name": "session::lifecycle", |
| 48 | "kind": "slice", |
| 49 | "scope": ["crates/tui/src/commands/groups/session/branch.rs"], |
| 50 | }, |
| 51 | { |
| 52 | "name": "session::control", |
| 53 | "kind": "slice", |
| 54 | "scope": ["crates/tui/src/commands/groups/session/relay.rs"], |
| 55 | }, |
| 56 | ], |
| 57 | }, |
| 58 | }, |
| 59 | "frontier": ["session", "utility"], |
| 60 | } |
| 61 | |
| 62 | |
| 63 | class SchemaTests(unittest.TestCase): |
| 64 | def test_valid_document_passes(self) -> None: |
| 65 | self.assertEqual(mod.validate_topology_document(sample_topology()), []) |
| 66 | |
| 67 | def test_unknown_schema_version_fails(self) -> None: |
| 68 | doc = sample_topology() |
| 69 | doc["schema_version"] = 2 |
| 70 | violations = mod.validate_topology_document(doc) |
| 71 | self.assertEqual(len(violations), 1) |
| 72 | self.assertIn("schema_version", str(violations[0])) |
| 73 | |
| 74 | def test_missing_schema_version_fails(self) -> None: |
| 75 | doc = sample_topology() |
| 76 | del doc["schema_version"] |
| 77 | self.assertTrue(mod.validate_topology_document(doc)) |
| 78 | |
| 79 | def test_unknown_group_field_fails(self) -> None: |
| 80 | doc = sample_topology() |
| 81 | doc["topology"]["utility"]["extra"] = True |
| 82 | violations = mod.validate_topology_document(doc) |
| 83 | self.assertTrue(any("extra" in str(v) for v in violations)) |
| 84 | |
| 85 | def test_slice_without_group_prefix_fails(self) -> None: |
| 86 | doc = sample_topology() |
| 87 | doc["topology"]["session"]["slices"][0]["name"] = "other::slice" |
| 88 | violations = mod.validate_topology_document(doc) |
| 89 | self.assertTrue(any("must start with" in str(v) for v in violations)) |
| 90 | |
| 91 | def test_duplicate_slice_name_fails(self) -> None: |
| 92 | doc = sample_topology() |
| 93 | doc["topology"]["session"]["slices"].append( |
| 94 | doc["topology"]["session"]["slices"][0] |
| 95 | ) |
| 96 | violations = mod.validate_topology_document(doc) |
| 97 | self.assertTrue(any("duplicate slice" in str(v) for v in violations)) |
| 98 | |
| 99 | |
| 100 | class FrontierTests(unittest.TestCase): |
| 101 | def test_valid_root_frontier_passes(self) -> None: |
| 102 | doc = sample_topology() |
| 103 | self.assertEqual(mod.validate_frontier(doc["topology"], doc["frontier"]), []) |
| 104 | |
| 105 | def test_unsorted_frontier_fails(self) -> None: |
| 106 | doc = sample_topology() |
| 107 | violations = mod.validate_frontier(doc["topology"], ["utility", "session"]) |
| 108 | self.assertTrue(any("sorted" in str(v) for v in violations)) |
| 109 | |
| 110 | def test_duplicate_frontier_entry_fails(self) -> None: |
| 111 | doc = sample_topology() |
| 112 | violations = mod.validate_frontier(doc["topology"], ["session", "session"]) |
| 113 | self.assertTrue(any("duplicates" in str(v) for v in violations)) |
| 114 | |
| 115 | def test_unknown_frontier_leaf_fails(self) -> None: |
| 116 | doc = sample_topology() |
| 117 | violations = mod.validate_frontier(doc["topology"], ["session", "ghost"]) |
| 118 | self.assertTrue(any("not a declared topology leaf" in str(v) for v in violations)) |
| 119 | |
| 120 | def test_removal_is_valid_transition(self) -> None: |
| 121 | doc = sample_topology() |
| 122 | old = ["session", "utility"] |
| 123 | new = ["utility"] |
| 124 | self.assertEqual(mod.is_valid_frontier_transition(doc["topology"], old, new), []) |
| 125 | |
| 126 | def test_parent_to_all_children_is_valid_transition(self) -> None: |
| 127 | doc = sample_topology() |
| 128 | old = ["session", "utility"] |
| 129 | new = ["session::control", "session::lifecycle", "utility"] |
| 130 | self.assertEqual(mod.is_valid_frontier_transition(doc["topology"], old, new), []) |
| 131 | |
| 132 | def test_partial_split_is_rejected(self) -> None: |
| 133 | doc = sample_topology() |
| 134 | old = ["session", "utility"] |
| 135 | new = ["session::lifecycle", "utility"] |
| 136 | violations = mod.is_valid_frontier_transition(doc["topology"], old, new) |
| 137 | self.assertTrue(violations, "partial split must be rejected") |
| 138 | |
| 139 | def test_arbitrary_addition_is_rejected(self) -> None: |
| 140 | doc = sample_topology() |
| 141 | old = ["session", "utility"] |
| 142 | new = ["ghost", "session", "utility"] |
| 143 | violations = mod.is_valid_frontier_transition(doc["topology"], old, new) |
| 144 | self.assertTrue(violations, "arbitrary growth must be rejected") |
| 145 | |
| 146 | |
| 147 | class LiveTransitionTests(unittest.TestCase): |
| 148 | def test_invalid_current_manifest_fails_closed_before_transition(self) -> None: |
| 149 | violations = mod.validate_baseline_transition({}, None) |
| 150 | self.assertTrue(any("current topology is invalid" in str(v) for v in violations)) |
| 151 | |
| 152 | def test_initial_manifest_must_start_at_all_roots(self) -> None: |
| 153 | doc = sample_topology() |
| 154 | self.assertEqual(mod.validate_baseline_transition(doc, None), []) |
| 155 | doc["frontier"] = ["utility"] |
| 156 | violations = mod.validate_baseline_transition(doc, None) |
| 157 | self.assertTrue(any("first manifest revision" in str(v) for v in violations)) |
| 158 | |
| 159 | def test_live_transition_rejects_topology_mutation(self) -> None: |
| 160 | previous = sample_topology() |
| 161 | current = json.loads(json.dumps(previous)) |
| 162 | current["topology"]["utility"]["scope"].append("new.rs") |
| 163 | violations = mod.validate_baseline_transition(current, previous) |
| 164 | self.assertTrue(any("topology is immutable" in str(v) for v in violations)) |
| 165 | |
| 166 | def test_live_transition_accepts_documented_split(self) -> None: |
| 167 | previous = sample_topology() |
| 168 | current = json.loads(json.dumps(previous)) |
| 169 | current["frontier"] = ["session::control", "session::lifecycle", "utility"] |
| 170 | self.assertEqual(mod.validate_baseline_transition(current, previous), []) |
| 171 | |
| 172 | def test_live_transition_rejects_arbitrary_growth(self) -> None: |
| 173 | previous = sample_topology() |
| 174 | previous["frontier"] = ["utility"] |
| 175 | current = json.loads(json.dumps(previous)) |
| 176 | current["frontier"] = ["session", "utility"] |
| 177 | violations = mod.validate_baseline_transition(current, previous) |
| 178 | self.assertTrue(any("arbitrary growth" in str(v) for v in violations)) |
| 179 | |
| 180 | def test_pending_projection_must_equal_json_frontier(self) -> None: |
| 181 | doc = sample_topology() |
| 182 | self.assertEqual( |
| 183 | mod.validate_pending_projection(doc, ["session", "utility"]), [] |
| 184 | ) |
| 185 | violations = mod.validate_pending_projection(doc, ["utility"]) |
| 186 | self.assertTrue(any("does not equal JSON frontier" in str(v) for v in violations)) |
| 187 | |
| 188 | def test_pending_groups_parser_is_string_only_and_order_preserving(self) -> None: |
| 189 | with tempfile.TemporaryDirectory() as directory: |
| 190 | path = Path(directory) / "contract.rs" |
| 191 | path.write_text( |
| 192 | 'pub(crate) const PENDING_GROUPS: &[&str] = &["session", "utility"];\n', |
| 193 | encoding="utf-8", |
| 194 | ) |
| 195 | self.assertEqual(mod.load_pending_groups(path), ["session", "utility"]) |
| 196 | |
| 197 | |
| 198 | class GitBaselineTests(unittest.TestCase): |
| 199 | def setUp(self) -> None: |
| 200 | directory = tempfile.TemporaryDirectory() |
| 201 | self.addCleanup(directory.cleanup) |
| 202 | self.root = Path(directory.name) |
| 203 | self.git("init", "-q", "--template=", "-b", "main") |
| 204 | self.git("config", "user.name", "Manifest test") |
| 205 | self.git("config", "user.email", "manifest@example.invalid") |
| 206 | self.git("config", "commit.gpgsign", "false") |
| 207 | |
| 208 | def git(self, *args: str) -> str: |
| 209 | return subprocess.check_output( |
| 210 | ["git", *args], cwd=self.root, text=True, stderr=subprocess.PIPE, |
| 211 | ).strip() |
| 212 | |
| 213 | def commit(self, frontier: list[str]) -> str: |
| 214 | doc = sample_topology() |
| 215 | doc["frontier"] = frontier |
| 216 | target = self.root / mod.TOPOLOGY_REPO_PATH |
| 217 | target.parent.mkdir(exist_ok=True) |
| 218 | target.write_text(json.dumps(doc), encoding="utf-8") |
| 219 | self.git("add", mod.TOPOLOGY_REPO_PATH) |
| 220 | self.git("commit", "-qm", "Update frontier", "--allow-empty") |
| 221 | return self.git("rev-parse", "HEAD") |
| 222 | |
| 223 | def test_aligned_main_uses_parent_and_still_rejects_growth(self) -> None: |
| 224 | previous = self.commit(["utility"]) |
| 225 | head = self.commit(["session", "utility"]) |
| 226 | self.git("update-ref", "refs/remotes/origin/main", head) |
| 227 | baseline = mod.detect_local_baseline_ref(self.root) |
| 228 | self.assertEqual(baseline, previous) |
| 229 | violations = mod.validate_baseline_transition( |
| 230 | mod.load_topology_at_ref(head, self.root), |
| 231 | mod.load_topology_at_ref(baseline, self.root), |
| 232 | ) |
| 233 | self.assertTrue(any("arbitrary growth" in str(v) for v in violations)) |
| 234 | |
| 235 | def test_feature_branch_keeps_merge_base_across_multiple_commits(self) -> None: |
| 236 | base = self.commit(["session", "utility"]) |
| 237 | self.git("update-ref", "refs/remotes/origin/main", base) |
| 238 | self.commit(["utility"]) |
| 239 | self.commit(["utility"]) |
| 240 | self.assertEqual(mod.detect_local_baseline_ref(self.root), base) |
| 241 | |
| 242 | def test_history_without_remote_uses_parent(self) -> None: |
| 243 | previous = self.commit(["session", "utility"]) |
| 244 | self.commit(["utility"]) |
| 245 | self.assertEqual(mod.detect_local_baseline_ref(self.root), previous) |
| 246 | |
| 247 | def test_initial_commit_has_no_baseline(self) -> None: |
| 248 | head = self.commit(["session", "utility"]) |
| 249 | self.git("update-ref", "refs/remotes/origin/main", head) |
| 250 | self.assertIsNone(mod.detect_local_baseline_ref(self.root)) |
| 251 | |
| 252 | |
| 253 | class SelectorTests(unittest.TestCase): |
| 254 | def test_free_selector_passes(self) -> None: |
| 255 | selector = {"kind": "free", "item": ["crate", "commands", "groups", "session", "run_save"]} |
| 256 | self.assertEqual(mod.validate_selector(selector, "s"), []) |
| 257 | |
| 258 | def test_inherent_selector_passes(self) -> None: |
| 259 | selector = { |
| 260 | "kind": "inherent", |
| 261 | "self_type": { |
| 262 | "tag": "path", |
| 263 | "absolute": True, |
| 264 | "segments": [ |
| 265 | {"name": "crate"}, |
| 266 | {"name": "commands", "args": []}, |
| 267 | {"name": "groups", "args": []}, |
| 268 | {"name": "session", "args": []}, |
| 269 | {"name": "branch", "args": []}, |
| 270 | {"name": "BranchCmd", "args": []}, |
| 271 | ], |
| 272 | }, |
| 273 | "method": "execute", |
| 274 | } |
| 275 | self.assertEqual(mod.validate_selector(selector, "s"), []) |
| 276 | |
| 277 | def test_trait_impl_selector_passes(self) -> None: |
| 278 | selector = { |
| 279 | "kind": "trait_impl", |
| 280 | "self_type": {"tag": "path", "absolute": True, "segments": [{"name": "BranchCmd"}]}, |
| 281 | "trait_path": {"tag": "path", "absolute": True, "segments": [{"name": "RegisterCommand"}]}, |
| 282 | "method": "execute", |
| 283 | } |
| 284 | self.assertEqual(mod.validate_selector(selector, "s"), []) |
| 285 | |
| 286 | def test_unknown_selector_kind_fails(self) -> None: |
| 287 | selector = {"kind": "static", "item": ["a", "b"]} |
| 288 | violations = mod.validate_selector(selector, "s") |
| 289 | self.assertTrue(any("unknown selector kind" in str(v) for v in violations)) |
| 290 | |
| 291 | def test_free_selector_missing_function_fails(self) -> None: |
| 292 | selector = {"kind": "free", "item": ["crate"]} |
| 293 | violations = mod.validate_selector(selector, "s") |
| 294 | self.assertTrue(any("module path array" in str(v) for v in violations)) |
| 295 | |
| 296 | def test_inherent_selector_missing_method_fails(self) -> None: |
| 297 | selector = {"kind": "inherent", "self_type": {"tag": "never"}} |
| 298 | violations = mod.validate_selector(selector, "s") |
| 299 | self.assertTrue(any("inherent.method" in str(v) for v in violations)) |
| 300 | |
| 301 | def test_unknown_self_type_tag_fails(self) -> None: |
| 302 | selector = {"kind": "inherent", "self_type": {"tag": "fn_ptr"}, "method": "execute"} |
| 303 | violations = mod.validate_selector(selector, "s") |
| 304 | self.assertTrue(any("unknown type node tag" in str(v) for v in violations)) |
| 305 | |
| 306 | |
| 307 | class TypeAlgebraTests(unittest.TestCase): |
| 308 | def test_primitive_and_never_pass(self) -> None: |
| 309 | self.assertEqual(mod.validate_type_node({"tag": "primitive", "name": "u8"}, "t"), []) |
| 310 | self.assertEqual(mod.validate_type_node({"tag": "never"}, "t"), []) |
| 311 | |
| 312 | def test_unknown_primitive_fails(self) -> None: |
| 313 | violations = mod.validate_type_node({"tag": "primitive", "name": "u24"}, "t") |
| 314 | self.assertTrue(violations) |
| 315 | |
| 316 | def test_tuple_and_slice_pass(self) -> None: |
| 317 | self.assertEqual( |
| 318 | mod.validate_type_node({"tag": "tuple", "elems": [{"tag": "never"}]}, "t"), [] |
| 319 | ) |
| 320 | self.assertEqual( |
| 321 | mod.validate_type_node({"tag": "slice", "inner": {"tag": "primitive", "name": "u8"}}, "t"), |
| 322 | [], |
| 323 | ) |
| 324 | |
| 325 | def test_reference_with_mut_passes(self) -> None: |
| 326 | self.assertEqual( |
| 327 | mod.validate_type_node( |
| 328 | {"tag": "reference", "mut": True, "inner": {"tag": "primitive", "name": "str"}}, |
| 329 | "t", |
| 330 | ), |
| 331 | [], |
| 332 | ) |
| 333 | |
| 334 | def test_array_with_byte_len_passes(self) -> None: |
| 335 | node = { |
| 336 | "tag": "array", |
| 337 | "inner": {"tag": "primitive", "name": "u8"}, |
| 338 | "len": {"tag": "int", "negative": False, "magnitude": "8", "suffix": None}, |
| 339 | } |
| 340 | self.assertEqual(mod.validate_type_node(node, "t"), []) |
| 341 | |
| 342 | def test_generic_const_argument_normalization(self) -> None: |
| 343 | # byte literal 255 with u8 suffix passes; 256 fails with field diagnostic |
| 344 | ok = {"tag": "const", "atom": {"tag": "int", "negative": False, "magnitude": "255", "suffix": "u8"}} |
| 345 | self.assertEqual(mod.validate_generic_arg(ok, "g"), []) |
| 346 | bad = {"tag": "const", "atom": {"tag": "int", "negative": False, "magnitude": "256", "suffix": "u8"}} |
| 347 | violations = mod.validate_generic_arg(bad, "g") |
| 348 | self.assertTrue(any("byte value 256 exceeds" in str(v) for v in violations)) |
| 349 | |
| 350 | |
| 351 | class ConstAtomTests(unittest.TestCase): |
| 352 | def test_bool_passes(self) -> None: |
| 353 | self.assertEqual(mod.validate_const_atom({"tag": "bool", "value": True}, "c"), []) |
| 354 | |
| 355 | def test_bool_non_bool_value_fails(self) -> None: |
| 356 | violations = mod.validate_const_atom({"tag": "bool", "value": 1}, "c") |
| 357 | self.assertTrue(violations) |
| 358 | |
| 359 | def test_integer_canonical_magnitude(self) -> None: |
| 360 | self.assertEqual( |
| 361 | mod.validate_const_atom( |
| 362 | {"tag": "int", "negative": False, "magnitude": "0", "suffix": None}, "c" |
| 363 | ), |
| 364 | [], |
| 365 | ) |
| 366 | self.assertEqual( |
| 367 | mod.validate_const_atom( |
| 368 | {"tag": "int", "negative": False, "magnitude": "42", "suffix": "u32"}, "c" |
| 369 | ), |
| 370 | [], |
| 371 | ) |
| 372 | violations = mod.validate_const_atom( |
| 373 | {"tag": "int", "negative": False, "magnitude": "042", "suffix": None}, "c" |
| 374 | ) |
| 375 | self.assertTrue(any("leading zeros" in str(v) for v in violations)) |
| 376 | |
| 377 | def test_negative_zero_fails(self) -> None: |
| 378 | violations = mod.validate_const_atom( |
| 379 | {"tag": "int", "negative": True, "magnitude": "0", "suffix": None}, "c" |
| 380 | ) |
| 381 | self.assertTrue(any("negative zero" in str(v) for v in violations)) |
| 382 | |
| 383 | def test_unknown_suffix_fails(self) -> None: |
| 384 | violations = mod.validate_const_atom( |
| 385 | {"tag": "int", "negative": False, "magnitude": "1", "suffix": "u33"}, "c" |
| 386 | ) |
| 387 | self.assertTrue(any("suffix" in str(v) for v in violations)) |
| 388 | |
| 389 | def test_char_scalar_passes(self) -> None: |
| 390 | self.assertEqual(mod.validate_const_atom({"tag": "char", "scalar": "x"}, "c"), []) |
| 391 | |
| 392 | def test_char_multi_scalar_fails(self) -> None: |
| 393 | violations = mod.validate_const_atom({"tag": "char", "scalar": "xy"}, "c") |
| 394 | self.assertTrue(any("exactly one" in str(v) for v in violations)) |
| 395 | |
| 396 | def test_path_const_passes(self) -> None: |
| 397 | atom = {"tag": "path", "absolute": True, "segments": ["SIZE"]} |
| 398 | self.assertEqual(mod.validate_const_atom(atom, "c"), []) |
| 399 | |
| 400 | def test_unknown_tag_fails(self) -> None: |
| 401 | violations = mod.validate_const_atom({"tag": "float", "value": 1.0}, "c") |
| 402 | self.assertTrue(any("unknown const atom tag" in str(v) for v in violations)) |
| 403 | |
| 404 | def test_extra_field_fails(self) -> None: |
| 405 | violations = mod.validate_const_atom( |
| 406 | {"tag": "int", "negative": False, "magnitude": "1", "suffix": None, "extra": True}, "c" |
| 407 | ) |
| 408 | self.assertTrue(any("exactly" in str(v) for v in violations)) |
| 409 | |
| 410 | |
| 411 | class LiveGateTests(unittest.TestCase): |
| 412 | def test_real_topology_passes_live_gate(self) -> None: |
| 413 | doc = mod.load_topology() |
| 414 | self.assertEqual(mod.validate_topology_document(doc), []) |
| 415 | |
| 416 | def test_topology_artifact_is_sorted_unique(self) -> None: |
| 417 | doc = mod.load_topology() |
| 418 | frontier = doc["frontier"] |
| 419 | self.assertEqual(frontier, sorted(frontier)) |
| 420 | self.assertEqual(len(frontier), len(set(frontier))) |
| 421 | # FEAT-018 removed utility, FEAT-019 removed memory, FEAT-020 removed plugins, |
| 422 | # FEAT-021 removed project, FEAT-022 removed skills, FEAT-029 completed debug, |
| 423 | # and FEAT-026 completed session; config/core remain. |
| 424 | self.assertEqual( |
| 425 | set(frontier), |
| 426 | {"config", "core"}, |
| 427 | ) |
| 428 | |
| 429 | |
| 430 | class DebugGroupFrontierTests(unittest.TestCase): |
| 431 | def test_complete_debug_inventory_has_no_pending_host_handlers(self) -> None: |
| 432 | doc = mod.load_topology() |
| 433 | debug = doc["topology"]["debug"] |
| 434 | actual = {p.relative_to(ROOT).as_posix() for p in |
| 435 | (ROOT / "crates/tui/src/commands/groups/debug").rglob("*.rs")} |
| 436 | self.assertEqual(set(mod.group_source_scope("debug", debug, ROOT)), actual) |
| 437 | self.assertNotIn("debug", doc["frontier"]) |
| 438 | self.assertNotIn("debug", mod.load_pending_groups()) |
| 439 | self.assertEqual(mod.check_source_frontier({"debug": debug}, [], ROOT), []) |
| 440 | self.assertIn("crates/tui/src/commands/groups/debug/receipts.rs", actual) |
| 441 | |
| 442 | def test_omitted_receipts_and_new_nested_commands_fail_inventory(self) -> None: |
| 443 | for filename in ["receipts.rs", "new/command.rs"]: |
| 444 | with self.subTest(filename=filename), tempfile.TemporaryDirectory() as tmp: |
| 445 | root = Path(tmp) |
| 446 | directory = root / "crates/tui/src/commands/groups/debug" |
| 447 | directory.mkdir(parents=True) |
| 448 | (directory / "mod.rs").write_text("// complete portable module\n") |
| 449 | omitted = directory / filename |
| 450 | omitted.parent.mkdir(parents=True, exist_ok=True) |
| 451 | omitted.write_text("pub fn receipt(app: &mut App) {}\n") |
| 452 | scope = [(directory / "mod.rs").relative_to(root).as_posix()] |
| 453 | topology = {"debug": {"kind": "group", "scope": scope, "slices": []}} |
| 454 | violations = mod.check_source_frontier(topology, [], root) |
| 455 | self.assertTrue(any(v.category == "stale-removal" and |
| 456 | v.location.endswith("::receipt") |
| 457 | for v in violations), violations) |
| 458 | # Adding the historical scope entry cannot declare the handler portable. |
| 459 | scope.append(omitted.relative_to(root).as_posix()) |
| 460 | violations = mod.check_source_frontier(topology, [], root) |
| 461 | self.assertTrue(any(v.category == "stale-removal" for v in violations), violations) |
| 462 | |
| 463 | def test_omitted_pure_file_is_automatically_included_in_scan(self) -> None: |
| 464 | with tempfile.TemporaryDirectory() as tmp: |
| 465 | root = Path(tmp) |
| 466 | directory = root / "crates/tui/src/commands/groups/debug" |
| 467 | directory.mkdir(parents=True) |
| 468 | (directory / "future.rs").write_text("pub fn future() {}\n") |
| 469 | node = {"kind": "group", "scope": [], "slices": []} |
| 470 | self.assertEqual(mod.group_source_scope("debug", node, root), |
| 471 | ["crates/tui/src/commands/groups/debug/future.rs"]) |
| 472 | self.assertEqual(mod.check_source_frontier({"debug": node}, [], root), []) |
| 473 | |
| 474 | |
| 475 | class SourceScanTests(unittest.TestCase): |
| 476 | """Hermetic fixtures for the AST-resolved source scan (Task 3.5/3.6).""" |
| 477 | |
| 478 | def _write_group(self, tmpdir: Path, group: str, files: dict[str, str]) -> Path: |
| 479 | """Write a synthetic group tree under a temp groups root.""" |
| 480 | base = tmpdir / group |
| 481 | base.mkdir(parents=True, exist_ok=True) |
| 482 | for name, content in files.items(): |
| 483 | (base / name).write_text(content, encoding="utf-8") |
| 484 | return tmpdir |
| 485 | |
| 486 | def test_parse_finds_concrete_app_free_fn(self) -> None: |
| 487 | source = ( |
| 488 | "use crate::tui::app::App;\n" |
| 489 | "pub fn run_config(app: &mut App, arg: Option<&str>) -> CommandResult {\n" |
| 490 | " CommandResult::ok()\n" |
| 491 | "}\n" |
| 492 | ) |
| 493 | import tempfile |
| 494 | with tempfile.TemporaryDirectory() as d: |
| 495 | root = Path(d) |
| 496 | (root / "config").mkdir() |
| 497 | (root / "config" / "mod.rs").write_text(source, encoding="utf-8") |
| 498 | items = mod.parse_rust_file(root / "config" / "mod.rs", root) |
| 499 | apps = [it for it in items if it.is_concrete_app] |
| 500 | self.assertEqual(len(apps), 1) |
| 501 | self.assertEqual(apps[0].kind, "free") |
| 502 | self.assertEqual(apps[0].qual_path, "crate::commands::groups::config::run_config") |
| 503 | |
| 504 | def test_parse_ignores_non_app_fns(self) -> None: |
| 505 | source = ( |
| 506 | "fn helper(value: u32) -> u32 { value }\n" |
| 507 | "fn run(app: &mut crate::tui::app::App, arg: Option<&str>) -> CommandResult {\n" |
| 508 | " CommandResult::ok()\n" |
| 509 | "}\n" |
| 510 | ) |
| 511 | import tempfile |
| 512 | with tempfile.TemporaryDirectory() as d: |
| 513 | root = Path(d) |
| 514 | (root / "core").mkdir() |
| 515 | (root / "core" / "mod.rs").write_text(source, encoding="utf-8") |
| 516 | items = mod.parse_rust_file(root / "core" / "mod.rs", root) |
| 517 | self.assertEqual(sum(1 for it in items if it.is_concrete_app), 1) |
| 518 | |
| 519 | def test_parse_finds_trait_impl_and_inherent_methods(self) -> None: |
| 520 | source = ( |
| 521 | "pub struct BranchCmd;\n" |
| 522 | "impl RegisterCommand for BranchCmd {\n" |
| 523 | " fn info() -> &'static CommandInfo { &INFO }\n" |
| 524 | " fn execute(app: &mut App, arg: Option<&str>) -> CommandResult {\n" |
| 525 | " branch(app, arg)\n" |
| 526 | " }\n" |
| 527 | "}\n" |
| 528 | "impl BranchCmd {\n" |
| 529 | " pub fn helper(&self) -> u32 { 1 }\n" |
| 530 | "}\n" |
| 531 | ) |
| 532 | import tempfile |
| 533 | with tempfile.TemporaryDirectory() as d: |
| 534 | root = Path(d) |
| 535 | (root / "session").mkdir() |
| 536 | (root / "session" / "mod.rs").write_text(source, encoding="utf-8") |
| 537 | items = mod.parse_rust_file(root / "session" / "mod.rs", root) |
| 538 | trait_exec = [it for it in items if it.kind == "trait_impl" and it.name == "execute"] |
| 539 | self.assertEqual(len(trait_exec), 1) |
| 540 | self.assertTrue(trait_exec[0].is_concrete_app) |
| 541 | inherent = [it for it in items if it.kind == "inherent"] |
| 542 | self.assertEqual(len(inherent), 1) |
| 543 | self.assertEqual(inherent[0].name, "helper") |
| 544 | self.assertFalse(inherent[0].is_concrete_app) |
| 545 | |
| 546 | def test_scope_file_missing_fails(self) -> None: |
| 547 | violations = mod.scan_leaf_handlers(["crates/tui/src/commands/groups/core/ghost.rs"], Path("/nonexistent"))[1] |
| 548 | self.assertTrue(any("missing" in str(v) for v in violations)) |
| 549 | |
| 550 | def test_frontier_matches_group_source(self) -> None: |
| 551 | doc = sample_topology() |
| 552 | # utility scope has one concrete-App handler; session scope has one. |
| 553 | doc["topology"]["utility"]["scope"] = ["utility/mod.rs"] |
| 554 | doc["topology"]["session"]["scope"] = ["session/mod.rs"] |
| 555 | import tempfile |
| 556 | with tempfile.TemporaryDirectory() as d: |
| 557 | root = Path(d) |
| 558 | (root / "utility").mkdir(parents=True) |
| 559 | (root / "session").mkdir(parents=True) |
| 560 | (root / "utility" / "mod.rs").write_text( |
| 561 | "use crate::tui::app::App;\n" |
| 562 | "fn run_util(app: &mut App, arg: Option<&str>) -> CommandResult { CommandResult::ok() }\n", |
| 563 | encoding="utf-8", |
| 564 | ) |
| 565 | (root / "session" / "mod.rs").write_text( |
| 566 | "use crate::tui::app::App;\n" |
| 567 | "fn run_save(app: &mut App, arg: Option<&str>) -> CommandResult { CommandResult::ok() }\n", |
| 568 | encoding="utf-8", |
| 569 | ) |
| 570 | violations = mod.check_source_frontier(doc["topology"], doc["frontier"], root) |
| 571 | self.assertEqual(violations, []) |
| 572 | |
| 573 | def test_cheating_removal_fails(self) -> None: |
| 574 | doc = sample_topology() |
| 575 | doc["topology"]["utility"]["scope"] = ["utility/mod.rs"] |
| 576 | doc["topology"]["session"]["scope"] = ["session/mod.rs"] |
| 577 | import tempfile |
| 578 | with tempfile.TemporaryDirectory() as d: |
| 579 | root = Path(d) |
| 580 | (root / "utility").mkdir(parents=True) |
| 581 | (root / "session").mkdir(parents=True) |
| 582 | (root / "utility" / "mod.rs").write_text( |
| 583 | "use crate::tui::app::App;\n" |
| 584 | "fn run_util(app: &mut App, arg: Option<&str>) -> CommandResult { CommandResult::ok() }\n", |
| 585 | encoding="utf-8", |
| 586 | ) |
| 587 | (root / "session" / "mod.rs").write_text( |
| 588 | "use crate::tui::app::App;\n" |
| 589 | "fn run_save(app: &mut App, arg: Option<&str>) -> CommandResult { CommandResult::ok() }\n", |
| 590 | encoding="utf-8", |
| 591 | ) |
| 592 | # Cheat: remove utility from the frontier while its handler remains. |
| 593 | violations = mod.check_source_frontier(doc["topology"], ["session"], root) |
| 594 | self.assertTrue(any("stale-removal" in str(v) for v in violations)) |
| 595 | |
| 596 | def test_stale_frontier_entry_fails(self) -> None: |
| 597 | doc = sample_topology() |
| 598 | doc["topology"]["utility"]["scope"] = ["utility/mod.rs"] |
| 599 | doc["topology"]["session"]["scope"] = ["session/mod.rs"] |
| 600 | import tempfile |
| 601 | with tempfile.TemporaryDirectory() as d: |
| 602 | root = Path(d) |
| 603 | (root / "utility").mkdir(parents=True) |
| 604 | (root / "session").mkdir(parents=True) |
| 605 | # utility has NO concrete-App handler (migrated: uses a context) |
| 606 | (root / "utility" / "mod.rs").write_text( |
| 607 | "fn run_util(contexts: CommandContexts<'_>, arg: Option<&str>) -> CommandResult { CommandResult::ok() }\n", |
| 608 | encoding="utf-8", |
| 609 | ) |
| 610 | (root / "session" / "mod.rs").write_text( |
| 611 | "use crate::tui::app::App;\n" |
| 612 | "fn run_save(app: &mut App, arg: Option<&str>) -> CommandResult { CommandResult::ok() }\n", |
| 613 | encoding="utf-8", |
| 614 | ) |
| 615 | violations = mod.check_source_frontier(doc["topology"], ["session", "utility"], root) |
| 616 | self.assertTrue(any("stale-entry" in str(v) for v in violations)) |
| 617 | |
| 618 | def test_retained_host_exempts_declared_machinery_from_stale_removal(self) -> None: |
| 619 | """A migrated group may declare dispatcher host machinery (FEAT-042) |
| 620 | that keeps `&mut App`; the gate exempts it and flags the rest.""" |
| 621 | topology = { |
| 622 | "alpha": { |
| 623 | "kind": "group", |
| 624 | "scope": ["alpha/mod.rs"], |
| 625 | "slices": [], |
| 626 | } |
| 627 | } |
| 628 | frontier: list[str] = [] |
| 629 | import tempfile |
| 630 | with tempfile.TemporaryDirectory() as d: |
| 631 | root = Path(d) |
| 632 | (root / "alpha").mkdir(parents=True, exist_ok=True) |
| 633 | (root / "alpha" / "mod.rs").write_text( |
| 634 | "use crate::tui::app::App;\n" |
| 635 | "fn retained(app: &mut App, arg: Option<&str>) {} \n" |
| 636 | "fn stale(app: &mut App, arg: Option<&str>) {} \n", |
| 637 | encoding="utf-8", |
| 638 | ) |
| 639 | # stub RETAINED_HOST_MACHINERY for the hermetic group |
| 640 | original = mod.RETAINED_HOST_MACHINERY |
| 641 | try: |
| 642 | mod.RETAINED_HOST_MACHINERY = { |
| 643 | "alpha": [ |
| 644 | {"kind": "free", "item": ["crate", "commands", "groups", "alpha", "retained"]} |
| 645 | ] |
| 646 | } |
| 647 | violations = mod.check_source_frontier(topology, frontier, root) |
| 648 | finally: |
| 649 | mod.RETAINED_HOST_MACHINERY = original |
| 650 | kinds = [v.category for v in violations] |
| 651 | self.assertNotIn("retained-host", kinds) |
| 652 | self.assertEqual(kinds.count("stale-removal"), 1, violations) |
| 653 | |
| 654 | def test_retained_host_fails_closed_when_signature_lost(self) -> None: |
| 655 | """If retained machinery loses its concrete-App signature, the gate fails.""" |
| 656 | topology = { |
| 657 | "alpha": { |
| 658 | "kind": "group", |
| 659 | "scope": ["alpha/mod.rs"], |
| 660 | "slices": [], |
| 661 | } |
| 662 | } |
| 663 | frontier: list[str] = [] |
| 664 | import tempfile |
| 665 | with tempfile.TemporaryDirectory() as d: |
| 666 | root = Path(d) |
| 667 | (root / "alpha").mkdir(parents=True, exist_ok=True) |
| 668 | (root / "alpha" / "mod.rs").write_text( |
| 669 | "fn retained(arg: Option<&str>) {} \n", |
| 670 | encoding="utf-8", |
| 671 | ) |
| 672 | original = mod.RETAINED_HOST_MACHINERY |
| 673 | try: |
| 674 | mod.RETAINED_HOST_MACHINERY = { |
| 675 | "alpha": [ |
| 676 | {"kind": "free", "item": ["crate", "commands", "groups", "alpha", "retained"]} |
| 677 | ] |
| 678 | } |
| 679 | violations = mod.check_source_frontier(topology, frontier, root) |
| 680 | finally: |
| 681 | mod.RETAINED_HOST_MACHINERY = original |
| 682 | self.assertTrue( |
| 683 | any(v.category == "retained-host" for v in violations), |
| 684 | f"expected retained-host violation, got {violations}", |
| 685 | ) |
| 686 | |
| 687 | def test_live_source_gate_passes(self) -> None: |
| 688 | doc = mod.load_topology() |
| 689 | violations = mod.check_source_frontier(doc["topology"], doc["frontier"]) |
| 690 | self.assertEqual(violations, []) |
| 691 | |
| 692 | |
| 693 | if __name__ == "__main__": |
| 694 | unittest.main() |
| 695 |