|
| 1 | +import json |
| 2 | +import os |
| 3 | +from pathlib import Path |
| 4 | +import subprocess |
| 5 | +import tempfile |
| 6 | +import unittest |
| 7 | + |
| 8 | +import yaml |
| 9 | + |
| 10 | + |
| 11 | +WORKFLOW = Path(__file__).resolve().parents[1] / ".github/workflows/breakglass-merge.yml" |
| 12 | + |
| 13 | +# Replace only the remote GitHub API; execute the workflow's real shell. |
| 14 | +FAKE_GH = r'''#!/usr/bin/env python3 |
| 15 | +import json, os, sys |
| 16 | +from pathlib import Path |
| 17 | +path = Path(os.environ["FAKE_STATE"]) |
| 18 | +state = json.loads(path.read_text()) |
| 19 | +args = sys.argv[2:] |
| 20 | +method = args[args.index("-X") + 1] if "-X" in args else "GET" |
| 21 | +endpoint = next(a for a in args if a.startswith("/repos/")).split("?", 1)[0] |
| 22 | +data = {} |
| 23 | +for i, arg in enumerate(args): |
| 24 | + if arg in ("-f", "-F"): |
| 25 | + key, value = args[i + 1].split("=", 1) |
| 26 | + data[key] = json.loads(value) if arg == "-F" and value in ("true", "false") else value |
| 27 | +parts = endpoint.split("/") |
| 28 | +number = int(parts[5]) if len(parts) > 5 else None |
| 29 | +result = {} |
| 30 | +error = None |
| 31 | +if method == "GET" and len(parts) == 4: |
| 32 | + result = state["repo"] |
| 33 | +elif method == "GET" and parts[4] == "stacks": |
| 34 | + order = state.get("stack_order", [int(n) for n in state["prs"]]) |
| 35 | + result = {"pull_requests": [state["prs"][str(n)] for n in order]} |
| 36 | +elif method == "GET" and "/merge-async/" in endpoint: |
| 37 | + state["polls"] = state.get("polls", 0) + 1 |
| 38 | + status = "pending" if state["polls"] < state.get("complete_after", 1) else state.get("terminal_status", "merged") |
| 39 | + if status == "merged": |
| 40 | + state["prs"][str(number)]["state"] = "closed" |
| 41 | + state["merged"] = [number] |
| 42 | + result = {"status": status, "details": {"message": "merge conflict" if status == "failed" else status}} |
| 43 | +elif method == "GET": |
| 44 | + key = str(number) |
| 45 | + state.setdefault("reads", {})[key] = state.get("reads", {}).get(key, 0) + 1 |
| 46 | + if state.get("change_head") == number and state["reads"][key] >= 2: |
| 47 | + state["prs"][key]["head"]["sha"] = "changed" |
| 48 | + if state.get("change_base") == number and state["reads"][key] >= 2: |
| 49 | + state["prs"][key]["base"]["ref"] = "unexpected" |
| 50 | + if state.get("change_stack") and state["reads"][key] >= 2: |
| 51 | + state["stack_order"] = [4, 1, 2, 3] |
| 52 | + state["prs"]["4"] = {"number": 4, "state": "open"} |
| 53 | + result = state["prs"][key] |
| 54 | +elif endpoint.endswith("/comments"): |
| 55 | + if state.get("fail_comment") == number: |
| 56 | + error = "comment unavailable" |
| 57 | + del state["fail_comment"] |
| 58 | + else: |
| 59 | + state.setdefault("comments", []).append({"number": number, **data}) |
| 60 | +elif endpoint.endswith("/merge-async"): |
| 61 | + state.setdefault("requests", []).append({"number": number, **data}) |
| 62 | + if state.get("submit_error"): |
| 63 | + error = "HTTP 403 forbidden" |
| 64 | + elif not any(c["number"] == number for c in state.get("comments", [])): |
| 65 | + error = "missing audit comment before merge" |
| 66 | + else: |
| 67 | + result = {"status": "pending", "details": {"uuid": "request-id", "expected_head_sha": data["sha"]}} |
| 68 | +elif endpoint.endswith("/merge"): |
| 69 | + error = "Merging stacked PRs via this endpoint is not supported. Use the asynchronous merge endpoint instead. (HTTP 403)" |
| 70 | +else: |
| 71 | + raise RuntimeError((method, endpoint, data)) |
| 72 | +path.write_text(json.dumps(state)) |
| 73 | +if error: |
| 74 | + print(error, file=sys.stderr) |
| 75 | + sys.exit(1) |
| 76 | +print(json.dumps(result)) |
| 77 | +''' |
| 78 | + |
| 79 | + |
| 80 | +def pr(number, head, base, stacked=True): |
| 81 | + return { |
| 82 | + "number": number, "state": "open", "draft": False, |
| 83 | + "stack": {"number": 42, "position": number, "base": {"ref": "main"}} if stacked else None, |
| 84 | + "author_association": "MEMBER", |
| 85 | + "head": {"ref": head, "sha": f"sha-{number}", "repo": {"full_name": "kernel/example"}}, |
| 86 | + "base": {"ref": base, "repo": {"full_name": "kernel/example"}}, |
| 87 | + } |
| 88 | + |
| 89 | + |
| 90 | +class BreakglassTests(unittest.TestCase): |
| 91 | + def run_workflow(self, *, body="/breakglass emergency fix needed", association="MEMBER", number=1, **changes): |
| 92 | + state = { |
| 93 | + "repo": {"default_branch": "main", "allow_merge_commit": True, "allow_squash_merge": True}, |
| 94 | + "prs": {"1": pr(1, "first", "main"), "2": pr(2, "second", "first"), "3": pr(3, "third", "second")}, |
| 95 | + **changes, |
| 96 | + } |
| 97 | + workflow = yaml.safe_load(WORKFLOW.read_text()) |
| 98 | + steps = workflow["jobs"]["merge"]["steps"] |
| 99 | + merge = next(s["run"] for s in steps if s["name"] == "Merge without approval") |
| 100 | + refusal = next(s["run"] for s in steps if s["name"] == "Explain the refusal") |
| 101 | + with tempfile.TemporaryDirectory() as tmp: |
| 102 | + root = Path(tmp) |
| 103 | + (root / "gh").write_text(FAKE_GH) |
| 104 | + (root / "gh").chmod(0o755) |
| 105 | + (root / "sleep").write_text("#!/bin/sh\nexit 0\n") |
| 106 | + (root / "sleep").chmod(0o755) |
| 107 | + (root / "state.json").write_text(json.dumps(state)) |
| 108 | + env = { |
| 109 | + **os.environ, "PATH": f"{root}:{os.environ['PATH']}", |
| 110 | + "FAKE_STATE": str(root / "state.json"), "ACTOR": "operator", |
| 111 | + "REPO": "kernel/example", "NUMBER": str(number), "BODY": body, |
| 112 | + "REQUESTER_ASSOCIATION": association, "GITHUB_STEP_SUMMARY": str(root / "summary"), |
| 113 | + "RUN_URL": "https://example.com/run", |
| 114 | + } |
| 115 | + result = subprocess.run(["bash", "-c", merge], env=env, cwd=root, capture_output=True, text=True) |
| 116 | + if result.returncode: |
| 117 | + subprocess.run(["bash", "-c", refusal], env=env, cwd=root, check=True, capture_output=True) |
| 118 | + return result, json.loads((root / "state.json").read_text()) |
| 119 | + |
| 120 | + def test_merges_only_bottom_pr_using_async_endpoint(self): |
| 121 | + result, state = self.run_workflow(complete_after=2) |
| 122 | + self.assertEqual(result.returncode, 0, result.stderr) |
| 123 | + self.assertEqual(state["merged"], [1]) |
| 124 | + self.assertEqual(state["requests"], [{"number": 1, "merge_method": "squash", "sha": "sha-1", "merge_action": "direct_merge", "bypass_rules": True}]) |
| 125 | + self.assertEqual(state["prs"]["2"]["state"], "open") |
| 126 | + self.assertEqual(state["prs"]["3"]["state"], "open") |
| 127 | + self.assertEqual([c["number"] for c in state["comments"]], [1]) |
| 128 | + self.assertIn("emergency fix needed", state["comments"][0]["body"]) |
| 129 | + |
| 130 | + def test_refuses_higher_pr_with_open_pr_below_it(self): |
| 131 | + result, state = self.run_workflow(number=2) |
| 132 | + self.assertNotEqual(result.returncode, 0) |
| 133 | + self.assertNotIn("requests", state) |
| 134 | + self.assertNotIn("merged", state) |
| 135 | + self.assertIn("below this one first: #1", state["comments"][-1]["body"]) |
| 136 | + |
| 137 | + def test_allows_next_pr_after_lower_pr_has_merged(self): |
| 138 | + prs = {"1": pr(1, "first", "main"), "2": pr(2, "second", "main"), "3": pr(3, "third", "second")} |
| 139 | + prs["1"].update(state="closed", merged_at="2026-01-01T00:00:00Z") |
| 140 | + result, state = self.run_workflow(number=2, prs=prs) |
| 141 | + self.assertEqual(result.returncode, 0, result.stderr) |
| 142 | + self.assertEqual(state["merged"], [2]) |
| 143 | + self.assertEqual(state["prs"]["3"]["state"], "open") |
| 144 | + |
| 145 | + def test_unstacked_pr_uses_same_async_merge_path(self): |
| 146 | + result, state = self.run_workflow(prs={"1": pr(1, "first", "main", stacked=False)}) |
| 147 | + self.assertEqual(result.returncode, 0, result.stderr) |
| 148 | + self.assertEqual(state["merged"], [1]) |
| 149 | + self.assertEqual(state["requests"][0]["merge_method"], "squash") |
| 150 | + |
| 151 | + def test_rejects_unauthorized_requests_without_merge_submission(self): |
| 152 | + for case in ("requester", "fork", "author", "short_reason", "audit_failure"): |
| 153 | + with self.subTest(case=case): |
| 154 | + changes = {} |
| 155 | + prs = {"1": pr(1, "first", "main")} |
| 156 | + if case == "requester": |
| 157 | + changes["association"] = "CONTRIBUTOR" |
| 158 | + elif case == "fork": |
| 159 | + prs["1"]["head"]["repo"]["full_name"] = "external/example" |
| 160 | + elif case == "author": |
| 161 | + prs["1"]["author_association"] = "CONTRIBUTOR" |
| 162 | + elif case == "short_reason": |
| 163 | + changes["body"] = "/breakglass short" |
| 164 | + elif case == "audit_failure": |
| 165 | + changes["fail_comment"] = 1 |
| 166 | + result, state = self.run_workflow(prs=prs, **changes) |
| 167 | + self.assertNotEqual(result.returncode, 0) |
| 168 | + self.assertNotIn("requests", state) |
| 169 | + if case == "requester": |
| 170 | + self.assertNotIn("comments", state) |
| 171 | + |
| 172 | + def test_refuses_moved_head_base_or_new_downstack_pr(self): |
| 173 | + for change in ("change_head", "change_base", "change_stack"): |
| 174 | + with self.subTest(change=change): |
| 175 | + result, state = self.run_workflow(**{change: 1}) |
| 176 | + self.assertNotEqual(result.returncode, 0) |
| 177 | + self.assertNotIn("requests", state) |
| 178 | + self.assertNotIn("merged", state) |
| 179 | + |
| 180 | + def test_reports_terminal_failure_and_does_not_claim_success(self): |
| 181 | + result, state = self.run_workflow(terminal_status="failed") |
| 182 | + self.assertNotEqual(result.returncode, 0) |
| 183 | + self.assertNotIn("merged", state) |
| 184 | + self.assertIn("Could not merge: merge conflict", state["comments"][-1]["body"]) |
| 185 | + |
| 186 | + def test_timeout_or_queue_reports_unconfirmed_merge(self): |
| 187 | + for changes in ({"complete_after": 100}, {"terminal_status": "enqueued"}, {"submit_error": True}): |
| 188 | + with self.subTest(changes=changes): |
| 189 | + result, state = self.run_workflow(**changes) |
| 190 | + self.assertNotEqual(result.returncode, 0) |
| 191 | + self.assertNotIn("merged", state) |
| 192 | + body = state["comments"][-1]["body"] |
| 193 | + self.assertIn("merge not confirmed", body) |
| 194 | + self.assertNotIn("Nothing was merged", body) |
| 195 | + |
| 196 | + def test_ignores_other_commands_with_breakglass_prefix(self): |
| 197 | + result, state = self.run_workflow(body="/breakglass-stack emergency fix needed") |
| 198 | + self.assertEqual(result.returncode, 0, result.stderr) |
| 199 | + self.assertNotIn("requests", state) |
| 200 | + self.assertNotIn("comments", state) |
| 201 | + |
| 202 | + |
| 203 | +if __name__ == "__main__": |
| 204 | + unittest.main() |
0 commit comments