test_phase6_acceptance.py
python
sha256:408631a8dc866047a2cff7440933b461af5abd4b55e755599765e433abceb88b
feat(dev-safety): Phase 6 of #185 — wire together, document…
Sonnet 5
minor
⚠ breaking
3 days ago
| 1 | """Phase 6 of #185 (musehub staging): full dry-run acceptance test. |
| 2 | |
| 3 | Ties every prior phase together in one flow, against a REAL scratch muse |
| 4 | repo under tmp_path (never ~/ecosystem/muse): refresh a sandbox (which |
| 5 | itself takes a snapshot backup first), corrupt the sandbox on purpose, |
| 6 | confirm canonical `muse verify` stays green throughout, then exercise both |
| 7 | restore paths (snapshot and bundle) successfully against a separately |
| 8 | corrupted canonical. |
| 9 | """ |
| 10 | import json |
| 11 | import shutil |
| 12 | import subprocess |
| 13 | import sys |
| 14 | from pathlib import Path |
| 15 | |
| 16 | import pytest |
| 17 | |
| 18 | sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "scripts" / "dev")) |
| 19 | from backup import create_bundle_backup, list_snapshots, restore_from_bundle, restore_from_snapshot # noqa: E402 |
| 20 | from sandbox import refresh_sandbox # noqa: E402 |
| 21 | |
| 22 | MUSE = shutil.which("muse") |
| 23 | |
| 24 | |
| 25 | def _run(args: list[str], cwd: Path) -> subprocess.CompletedProcess: |
| 26 | return subprocess.run([MUSE, *args], cwd=cwd, capture_output=True, text=True, check=True) |
| 27 | |
| 28 | |
| 29 | def _make_real_repo(root: Path, *, commits: int = 3) -> None: |
| 30 | root.mkdir(parents=True, exist_ok=True) |
| 31 | _run(["init"], cwd=root) |
| 32 | for i in range(commits): |
| 33 | (root / f"file{i}.txt").write_text(f"content {i}\n") |
| 34 | _run(["code", "add", "."], cwd=root) |
| 35 | _run(["commit", "-m", f"commit {i}"], cwd=root) |
| 36 | |
| 37 | |
| 38 | def _rev_parse(root: Path, ref: str) -> str: |
| 39 | return json.loads(_run(["rev-parse", ref, "--json"], cwd=root).stdout)["commit_id"] |
| 40 | |
| 41 | |
| 42 | def _verify_all_ok(root: Path) -> bool: |
| 43 | proc = subprocess.run([MUSE, "verify", "--json"], cwd=root, capture_output=True, text=True) |
| 44 | return json.loads(proc.stdout).get("all_ok", False) |
| 45 | |
| 46 | |
| 47 | def _corrupt_an_object(root: Path) -> None: |
| 48 | objects_root = root / ".muse" / "objects" / "sha256" |
| 49 | for shard in objects_root.iterdir(): |
| 50 | for obj in shard.iterdir(): |
| 51 | if obj.is_file(): |
| 52 | obj.chmod(0o644) |
| 53 | obj.write_text("CORRUPTED-FOR-TEST") |
| 54 | return |
| 55 | raise AssertionError("no object found to corrupt") |
| 56 | |
| 57 | |
| 58 | @pytest.fixture(autouse=True) |
| 59 | def _require_muse(): |
| 60 | if MUSE is None: |
| 61 | pytest.skip("muse not installed on this machine yet (Phase 2 not applied)") |
| 62 | |
| 63 | |
| 64 | class TestFullPipelineAcceptance: |
| 65 | def test_sandbox_corruption_never_reaches_canonical_and_both_restores_work( |
| 66 | self, tmp_path: Path, |
| 67 | ) -> None: |
| 68 | canonical = tmp_path / "ecosystem" / "muse" |
| 69 | sandbox_base = tmp_path / "sandboxes" |
| 70 | backup_base = tmp_path / "backups" |
| 71 | _make_real_repo(canonical) |
| 72 | good_main = _rev_parse(canonical, "main") |
| 73 | |
| 74 | # ── Step 1: refresh a sandbox — this must also produce a snapshot backup. |
| 75 | sandbox_path = refresh_sandbox( |
| 76 | "muse", canonical_root=canonical, sandbox_base=sandbox_base, backup_base=backup_base, |
| 77 | ) |
| 78 | snaps_after_refresh = list_snapshots("muse", backup_base=backup_base) |
| 79 | assert len(snaps_after_refresh) == 1, "sandbox-refresh must snapshot canonical first" |
| 80 | |
| 81 | # ── Step 2: corrupt the sandbox on purpose. |
| 82 | _corrupt_an_object(sandbox_path) |
| 83 | |
| 84 | # ── Step 3: canonical must stay green throughout. |
| 85 | assert _verify_all_ok(canonical) is True |
| 86 | assert _rev_parse(canonical, "main") == good_main |
| 87 | |
| 88 | # ── Step 4: take a real bundle backup of (still-healthy) canonical. |
| 89 | bundle_path = create_bundle_backup("muse", canonical_root=canonical, backup_base=backup_base) |
| 90 | |
| 91 | # ── Step 5: corrupt canonical itself now, and restore via snapshot. |
| 92 | _corrupt_an_object(canonical) |
| 93 | assert _verify_all_ok(canonical) is False |
| 94 | |
| 95 | snap_name = snaps_after_refresh[0].name |
| 96 | restore_from_snapshot("muse", snap_name, canonical_root=canonical, backup_base=backup_base, force=True) |
| 97 | assert _verify_all_ok(canonical) is True |
| 98 | assert _rev_parse(canonical, "main") == good_main |
| 99 | |
| 100 | # ── Step 6: corrupt canonical again, restore via bundle this time. |
| 101 | _corrupt_an_object(canonical) |
| 102 | assert _verify_all_ok(canonical) is False |
| 103 | |
| 104 | restore_from_bundle( |
| 105 | "muse", bundle_path.name, canonical_root=canonical, backup_base=backup_base, force=True, |
| 106 | ) |
| 107 | assert _verify_all_ok(canonical) is True |
| 108 | assert _rev_parse(canonical, "main") == good_main |
| 109 | |
| 110 | # ── Step 7: the sandbox's earlier corruption was never visible to |
| 111 | # canonical at any point in this whole flow — re-confirm explicitly. |
| 112 | assert (sandbox_path / ".muse").exists() |
| 113 | sandbox_verify = subprocess.run( |
| 114 | [MUSE, "verify", "--json"], cwd=sandbox_path, capture_output=True, text=True, |
| 115 | ) |
| 116 | # The sandbox itself may or may not still fail verify (it was |
| 117 | # deliberately corrupted and never repaired) — what matters is |
| 118 | # canonical was independent of it throughout. |
| 119 | assert _verify_all_ok(canonical) is True |
File History
1 commit
sha256:408631a8dc866047a2cff7440933b461af5abd4b55e755599765e433abceb88b
feat(dev-safety): Phase 6 of #185 — wire together, document…
Sonnet 5
minor
⚠
3 days ago