gabriel / muse public
test_backup.py python
187 lines 7.0 KB
Raw
sha256:a57e9ca1e385a1e7a0e3e28094bc35799950a3ed7b2421712d6fb7f0e523a4a0 feat(dev-safety): Phase 5 of #185 — belt-and-suspenders aut… Sonnet 5 patch 3 days ago
1 """Phase 5 of #185 (musehub staging): automated backups (belt and suspenders).
2
3 Two independent mechanisms — fast local APFS snapshots ("suspenders") and
4 verified `muse bundle` archives ("belt") — plus a restore path that refuses
5 to clobber a currently-healthy canonical repo. All tests build a REAL
6 scratch muse repo under tmp_path (via the real `muse` binary) and only ever
7 point canonical_root/backup_base at tmp_path — never the real
8 ~/ecosystem/muse, matching every prior phase's discipline.
9 """
10 import shutil
11 import subprocess
12 import sys
13 from pathlib import Path
14
15 import pytest
16
17 sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "scripts" / "dev"))
18 from backup import ( # noqa: E402
19 CanonicalHealthyError,
20 create_bundle_backup,
21 create_snapshot,
22 list_bundles,
23 list_snapshots,
24 restore_from_bundle,
25 restore_from_snapshot,
26 )
27
28 MUSE = shutil.which("muse")
29
30
31 def _run(args: list[str], cwd: Path) -> subprocess.CompletedProcess:
32 return subprocess.run([MUSE, *args], cwd=cwd, capture_output=True, text=True, check=True)
33
34
35 def _make_real_repo(root: Path, *, commits: int = 3) -> None:
36 root.mkdir(parents=True, exist_ok=True)
37 _run(["init"], cwd=root)
38 for i in range(commits):
39 (root / f"file{i}.txt").write_text(f"content {i}\n")
40 _run(["code", "add", "."], cwd=root)
41 _run(["commit", "-m", f"commit {i}"], cwd=root)
42
43
44 def _rev_parse(root: Path, ref: str) -> str:
45 import json
46 proc = _run(["rev-parse", ref, "--json"], cwd=root)
47 return json.loads(proc.stdout)["commit_id"]
48
49
50 def _verify_all_ok(root: Path) -> bool:
51 import json
52 proc = subprocess.run([MUSE, "verify", "--json"], cwd=root, capture_output=True, text=True)
53 return json.loads(proc.stdout).get("all_ok", False)
54
55
56 def _corrupt_an_object(root: Path) -> None:
57 objects_root = root / ".muse" / "objects" / "sha256"
58 for shard in objects_root.iterdir():
59 for obj in shard.iterdir():
60 if obj.is_file():
61 obj.chmod(0o644)
62 obj.write_text("CORRUPTED-FOR-TEST")
63 return
64 raise AssertionError("no object found to corrupt")
65
66
67 @pytest.fixture(autouse=True)
68 def _require_muse():
69 if MUSE is None:
70 pytest.skip("muse not installed on this machine yet (Phase 2 not applied)")
71
72
73 class TestSnapshotBackupRestore:
74 def test_full_round_trip_via_snapshot(self, tmp_path: Path) -> None:
75 canonical = tmp_path / "ecosystem" / "muse"
76 backup_base = tmp_path / "backups"
77 _make_real_repo(canonical)
78 good_dev = _rev_parse(canonical, "main")
79
80 snap = create_snapshot("muse", canonical_root=canonical, backup_base=backup_base)
81 assert snap.exists()
82
83 _corrupt_an_object(canonical)
84 assert _verify_all_ok(canonical) is False
85
86 restore_from_snapshot("muse", snap.name, canonical_root=canonical, backup_base=backup_base, force=True)
87
88 assert _verify_all_ok(canonical) is True
89 assert _rev_parse(canonical, "main") == good_dev
90
91 def test_rotation_keeps_only_newest_n(self, tmp_path: Path) -> None:
92 canonical = tmp_path / "ecosystem" / "muse"
93 backup_base = tmp_path / "backups"
94 _make_real_repo(canonical, commits=1)
95
96 paths = [
97 create_snapshot("muse", canonical_root=canonical, backup_base=backup_base, keep=3)
98 for _ in range(8)
99 ]
100
101 remaining = list_snapshots("muse", backup_base=backup_base)
102 assert len(remaining) == 3
103 assert {p.name for p in remaining} == {p.name for p in paths[-3:]}
104
105 def test_refuses_to_restore_over_healthy_canonical_without_force(self, tmp_path: Path) -> None:
106 canonical = tmp_path / "ecosystem" / "muse"
107 backup_base = tmp_path / "backups"
108 _make_real_repo(canonical)
109 snap = create_snapshot("muse", canonical_root=canonical, backup_base=backup_base)
110
111 assert _verify_all_ok(canonical) is True
112 with pytest.raises(CanonicalHealthyError):
113 restore_from_snapshot("muse", snap.name, canonical_root=canonical, backup_base=backup_base)
114
115 def test_restores_without_force_when_canonical_already_unhealthy(self, tmp_path: Path) -> None:
116 canonical = tmp_path / "ecosystem" / "muse"
117 backup_base = tmp_path / "backups"
118 _make_real_repo(canonical)
119 good_dev = _rev_parse(canonical, "main")
120 snap = create_snapshot("muse", canonical_root=canonical, backup_base=backup_base)
121
122 _corrupt_an_object(canonical)
123 assert _verify_all_ok(canonical) is False
124
125 restore_from_snapshot("muse", snap.name, canonical_root=canonical, backup_base=backup_base) # no force needed
126
127 assert _verify_all_ok(canonical) is True
128 assert _rev_parse(canonical, "main") == good_dev
129
130
131 class TestBundleBackupRestore:
132 def test_full_round_trip_via_bundle(self, tmp_path: Path) -> None:
133 canonical = tmp_path / "ecosystem" / "muse"
134 backup_base = tmp_path / "backups"
135 _make_real_repo(canonical)
136 good_dev = _rev_parse(canonical, "main")
137
138 bundle_path = create_bundle_backup("muse", canonical_root=canonical, backup_base=backup_base)
139 assert bundle_path.exists()
140
141 _corrupt_an_object(canonical)
142 assert _verify_all_ok(canonical) is False
143
144 restore_from_bundle("muse", bundle_path.name, canonical_root=canonical, backup_base=backup_base, force=True)
145
146 assert _verify_all_ok(canonical) is True
147 assert _rev_parse(canonical, "main") == good_dev
148
149 def test_bundle_diff_reports_zero_new_commits_after_restore(self, tmp_path: Path) -> None:
150 import json
151
152 canonical = tmp_path / "ecosystem" / "muse"
153 backup_base = tmp_path / "backups"
154 _make_real_repo(canonical)
155
156 bundle_path = create_bundle_backup("muse", canonical_root=canonical, backup_base=backup_base)
157 _corrupt_an_object(canonical)
158 restore_from_bundle("muse", bundle_path.name, canonical_root=canonical, backup_base=backup_base, force=True)
159
160 proc = _run(["bundle", "diff", str(bundle_path), "--json"], cwd=canonical)
161 result = json.loads(proc.stdout)
162 assert result["new_commits"] == 0
163
164 def test_bundle_is_verified_at_creation_time(self, tmp_path: Path) -> None:
165 canonical = tmp_path / "ecosystem" / "muse"
166 backup_base = tmp_path / "backups"
167 _make_real_repo(canonical)
168
169 bundle_path = create_bundle_backup("muse", canonical_root=canonical, backup_base=backup_base)
170
171 proc = _run(["bundle", "verify", str(bundle_path), "--json"], cwd=canonical)
172 import json
173 assert json.loads(proc.stdout)["all_ok"] is True
174
175 def test_rotation_keeps_only_newest_n(self, tmp_path: Path) -> None:
176 canonical = tmp_path / "ecosystem" / "muse"
177 backup_base = tmp_path / "backups"
178 _make_real_repo(canonical, commits=1)
179
180 paths = [
181 create_bundle_backup("muse", canonical_root=canonical, backup_base=backup_base, keep=3)
182 for _ in range(8)
183 ]
184
185 remaining = list_bundles("muse", backup_base=backup_base)
186 assert len(remaining) == 3
187 assert {p.name for p in remaining} == {p.name for p in paths[-3:]}
File History 1 commit
sha256:a57e9ca1e385a1e7a0e3e28094bc35799950a3ed7b2421712d6fb7f0e523a4a0 feat(dev-safety): Phase 5 of #185 — belt-and-suspenders aut… Sonnet 5 patch 3 days ago