test_post_land_sync_integrity.py
file-level
1
files
1
commits
0
hotspots
0
🧊 dead
0
💥 blast risk
| 1 | """Data-integrity tier §PLS.10 — dirty bytes untouched; clean sync matches origin; ff-only.""" |
| 2 | |
| 3 | from __future__ import annotations |
| 4 | |
| 5 | import subprocess |
| 6 | from pathlib import Path |
| 7 | |
| 8 | import pytest |
| 9 | |
| 10 | from tests.support import gh_merged_runner, pls_config |
| 11 | from tools.close_ritual.pr_land import run_pr_land |
| 12 | |
| 13 | |
| 14 | def _git(cwd: Path, *args: str) -> str: |
| 15 | completed = subprocess.run( |
| 16 | ["git", *args], cwd=str(cwd), capture_output=True, text=True, check=True |
| 17 | ) |
| 18 | return completed.stdout.strip() |
| 19 | |
| 20 | |
| 21 | @pytest.fixture |
| 22 | def land_repos(tmp_path: Path) -> tuple[Path, Path]: |
| 23 | origin = tmp_path / "origin.git" |
| 24 | origin.mkdir() |
| 25 | _git(origin, "init", "--bare", "-b", "main") |
| 26 | |
| 27 | work = tmp_path / "work" |
| 28 | _git(tmp_path, "clone", str(origin), "work") |
| 29 | _git(work, "config", "user.email", "[email protected]") |
| 30 | _git(work, "config", "user.name", "test") |
| 31 | (work / "tracked.md").write_text("v1\n", encoding="utf-8") |
| 32 | (work / ".gitignore").write_text(".overseer/\n", encoding="utf-8") |
| 33 | _git(work, "add", ".") |
| 34 | _git(work, "commit", "-m", "seed") |
| 35 | _git(work, "push", "-u", "origin", "main") |
| 36 | |
| 37 | other = tmp_path / "other" |
| 38 | _git(tmp_path, "clone", str(origin), "other") |
| 39 | _git(other, "config", "user.email", "[email protected]") |
| 40 | _git(other, "config", "user.name", "test") |
| 41 | (other / "tracked.md").write_text("v2 landed\n", encoding="utf-8") |
| 42 | _git(other, "add", ".") |
| 43 | _git(other, "commit", "-m", "landed PR") |
| 44 | _git(other, "push", "origin", "main") |
| 45 | |
| 46 | (work / ".overseer").mkdir() |
| 47 | return work, origin |
| 48 | |
| 49 | |
| 50 | def test_dirty_unique_bytes_unchanged_after_skipped_sync( |
| 51 | land_repos: tuple[Path, Path], |
| 52 | ) -> None: |
| 53 | work, _origin = land_repos |
| 54 | unique = "UNIQUE-LOCAL-CONTENT-9c2f\n" |
| 55 | (work / "tracked.md").write_text(unique, encoding="utf-8") |
| 56 | |
| 57 | result = run_pr_land( |
| 58 | "3", |
| 59 | authorization="operator: integrity dirty", |
| 60 | runner=gh_merged_runner(), |
| 61 | sleep_fn=lambda _s: None, |
| 62 | repo_root=work, |
| 63 | config=pls_config(work), |
| 64 | ) |
| 65 | |
| 66 | assert result.post_land_sync["status"] == "skipped_dirty" |
| 67 | assert (work / "tracked.md").read_text(encoding="utf-8") == unique |
| 68 | |
| 69 | |
| 70 | def test_clean_sync_bytes_equal_origin_main_no_invented_commits( |
| 71 | land_repos: tuple[Path, Path], |
| 72 | ) -> None: |
| 73 | work, _origin = land_repos |
| 74 | origin_tip = _git(work, "ls-remote", "origin", "main").split()[0] |
| 75 | |
| 76 | result = run_pr_land( |
| 77 | "3", |
| 78 | authorization="operator: integrity clean", |
| 79 | runner=gh_merged_runner(), |
| 80 | sleep_fn=lambda _s: None, |
| 81 | repo_root=work, |
| 82 | config=pls_config(work), |
| 83 | ) |
| 84 | |
| 85 | assert result.post_land_sync["status"] == "synced" |
| 86 | # Working-tree bytes for a tracked file equal origin/main after pull. |
| 87 | assert (work / "tracked.md").read_text(encoding="utf-8") == "v2 landed\n" |
| 88 | # ff-only: local main is exactly the origin tip — no merge/invented commits. |
| 89 | assert _git(work, "rev-parse", "main") == origin_tip |
| 90 | assert _git(work, "rev-parse", "main") == _git(work, "rev-parse", "origin/main") |