test_close_ritual_pr_land.py
python
sha256:0e9549ec7b463911bc08b7d586dc320b1ac9b1f5c943ee7e3865dcc6cb0f6f83
chore(governance): sync handover+roadmap to 84db8c8 (drift:…
Human
2 days ago
| 1 | """Unit + integration tests for close_ritual pr-land.""" |
| 2 | |
| 3 | from __future__ import annotations |
| 4 | |
| 5 | import json |
| 6 | import subprocess |
| 7 | import unittest |
| 8 | from unittest.mock import patch |
| 9 | |
| 10 | from tools.close_ritual.pr_land import ( |
| 11 | EXIT_CHECKS_FAILED, |
| 12 | EXIT_OK, |
| 13 | EXIT_UNAUTHORIZED, |
| 14 | run_pr_land, |
| 15 | ) |
| 16 | |
| 17 | |
| 18 | def _checks_tsv(*rows: tuple[str, str]) -> str: |
| 19 | return "\n".join(f"{name}\t{state}\t0s\thttps://example.test/{name}" for name, state in rows) + "\n" |
| 20 | |
| 21 | |
| 22 | class PrLandAuthorizationTests(unittest.TestCase): |
| 23 | def test_refuses_without_authorization(self) -> None: |
| 24 | result = run_pr_land("1", authorization="") |
| 25 | self.assertEqual(result.exit_code, EXIT_UNAUTHORIZED) |
| 26 | self.assertFalse(result.auto_merge) |
| 27 | |
| 28 | def test_merges_when_green(self) -> None: |
| 29 | def runner(cmd: list[str]) -> subprocess.CompletedProcess[str]: |
| 30 | if cmd[:3] == ["gh", "pr", "view"]: |
| 31 | return subprocess.CompletedProcess( |
| 32 | cmd, 0, stdout=json.dumps({"number": 1, "state": "OPEN"}), stderr="" |
| 33 | ) |
| 34 | if cmd[:3] == ["gh", "pr", "checks"]: |
| 35 | return subprocess.CompletedProcess( |
| 36 | cmd, 0, stdout=_checks_tsv(("verify", "pass")), stderr="" |
| 37 | ) |
| 38 | if cmd[:3] == ["gh", "pr", "merge"]: |
| 39 | return subprocess.CompletedProcess(cmd, 0, stdout="", stderr="") |
| 40 | raise AssertionError(cmd) |
| 41 | |
| 42 | result = run_pr_land( |
| 43 | "1", |
| 44 | authorization="operator: land after green", |
| 45 | runner=runner, |
| 46 | sleep_fn=lambda _s: None, |
| 47 | ) |
| 48 | self.assertEqual(result.exit_code, EXIT_OK) |
| 49 | self.assertTrue(result.merged) |
| 50 | self.assertFalse(result.auto_merge) |
| 51 | |
| 52 | def test_failed_check_exit_2(self) -> None: |
| 53 | def runner(cmd: list[str]) -> subprocess.CompletedProcess[str]: |
| 54 | if cmd[:3] == ["gh", "pr", "view"]: |
| 55 | return subprocess.CompletedProcess( |
| 56 | cmd, 0, stdout=json.dumps({"number": 2, "state": "OPEN"}), stderr="" |
| 57 | ) |
| 58 | if cmd[:3] == ["gh", "pr", "checks"]: |
| 59 | return subprocess.CompletedProcess( |
| 60 | cmd, 1, stdout=_checks_tsv(("verify", "fail")), stderr="" |
| 61 | ) |
| 62 | raise AssertionError(cmd) |
| 63 | |
| 64 | result = run_pr_land( |
| 65 | "2", |
| 66 | authorization="operator: expect fail", |
| 67 | runner=runner, |
| 68 | sleep_fn=lambda _s: None, |
| 69 | ) |
| 70 | self.assertEqual(result.exit_code, EXIT_CHECKS_FAILED) |
| 71 | self.assertIn("verify", result.failing) |
| 72 | |
| 73 | |
| 74 | if __name__ == "__main__": |
| 75 | unittest.main() |
File History
1 commit
sha256:6abcf1fa82a7a621ccbc945f19acdba5bc0db54569599404a1452fb4a096a199
fix(ISR): default require_independent_second_reviewer to require
Human
minor
⚠
2 days ago