aaronrene / overseer-kit public
test_k7_bridge_idempotency.py python
93 lines 3.6 KB
Raw
sha256:0e9549ec7b463911bc08b7d586dc320b1ac9b1f5c943ee7e3865dcc6cb0f6f83 chore(governance): sync handover+roadmap to 84db8c8 (drift:… Human 2 days ago
1 """Data-integrity tests for K7 bridge footprint (§K7.8 data-integrity tier)."""
2
3 from __future__ import annotations
4
5 from pathlib import Path
6 from unittest.mock import patch
7
8 from cli.digest import FootprintRecord, compute_footprint_digest, sha256_hex
9 from cli.footprint import MUSE_BRIDGE_DEPLOY_DEST, MUSE_BRIDGE_WORKFLOW_DEST, resolve_footprint
10 from cli.kit_root import kit_root
11 from cli.version_lock import read_version_lock
12 from tests.support import FIXTURES, muse_mirror_status_runner, run_cli
13
14
15 def test_muse_mirror_init_twice_identical_digest(tmp_path: Path) -> None:
16 runner = muse_mirror_status_runner(tmp_path)
17 args = [
18 "init",
19 "--from-config",
20 str(FIXTURES / "config-muse-git-mirror.yaml"),
21 "--non-interactive",
22 ]
23 assert run_cli(args, cwd=tmp_path, kit=kit_root(), runner=runner) == 0
24 lock1 = read_version_lock(tmp_path / ".overseer" / "version.lock")
25 assert run_cli(args, cwd=tmp_path, kit=kit_root(), runner=runner) == 0
26 lock2 = read_version_lock(tmp_path / ".overseer" / "version.lock")
27 assert lock1.footprint_digest == lock2.footprint_digest
28
29
30 def test_kit_only_digest_includes_bridge_when_muse_mirror(tmp_path: Path) -> None:
31 run_cli(
32 [
33 "init",
34 "--from-config",
35 str(FIXTURES / "config-muse-git-mirror.yaml"),
36 "--non-interactive",
37 ],
38 cwd=tmp_path,
39 kit=kit_root(),
40 runner=muse_mirror_status_runner(tmp_path),
41 )
42 from adapters.config import load_config
43
44 config = load_config(tmp_path / ".overseer" / "config.yaml")
45 rendered = resolve_footprint(config, kit=kit_root())
46 records = [
47 FootprintRecord(item.destination, sha256_hex((tmp_path / item.destination).read_bytes()))
48 for item in rendered
49 ]
50 expected = compute_footprint_digest(records)
51 lock = read_version_lock(tmp_path / ".overseer" / "version.lock")
52 assert lock.footprint_digest == expected
53 assert MUSE_BRIDGE_WORKFLOW_DEST in {e.path for e in lock.footprint}
54
55
56 def test_git_only_digest_omits_bridge_destinations(tmp_path: Path) -> None:
57 run_cli(["init", "--regime", "git-only", "--non-interactive"], cwd=tmp_path, kit=kit_root())
58 lock = read_version_lock(tmp_path / ".overseer" / "version.lock")
59 paths = {e.path for e in lock.footprint}
60 assert MUSE_BRIDGE_WORKFLOW_DEST not in paths
61 assert MUSE_BRIDGE_DEPLOY_DEST not in paths
62
63
64 def test_bridge_write_failure_leaves_lock_unchanged(tmp_path: Path) -> None:
65 run_cli(
66 ["init", "--regime", "git-only", "--non-interactive"],
67 cwd=tmp_path,
68 kit=kit_root(),
69 )
70 before = read_version_lock(tmp_path / ".overseer" / "version.lock")
71 mirror_cfg = (FIXTURES / "config-muse-git-mirror.yaml").read_text(encoding="utf-8")
72 (tmp_path / ".overseer" / "config.yaml").write_text(mirror_cfg, encoding="utf-8")
73 calls = {"n": 0}
74
75 def flaky_write(path: Path, data: bytes, *, destination: str) -> None:
76 calls["n"] += 1
77 if MUSE_BRIDGE_DEPLOY_DEST in destination and calls["n"] >= 1:
78 from cli.atomic import WriteFailure
79
80 raise WriteFailure(path, OSError("simulated failure"))
81 from cli.footprint_writes import write_footprint_bytes as real
82
83 real(path, data, destination=destination)
84
85 with patch("cli.commands.sync.write_footprint_bytes", side_effect=flaky_write):
86 code = run_cli(
87 ["sync", "-y"],
88 cwd=tmp_path,
89 runner=muse_mirror_status_runner(tmp_path),
90 )
91 assert code == 5
92 after = read_version_lock(tmp_path / ".overseer" / "version.lock")
93 assert after.footprint_digest == before.footprint_digest
File History 1 commit
sha256:6abcf1fa82a7a621ccbc945f19acdba5bc0db54569599404a1452fb4a096a199 fix(ISR): default require_independent_second_reviewer to require Human minor ⚠ 2 days ago