test_land_closeout.py
python
sha256:a78e7e5a8740e03315f325d19edeb3aa1b306b3337d04abbaa9a9e0f3bbeb7a1
docs: MuseHub-first before ISR #74 — staging solidify NEXT
Human
1 day ago
| 1 | """Unit tests for land-closeout resolution (§PMHF.10 unit). |
| 2 | |
| 3 | Covers the frozen ``check_land_closeout`` resolution table, the §PMHF.4.2 |
| 4 | vocabulary fallback (including the frozen false-positive exclusions), the |
| 5 | ``land_phase_conflicts_queue_done`` token-intersection check, and the |
| 6 | ``ok`` vs ``land_complete`` distinction. |
| 7 | """ |
| 8 | |
| 9 | from __future__ import annotations |
| 10 | |
| 11 | from pathlib import Path |
| 12 | from unittest.mock import MagicMock |
| 13 | |
| 14 | from adapters.runner import CommandResult, RecordingRunner |
| 15 | from adapters.types import AnchorResult, HeadResult, StatusResult |
| 16 | from tests.support import ( |
| 17 | LAND_A_MARKER, |
| 18 | LAND_B_MARKER, |
| 19 | land_a_fence_body, |
| 20 | land_b_fence_body, |
| 21 | land_handover_text, |
| 22 | land_roadmap_text, |
| 23 | load_fixture_config, |
| 24 | seed_land_repo, |
| 25 | write_config, |
| 26 | ) |
| 27 | from tools.governance_hygiene.next_regen import ( |
| 28 | LAND_PHASE_A, |
| 29 | LAND_PHASE_B, |
| 30 | LAND_PHASE_UNREADABLE, |
| 31 | extract_land_id, |
| 32 | land_queue_conflict, |
| 33 | resolve_land_phase, |
| 34 | set_marker_land_phase, |
| 35 | strip_land_parenthetical, |
| 36 | ) |
| 37 | from tools.land_closeout import check_land_closeout, land_complete |
| 38 | |
| 39 | |
| 40 | def _adapter(*, tip: str = "cafebabe") -> MagicMock: |
| 41 | adapter = MagicMock() |
| 42 | adapter.status.return_value = StatusResult( |
| 43 | regime="git-only", dirty=False, branch="main", muse_dirty=None, git_dirty=False |
| 44 | ) |
| 45 | adapter.read_head.return_value = HeadResult(sha=tip, kind="git") |
| 46 | adapter.read_canonical_anchor.return_value = AnchorResult( |
| 47 | anchor_sha=tip, source="origin/main" |
| 48 | ) |
| 49 | return adapter |
| 50 | |
| 51 | |
| 52 | def _check(tmp_path: Path, **kwargs): |
| 53 | config = load_fixture_config(tmp_path, "config-git-only.yaml") |
| 54 | adapter = kwargs.pop("adapter", None) or _adapter(tip=kwargs.pop("tip", "cafebabe")) |
| 55 | return check_land_closeout(config, tmp_path, adapter=adapter, **kwargs) |
| 56 | |
| 57 | |
| 58 | # --- resolution table (§PMHF.5.2) --- |
| 59 | |
| 60 | |
| 61 | def test_not_applicable_without_lock(tmp_path: Path) -> None: |
| 62 | write_config(tmp_path, "config-git-only.yaml") |
| 63 | report = _check(tmp_path) |
| 64 | assert report.state == "not_applicable" |
| 65 | assert report.ok |
| 66 | assert land_complete(report) |
| 67 | |
| 68 | |
| 69 | def test_not_applicable_when_no_land_posture(tmp_path: Path) -> None: |
| 70 | seed_land_repo( |
| 71 | tmp_path, |
| 72 | handover_text=land_handover_text( |
| 73 | marker=None, |
| 74 | heading="PMHF-b Build", |
| 75 | fence_body="Model: Auto\nID: PMHF-b\n\nBuild the thing.\n", |
| 76 | ), |
| 77 | ) |
| 78 | report = _check(tmp_path) |
| 79 | assert report.state == "not_applicable" |
| 80 | assert report.land_phase is None |
| 81 | assert report.ok |
| 82 | |
| 83 | |
| 84 | def test_land_a_in_progress_when_aligned(tmp_path: Path) -> None: |
| 85 | seed_land_repo(tmp_path) |
| 86 | report = _check(tmp_path) |
| 87 | assert report.state == "land_a_in_progress" |
| 88 | assert report.land_phase == "land-a" |
| 89 | assert report.ok # status floor must not fail while waiting for merge |
| 90 | assert not land_complete(report) # …but land is NOT complete (§PMHF.3.3) |
| 91 | |
| 92 | |
| 93 | def test_post_merge_incomplete_when_d1_drifted(tmp_path: Path) -> None: |
| 94 | seed_land_repo(tmp_path, claim="deadbeef") |
| 95 | report = _check(tmp_path, tip="cafebabe") |
| 96 | assert report.state == "post_merge_incomplete" |
| 97 | assert not report.ok |
| 98 | assert report.d1 == "drifted" |
| 99 | assert (report.remediation or "").startswith( |
| 100 | "land-b required: ok governance-sync --dry-run then apply; " |
| 101 | "paste land-b; do not re-paste land-a" |
| 102 | ) |
| 103 | |
| 104 | |
| 105 | def test_post_merge_incomplete_when_stale_marker(tmp_path: Path) -> None: |
| 106 | seed_land_repo(tmp_path, marker_tip=None) # D1 aligned; sync marker absent |
| 107 | report = _check(tmp_path) |
| 108 | assert report.state == "post_merge_incomplete" |
| 109 | assert not report.ok |
| 110 | |
| 111 | |
| 112 | def test_land_b_in_progress_when_drifted(tmp_path: Path) -> None: |
| 113 | seed_land_repo( |
| 114 | tmp_path, |
| 115 | claim="deadbeef", |
| 116 | handover_text=land_handover_text( |
| 117 | claim="deadbeef", |
| 118 | marker=LAND_B_MARKER, |
| 119 | heading="PMHF land-b (post-merge sync)", |
| 120 | fence_body=land_b_fence_body(), |
| 121 | ), |
| 122 | ) |
| 123 | report = _check(tmp_path, tip="cafebabe") |
| 124 | assert report.state == "land_b_in_progress" |
| 125 | assert not report.ok |
| 126 | assert "governance-sync" in (report.remediation or "") |
| 127 | |
| 128 | |
| 129 | def test_complete_when_land_b_and_fresh(tmp_path: Path) -> None: |
| 130 | seed_land_repo( |
| 131 | tmp_path, |
| 132 | handover_text=land_handover_text( |
| 133 | marker=LAND_B_MARKER, |
| 134 | heading="PMHF land-b (post-merge sync)", |
| 135 | fence_body=land_b_fence_body(), |
| 136 | ), |
| 137 | ) |
| 138 | report = _check(tmp_path) |
| 139 | assert report.state == "complete" |
| 140 | assert report.ok |
| 141 | assert land_complete(report) |
| 142 | |
| 143 | |
| 144 | def test_unreadable_on_unknown_land_phase_value(tmp_path: Path) -> None: |
| 145 | bad_marker = LAND_A_MARKER.replace("land-phase=land-a", "land-phase=weird") |
| 146 | seed_land_repo(tmp_path, handover_text=land_handover_text(marker=bad_marker)) |
| 147 | report = _check(tmp_path) |
| 148 | assert report.state == "unreadable" |
| 149 | assert not report.ok |
| 150 | assert not land_complete(report) |
| 151 | |
| 152 | |
| 153 | def test_unreadable_freshness_not_masked_as_post_merge_incomplete(tmp_path: Path) -> None: |
| 154 | seed_land_repo(tmp_path) |
| 155 | adapter = MagicMock() |
| 156 | from adapters.errors import ReadError |
| 157 | |
| 158 | adapter.status.return_value = ReadError("git status", "boom") |
| 159 | report = _check(tmp_path, adapter=adapter) |
| 160 | assert report.state == "unreadable" # R2-M2: never masked |
| 161 | |
| 162 | |
| 163 | # --- §PMHF.4 marker + vocabulary fallback --- |
| 164 | |
| 165 | |
| 166 | def test_vocabulary_fallback_wait_for_merge_is_land_a(tmp_path: Path) -> None: |
| 167 | text = land_handover_text( |
| 168 | claim="deadbeef", |
| 169 | marker=None, |
| 170 | fence_body=( |
| 171 | "Model: Operator + Auto\nID: PMHF → main\n\n" |
| 172 | "Open PR #206 and wait for merge before continuing.\n" |
| 173 | ), |
| 174 | ) |
| 175 | assert resolve_land_phase(text) == LAND_PHASE_A |
| 176 | seed_land_repo(tmp_path, claim="deadbeef", handover_text=text) |
| 177 | report = _check(tmp_path, tip="cafebabe") |
| 178 | assert report.state == "post_merge_incomplete" |
| 179 | |
| 180 | |
| 181 | def test_bare_open_pr_alone_does_not_trigger_land_a() -> None: |
| 182 | text = land_handover_text( |
| 183 | marker=None, |
| 184 | heading="PMHF-b Build", |
| 185 | fence_body=( |
| 186 | "Model: Auto\nID: PMHF-b\n\n" |
| 187 | "Deliver:\n1. Build feature\n2. open PR from the feature branch\n" |
| 188 | "3. open/update PR as needed (Tier 3 applies to main)\n" |
| 189 | ), |
| 190 | ) |
| 191 | assert resolve_land_phase(text) is None |
| 192 | |
| 193 | |
| 194 | def test_conflicting_vocabulary_is_unreadable() -> None: |
| 195 | text = land_handover_text( |
| 196 | marker=None, |
| 197 | fence_body=( |
| 198 | "Model: Auto\nID: PMHF\n\n" |
| 199 | "wait for merge, then run land-b (post-merge sync)\n" |
| 200 | ), |
| 201 | ) |
| 202 | assert resolve_land_phase(text) == LAND_PHASE_UNREADABLE |
| 203 | |
| 204 | |
| 205 | def test_marker_attribute_beats_vocabulary(tmp_path: Path) -> None: |
| 206 | # Marker says land-b; fence still carries land-a vocabulary. |
| 207 | text = land_handover_text( |
| 208 | marker=LAND_B_MARKER, |
| 209 | fence_body="Model: Auto\nID: PMHF land-b (post-merge sync)\n\nwait for merge\n", |
| 210 | ) |
| 211 | assert resolve_land_phase(text) == LAND_PHASE_B |
| 212 | seed_land_repo(tmp_path, handover_text=text) |
| 213 | report = _check(tmp_path) |
| 214 | assert report.state == "complete" |
| 215 | |
| 216 | |
| 217 | def test_set_marker_land_phase_roundtrip() -> None: |
| 218 | text = land_handover_text() |
| 219 | cleared = set_marker_land_phase(text, None) |
| 220 | assert "land-phase=" not in cleared |
| 221 | stamped = set_marker_land_phase(cleared, "land-b") |
| 222 | assert resolve_land_phase(stamped) == LAND_PHASE_B |
| 223 | |
| 224 | |
| 225 | # --- §PMHF.3.3 queue conflict --- |
| 226 | |
| 227 | |
| 228 | def test_land_id_extraction_strips_parenthetical() -> None: |
| 229 | text = land_handover_text() |
| 230 | land_id = extract_land_id(text) |
| 231 | assert land_id == "PMHF → main (land-a)" |
| 232 | assert strip_land_parenthetical(land_id) == "PMHF → main" |
| 233 | |
| 234 | |
| 235 | def test_land_phase_conflicts_queue_done_token(tmp_path: Path) -> None: |
| 236 | seed_land_repo( |
| 237 | tmp_path, |
| 238 | roadmap_text=land_roadmap_text( |
| 239 | "| **PMHF → main** | Operator + Auto | **DONE** | Land PMHF |", |
| 240 | ), |
| 241 | ) |
| 242 | report = _check(tmp_path) |
| 243 | assert report.state == "unreadable" |
| 244 | assert "land_phase_conflicts_queue_done" in report.message |
| 245 | assert not report.ok |
| 246 | |
| 247 | |
| 248 | def test_historical_other_slice_land_row_does_not_conflict(tmp_path: Path) -> None: |
| 249 | seed_land_repo( |
| 250 | tmp_path, |
| 251 | roadmap_text=land_roadmap_text( |
| 252 | "| **GS-PASTE → main** | Operator + Auto | **DONE** | Landed earlier |", |
| 253 | "| **GFG → main** | Operator + Auto | **MERGED** | Landed earlier |", |
| 254 | "| **PMHF → main** | Operator + Auto | **TODO** | Land PMHF |", |
| 255 | ), |
| 256 | ) |
| 257 | report = _check(tmp_path) |
| 258 | assert report.state == "land_a_in_progress" |
| 259 | |
| 260 | |
| 261 | def test_hyphen_fragment_alone_does_not_conflict_with_other_slice() -> None: |
| 262 | """``GSW-FIX → main`` land-a must not conflict with historical ``GFG-D2-FIX → main``. |
| 263 | |
| 264 | ``phase_tokens`` splits hyphens, so a naive set intersection shares ``FIX``. |
| 265 | Slice-identifying match requires a compound token, not a bare fragment. |
| 266 | """ |
| 267 | roadmap = land_roadmap_text( |
| 268 | "| **GFG-D2-FIX → main** | Operator + Auto | **DONE** | Landed earlier |", |
| 269 | "| **GSW-FIX → main** | Operator + Auto | **TODO** | Land GSW-FIX |", |
| 270 | ) |
| 271 | assert not land_queue_conflict(roadmap, "GSW-FIX → main (land-a)") |
| 272 | assert land_queue_conflict( |
| 273 | land_roadmap_text("| **GSW-FIX → main** | Operator + Auto | **DONE** | landed |"), |
| 274 | "GSW-FIX → main (land-a)", |
| 275 | ) |
| 276 | |
| 277 | |
| 278 | def test_queue_conflict_requires_land_shaped_row() -> None: |
| 279 | # A DONE build row sharing the slice token is not a land row — no conflict. |
| 280 | roadmap = land_roadmap_text( |
| 281 | "| **PMHF-b Build** | Auto | **DONE** | build |", |
| 282 | "| **PMHF → main** | Operator + Auto | **TODO** | Land PMHF |", |
| 283 | ) |
| 284 | assert not land_queue_conflict(roadmap, "PMHF → main (land-a)") |
| 285 | assert land_queue_conflict( |
| 286 | land_roadmap_text("| **PMHF → main** | Operator + Auto | **DONE** | landed |"), |
| 287 | "PMHF → main (land-a)", |
| 288 | ) |
| 289 | |
| 290 | |
| 291 | # --- §PMHF.5.3 optional merged-PR enrichment --- |
| 292 | |
| 293 | |
| 294 | def _pr_handover(claim: str = "cafebabe") -> str: |
| 295 | return land_handover_text( |
| 296 | claim=claim, |
| 297 | fence_body=land_a_fence_body(paste_extra="PR #206 open — waiting for merge.\n"), |
| 298 | ) |
| 299 | |
| 300 | |
| 301 | def test_probe_merged_pr_catches_hand_edited_alignment(tmp_path: Path) -> None: |
| 302 | seed_land_repo(tmp_path, handover_text=_pr_handover()) |
| 303 | runner = RecordingRunner( |
| 304 | responses={ |
| 305 | "gh pr view 206 --json state,mergedAt": CommandResult( |
| 306 | stdout='{"state": "MERGED", "mergedAt": "2026-07-30T12:00:00Z"}', |
| 307 | stderr="", |
| 308 | exit_code=0, |
| 309 | ) |
| 310 | }, |
| 311 | calls=[], |
| 312 | ) |
| 313 | report = _check(tmp_path, runner=runner, probe_merged_pr=True) |
| 314 | assert report.optional_pr_merged is True |
| 315 | assert report.state == "post_merge_incomplete" |
| 316 | |
| 317 | |
| 318 | def test_probe_merged_pr_gh_failure_never_fails_open(tmp_path: Path) -> None: |
| 319 | seed_land_repo(tmp_path, handover_text=_pr_handover()) |
| 320 | runner = RecordingRunner(responses={}, calls=[]) # gh unavailable |
| 321 | report = _check(tmp_path, runner=runner, probe_merged_pr=True) |
| 322 | assert report.optional_pr_merged is None |
| 323 | assert report.state == "land_a_in_progress" # not complete, not fabricated |
| 324 | |
| 325 | |
| 326 | def test_no_gh_when_probe_disabled(tmp_path: Path) -> None: |
| 327 | seed_land_repo(tmp_path, handover_text=_pr_handover()) |
| 328 | runner = RecordingRunner(responses={}, calls=[]) |
| 329 | report = _check(tmp_path, runner=runner, probe_merged_pr=False) |
| 330 | assert report.optional_pr_merged is None |
| 331 | assert not any("gh" in call[0] for call in runner.calls) |
| 332 | |
| 333 | |
| 334 | # --- ok vs land_complete (§PMHF.3.3) --- |
| 335 | |
| 336 | |
| 337 | def test_ok_vs_land_complete_distinction(tmp_path: Path) -> None: |
| 338 | seed_land_repo(tmp_path) |
| 339 | report = _check(tmp_path) |
| 340 | assert report.state == "land_a_in_progress" |
| 341 | assert report.ok is True |
| 342 | assert land_complete(report) is False |
File History
1 commit
sha256:a78e7e5a8740e03315f325d19edeb3aa1b306b3337d04abbaa9a9e0f3bbeb7a1
docs: MuseHub-first before ISR #74 — staging solidify NEXT
Human
1 day ago