test_gsb_collision_stress.py
file-level
1
files
1
commits
0
hotspots
0
🧊 dead
0
💥 blast risk
| 1 | """Stress: repeated same-day ``--write`` with induced failure mid-C0 (§GSB.8). |
| 2 | |
| 3 | N≥20 alternating attempts where the ensure fails after a partial C0 tip |
| 4 | update (Muse leg succeeded, Git leg fails — and the reverse): every failure |
| 5 | must restore ``original_branch_state`` on both histories, leave docs and |
| 6 | marker untouched, and never accumulate stranded HEADs on |
| 7 | ``feat/governance-sync-*``. |
| 8 | """ |
| 9 | |
| 10 | from __future__ import annotations |
| 11 | |
| 12 | from datetime import date |
| 13 | from pathlib import Path |
| 14 | |
| 15 | from adapters.runner import CommandResult |
| 16 | from cli.kit_root import kit_root |
| 17 | from tests.support import gsw_runner, run_cli, seed_gsw_repo |
| 18 | |
| 19 | ROUNDS = 20 |
| 20 | |
| 21 | |
| 22 | def _feature_branch() -> str: |
| 23 | return f"feat/governance-sync-{date.today().isoformat()}" |
| 24 | |
| 25 | |
| 26 | def test_partial_c0_failure_never_strands_heads(tmp_path: Path) -> None: |
| 27 | branch = _feature_branch() |
| 28 | handover_path, roadmap_path = seed_gsw_repo(tmp_path, "muse+git-mirror") |
| 29 | original_handover = handover_path.read_bytes() |
| 30 | original_roadmap = roadmap_path.read_bytes() |
| 31 | |
| 32 | runner = gsw_runner( |
| 33 | tmp_path, |
| 34 | "muse+git-mirror", |
| 35 | existing_git_branches={branch}, |
| 36 | existing_muse_branches={branch}, |
| 37 | git_tips={branch: "gitstale"}, |
| 38 | muse_tips={branch: "sha256:stale"}, |
| 39 | git_ancestors={"feedface": {"gitstale"}}, |
| 40 | muse_ancestors={"sha256:musetip": {"sha256:stale"}}, |
| 41 | ) |
| 42 | |
| 43 | fail_muse_leg = False |
| 44 | original_run = runner.run |
| 45 | |
| 46 | def breaking_run(command: str, *, cwd: str | None = None) -> CommandResult: |
| 47 | if fail_muse_leg and command.startswith("muse") and " update-ref " in command: |
| 48 | runner.calls.append((command, cwd)) |
| 49 | return CommandResult(stdout="", stderr="induced muse tip failure", exit_code=1) |
| 50 | if not fail_muse_leg and command.startswith("git branch -f "): |
| 51 | # Partial C0: the Muse leg already fast-forwarded before this. |
| 52 | runner.calls.append((command, cwd)) |
| 53 | return CommandResult(stdout="", stderr="induced git tip failure", exit_code=1) |
| 54 | return original_run(command, cwd=cwd) |
| 55 | |
| 56 | runner.run = breaking_run # type: ignore[method-assign] |
| 57 | |
| 58 | for round_index in range(ROUNDS): |
| 59 | fail_muse_leg = round_index % 2 == 1 |
| 60 | # Re-strand the stale tips so both C0 legs stay active every round. |
| 61 | runner.git_tips[branch] = "gitstale" |
| 62 | runner.muse_tips[branch] = "sha256:stale" |
| 63 | |
| 64 | code = run_cli( |
| 65 | ["governance-sync", "--write"], cwd=tmp_path, runner=runner, kit=kit_root() |
| 66 | ) |
| 67 | assert code == 2, f"round {round_index}: expected exit 2 on induced C0 failure" |
| 68 | # §GSW.4.2 dual restore: no stranded HEAD on the dated sync branch. |
| 69 | assert runner.git_branch == "main", f"round {round_index}: stranded git HEAD" |
| 70 | assert runner.muse_branch == "main", f"round {round_index}: stranded muse HEAD" |
| 71 | assert handover_path.read_bytes() == original_handover |
| 72 | assert roadmap_path.read_bytes() == original_roadmap |
| 73 | assert not (tmp_path / ".overseer" / "last_governance_sync").exists() |
| 74 | |
| 75 | # No uniquified strand accumulation either: only the seeded dated branch |
| 76 | # plus main exist on each history. |
| 77 | assert {b for b in runner.git_branches if b.startswith("feat/governance-sync-")} == {branch} |
| 78 | assert {b for b in runner.muse_branches if b.startswith("feat/governance-sync-")} == {branch} |
| 79 | assert not any("--force" in c for c, _ in runner.calls) |