test_post_land_sync_unit.py
python
sha256:0e9549ec7b463911bc08b7d586dc320b1ac9b1f5c943ee7e3865dcc6cb0f6f83
chore(governance): sync handover+roadmap to 84db8c8 (drift:…
Human
2 days ago
| 1 | """Unit tier §PLS.10 — config parse, trigger matrix, dirty skip, checkout, exit 36.""" |
| 2 | |
| 3 | from __future__ import annotations |
| 4 | |
| 5 | from pathlib import Path |
| 6 | |
| 7 | import pytest |
| 8 | |
| 9 | from adapters.config import PostLandSyncConfig, load_config |
| 10 | from adapters.errors import ConfigError |
| 11 | from tests.support import FIXTURES, FakeGitRunner, gh_merged_runner, pls_config |
| 12 | from tools.close_ritual.post_land_sync import run_post_land_sync |
| 13 | from tools.close_ritual.pr_land import ( |
| 14 | EXIT_CHECKS_FAILED, |
| 15 | EXIT_OK, |
| 16 | EXIT_POST_LAND_SYNC, |
| 17 | run_pr_land, |
| 18 | ) |
| 19 | |
| 20 | |
| 21 | def _write_config(repo_root: Path, close_ritual_block: str) -> Path: |
| 22 | base = (FIXTURES / "config-git-only.yaml").read_text(encoding="utf-8") |
| 23 | dest = repo_root / ".overseer" / "config.yaml" |
| 24 | dest.parent.mkdir(parents=True, exist_ok=True) |
| 25 | dest.write_text(base + "\n" + close_ritual_block, encoding="utf-8") |
| 26 | return dest |
| 27 | |
| 28 | |
| 29 | # --- Config parse (§PLS.3.2 — fail-closed) --- |
| 30 | |
| 31 | |
| 32 | def test_defaults_when_block_omitted(tmp_path: Path) -> None: |
| 33 | config = load_config(_write_config(tmp_path, "")) |
| 34 | assert config.close_ritual.post_land_sync == PostLandSyncConfig( |
| 35 | enabled=False, strategy="ff_only", require_clean_worktree=True |
| 36 | ) |
| 37 | |
| 38 | |
| 39 | def test_defaults_when_post_land_sync_omitted(tmp_path: Path) -> None: |
| 40 | config = load_config(_write_config(tmp_path, "close_ritual:\n enabled: false\n")) |
| 41 | assert config.close_ritual.post_land_sync.enabled is False |
| 42 | assert config.close_ritual.post_land_sync.strategy == "ff_only" |
| 43 | assert config.close_ritual.post_land_sync.require_clean_worktree is True |
| 44 | |
| 45 | |
| 46 | def test_parse_explicit_opt_in(tmp_path: Path) -> None: |
| 47 | config = load_config( |
| 48 | _write_config( |
| 49 | tmp_path, |
| 50 | "close_ritual:\n" |
| 51 | " post_land_sync:\n" |
| 52 | " enabled: true\n" |
| 53 | " strategy: ff_only\n" |
| 54 | " require_clean_worktree: true\n", |
| 55 | ) |
| 56 | ) |
| 57 | assert config.close_ritual.post_land_sync.enabled is True |
| 58 | |
| 59 | |
| 60 | @pytest.mark.parametrize( |
| 61 | "block, fragment", |
| 62 | [ |
| 63 | ( |
| 64 | "close_ritual:\n post_land_sync:\n surprise: 1\n", |
| 65 | "unknown close_ritual.post_land_sync keys", |
| 66 | ), |
| 67 | ( |
| 68 | "close_ritual:\n post_land_sync:\n strategy: rebase\n", |
| 69 | "strategy must be ff_only", |
| 70 | ), |
| 71 | ( |
| 72 | "close_ritual:\n post_land_sync:\n require_clean_worktree: false\n", |
| 73 | "require_clean_worktree must be true", |
| 74 | ), |
| 75 | ( |
| 76 | "close_ritual:\n post_land_sync:\n enabled: yes please\n", |
| 77 | "enabled must be a boolean", |
| 78 | ), |
| 79 | ( |
| 80 | "close_ritual:\n post_land_sync:\n require_clean_worktree: sometimes\n", |
| 81 | "require_clean_worktree must be true", |
| 82 | ), |
| 83 | ], |
| 84 | ) |
| 85 | def test_parse_rejects_bad_values(tmp_path: Path, block: str, fragment: str) -> None: |
| 86 | with pytest.raises(ConfigError) as excinfo: |
| 87 | load_config(_write_config(tmp_path, block)) |
| 88 | assert fragment in str(excinfo.value) |
| 89 | |
| 90 | |
| 91 | # --- Trigger matrix (§PLS.4.1) --- |
| 92 | |
| 93 | |
| 94 | def test_enabled_merged_not_dry_run_enters_helper(repo_root: Path) -> None: |
| 95 | git = FakeGitRunner(branch="main") |
| 96 | result = run_pr_land( |
| 97 | "1", |
| 98 | authorization="operator: land", |
| 99 | runner=gh_merged_runner(), |
| 100 | sleep_fn=lambda _s: None, |
| 101 | repo_root=repo_root, |
| 102 | config=pls_config(repo_root), |
| 103 | git_runner=git, |
| 104 | ) |
| 105 | assert result.exit_code == EXIT_OK |
| 106 | assert result.post_land_sync["status"] == "synced" |
| 107 | assert git.calls[0] == ["git", "fetch", "origin"] |
| 108 | |
| 109 | |
| 110 | def test_no_config_defaults_to_disabled_always_present(repo_root: Path) -> None: |
| 111 | result = run_pr_land( |
| 112 | "1", |
| 113 | authorization="operator: land", |
| 114 | runner=gh_merged_runner(), |
| 115 | sleep_fn=lambda _s: None, |
| 116 | ) |
| 117 | assert result.exit_code == EXIT_OK |
| 118 | assert result.post_land_sync == { |
| 119 | "status": "disabled", |
| 120 | "remote": "", |
| 121 | "main_branch": "", |
| 122 | "messages": [], |
| 123 | } |
| 124 | assert "post_land_sync" in result.to_dict() |
| 125 | |
| 126 | |
| 127 | def test_disabled_config_no_git_argv(repo_root: Path) -> None: |
| 128 | git = FakeGitRunner() |
| 129 | result = run_pr_land( |
| 130 | "1", |
| 131 | authorization="operator: land", |
| 132 | runner=gh_merged_runner(), |
| 133 | sleep_fn=lambda _s: None, |
| 134 | repo_root=repo_root, |
| 135 | config=pls_config(repo_root, enabled=False), |
| 136 | git_runner=git, |
| 137 | ) |
| 138 | assert result.post_land_sync["status"] == "disabled" |
| 139 | assert git.calls == [] |
| 140 | |
| 141 | |
| 142 | def test_dry_run_never_syncs(repo_root: Path) -> None: |
| 143 | git = FakeGitRunner() |
| 144 | result = run_pr_land( |
| 145 | "1", |
| 146 | authorization="operator: land", |
| 147 | dry_run=True, |
| 148 | runner=gh_merged_runner(), |
| 149 | sleep_fn=lambda _s: None, |
| 150 | repo_root=repo_root, |
| 151 | config=pls_config(repo_root), |
| 152 | git_runner=git, |
| 153 | ) |
| 154 | assert result.exit_code == EXIT_OK |
| 155 | assert result.merged is False |
| 156 | assert result.post_land_sync["status"] == "not_applicable" |
| 157 | assert git.calls == [] |
| 158 | |
| 159 | |
| 160 | def test_checks_failed_never_syncs(repo_root: Path) -> None: |
| 161 | git = FakeGitRunner() |
| 162 | result = run_pr_land( |
| 163 | "1", |
| 164 | authorization="operator: land", |
| 165 | runner=gh_merged_runner(check_state="fail"), |
| 166 | sleep_fn=lambda _s: None, |
| 167 | repo_root=repo_root, |
| 168 | config=pls_config(repo_root), |
| 169 | git_runner=git, |
| 170 | ) |
| 171 | assert result.exit_code == EXIT_CHECKS_FAILED |
| 172 | assert result.post_land_sync["status"] == "not_applicable" |
| 173 | assert git.calls == [] |
| 174 | |
| 175 | |
| 176 | def test_already_merged_checks_failed_never_syncs(repo_root: Path) -> None: |
| 177 | git = FakeGitRunner() |
| 178 | result = run_pr_land( |
| 179 | "1", |
| 180 | authorization="operator: land", |
| 181 | runner=gh_merged_runner(pr_state="MERGED", check_state="fail"), |
| 182 | sleep_fn=lambda _s: None, |
| 183 | repo_root=repo_root, |
| 184 | config=pls_config(repo_root), |
| 185 | git_runner=git, |
| 186 | ) |
| 187 | assert result.exit_code == EXIT_CHECKS_FAILED |
| 188 | assert result.already_merged is True |
| 189 | assert result.post_land_sync["status"] == "not_applicable" |
| 190 | assert git.calls == [] |
| 191 | |
| 192 | |
| 193 | def test_muse_only_regime_skipped_zero_git_argv(repo_root: Path) -> None: |
| 194 | git = FakeGitRunner() |
| 195 | result = run_pr_land( |
| 196 | "1", |
| 197 | authorization="operator: land", |
| 198 | runner=gh_merged_runner(), |
| 199 | sleep_fn=lambda _s: None, |
| 200 | repo_root=repo_root, |
| 201 | config=pls_config(repo_root, "config-muse-only.yaml"), |
| 202 | git_runner=git, |
| 203 | ) |
| 204 | assert result.exit_code == EXIT_OK |
| 205 | assert result.post_land_sync["status"] == "regime_skipped" |
| 206 | assert result.post_land_sync["remote"] == "" |
| 207 | assert result.post_land_sync["main_branch"] == "" |
| 208 | assert git.calls == [] |
| 209 | |
| 210 | |
| 211 | # --- Dirty skip / checkout / hard-fail (§PLS.4.2, §PLS.5, §PLS.6) --- |
| 212 | |
| 213 | |
| 214 | def test_dirty_porcelain_skips_without_checkout_or_pull(repo_root: Path) -> None: |
| 215 | git = FakeGitRunner(porcelain=" M docs/OVERSEER-HANDOVER.md\n?? scratch.txt\n") |
| 216 | report = run_post_land_sync( |
| 217 | repo_root=repo_root, |
| 218 | regime="git-only", |
| 219 | remote="origin", |
| 220 | main_branch="main", |
| 221 | git_runner=git, |
| 222 | ) |
| 223 | assert report.status == "skipped_dirty" |
| 224 | ops = [c[1] for c in git.calls] |
| 225 | assert "checkout" not in ops |
| 226 | assert "pull" not in ops |
| 227 | assert any("dirty" in m for m in report.messages) |
| 228 | |
| 229 | |
| 230 | def test_clean_on_feature_branch_checks_out_main_then_ff_pull(repo_root: Path) -> None: |
| 231 | git = FakeGitRunner(branch="feat/pls-a") |
| 232 | report = run_post_land_sync( |
| 233 | repo_root=repo_root, |
| 234 | regime="git-only", |
| 235 | remote="origin", |
| 236 | main_branch="main", |
| 237 | git_runner=git, |
| 238 | ) |
| 239 | assert report.status == "synced" |
| 240 | assert ["git", "checkout", "main"] in git.calls |
| 241 | checkout_idx = git.calls.index(["git", "checkout", "main"]) |
| 242 | pull_idx = git.calls.index(["git", "pull", "--ff-only", "origin", "main"]) |
| 243 | assert checkout_idx < pull_idx |
| 244 | |
| 245 | |
| 246 | @pytest.mark.parametrize("failing_op", ["fetch", "status", "checkout", "pull"]) |
| 247 | def test_hard_fail_maps_to_exit_36_not_6(repo_root: Path, failing_op: str) -> None: |
| 248 | git = FakeGitRunner(branch="feat/pls-a", fail={failing_op}) |
| 249 | result = run_pr_land( |
| 250 | "1", |
| 251 | authorization="operator: land", |
| 252 | runner=gh_merged_runner(), |
| 253 | sleep_fn=lambda _s: None, |
| 254 | repo_root=repo_root, |
| 255 | config=pls_config(repo_root), |
| 256 | git_runner=git, |
| 257 | ) |
| 258 | assert result.exit_code == EXIT_POST_LAND_SYNC |
| 259 | assert result.exit_code == 36 |
| 260 | assert result.exit_code != 6 |
| 261 | assert result.merged is True |
| 262 | assert result.post_land_sync["status"] == "failed" |
| 263 | |
| 264 | |
| 265 | def test_missing_repo_root_when_enabled_hard_fails(repo_root: Path) -> None: |
| 266 | git = FakeGitRunner() |
| 267 | result = run_pr_land( |
| 268 | "1", |
| 269 | authorization="operator: land", |
| 270 | runner=gh_merged_runner(), |
| 271 | sleep_fn=lambda _s: None, |
| 272 | repo_root=None, |
| 273 | config=pls_config(repo_root), |
| 274 | git_runner=git, |
| 275 | ) |
| 276 | assert result.exit_code == EXIT_POST_LAND_SYNC |
| 277 | assert result.post_land_sync["status"] == "failed" |
| 278 | assert git.calls == [] |
File History
1 commit
sha256:6abcf1fa82a7a621ccbc945f19acdba5bc0db54569599404a1452fb4a096a199
fix(ISR): default require_independent_second_reviewer to require
Human
minor
⚠
2 days ago