"""ORCH-082 FR-1 — merge_gate.ensure_open_pr: idempotent open-code-PR actor. Covers TC-01..05 / AC-2 / AC-6 / AC-7. The actor guarantees an open code-PR (``head==branch`` AND ``base=="main"``) exists before the deterministic ``merge_pr``, without ever creating a duplicate. Gitea HTTP is mocked; the actor honours the strict never-raise contract (any error -> ``("failed", reason)``). """ import pytest from src import merge_gate REPO = "orchestrator" BRANCH = "feature/ORCH-082-x" class _Resp: """Minimal httpx.Response stand-in (status_code + json/text).""" def __init__(self, status_code, payload=None, text=""): self.status_code = status_code self._payload = payload if payload is not None else [] self.text = text def json(self): return self._payload @pytest.fixture(autouse=True) def _settings(monkeypatch): monkeypatch.setattr(merge_gate.settings, "merge_pr_timeout_s", 5) monkeypatch.setattr(merge_gate.settings, "gitea_owner", "owner") monkeypatch.setattr(merge_gate.settings, "gitea_token", "tok") monkeypatch.setattr(merge_gate.settings, "gitea_url", "http://gitea.test") # ORCH-093: these tests target the HTTP create/race logic of ensure_open_pr. # The new already-in-main guard (_branch_fully_in_main) runs real git; pin it # to "commits beyond main" (False) so the create path is exercised as intended. # The guard itself has dedicated coverage (test_merge_gate.py TC-09/10/11). monkeypatch.setattr(merge_gate, "_branch_fully_in_main", lambda r, b: False) def _install_httpx(monkeypatch, get_resp, post_resp=None, record=None): """Patch merge_gate's lazily-imported httpx with stub get/post callables.""" import httpx def fake_get(url, *a, **k): if record is not None: record.append(("GET", url, k.get("params"))) return get_resp() if callable(get_resp) else get_resp def fake_post(url, *a, **k): if record is not None: record.append(("POST", url, k.get("json"))) if post_resp is None: raise AssertionError("POST must NOT be called") return post_resp() if callable(post_resp) else post_resp monkeypatch.setattr(httpx, "get", fake_get) monkeypatch.setattr(httpx, "post", fake_post) # --------------------------------------------------------------------------- # TC-01: no open code-PR -> POST creates one -> ("created", N); base==main filter. # --------------------------------------------------------------------------- def test_tc01_creates_pr_when_absent(monkeypatch): record = [] _install_httpx( monkeypatch, get_resp=_Resp(200, []), # no open PRs at all post_resp=_Resp(201, {"number": 42}), record=record, ) status, detail = merge_gate.ensure_open_pr(REPO, BRANCH) assert (status, detail) == ("created", "42") # POST body targets head=branch, base=main. post = [r for r in record if r[0] == "POST"][0] assert post[2]["head"] == BRANCH assert post[2]["base"] == "main" # --------------------------------------------------------------------------- # TC-02: an open code-PR (head==branch AND base==main) already exists -> existed, # POST is never called (no duplicate). # --------------------------------------------------------------------------- def test_tc02_existed_no_duplicate(monkeypatch): payload = [{"number": 7, "head": {"ref": BRANCH}, "base": {"ref": "main"}}] _install_httpx(monkeypatch, get_resp=_Resp(200, payload), post_resp=None) status, detail = merge_gate.ensure_open_pr(REPO, BRANCH) assert (status, detail) == ("existed", "7") # POST stub would raise if called # --------------------------------------------------------------------------- # TC-03 (AC-6): only a docs-PR (base != main) exists -> NOT a code-PR -> create on main. # --------------------------------------------------------------------------- def test_tc03_docs_pr_not_counted_creates_on_main(monkeypatch): record = [] # An open PR exists but onto a docs base, and another onto a different head. docs_payload = [ {"number": 9, "head": {"ref": BRANCH}, "base": {"ref": "docs/logs"}}, {"number": 10, "head": {"ref": "other/branch"}, "base": {"ref": "main"}}, ] _install_httpx( monkeypatch, get_resp=_Resp(200, docs_payload), post_resp=_Resp(201, {"number": 11}), record=record, ) status, detail = merge_gate.ensure_open_pr(REPO, BRANCH) assert (status, detail) == ("created", "11") assert any(r[0] == "POST" for r in record) # --------------------------------------------------------------------------- # TC-04 (AC-7): Gitea GET/POST raise -> ("failed", reason), never raises. # --------------------------------------------------------------------------- def test_tc04_never_raise_on_get_error(monkeypatch): import httpx def boom(*a, **k): raise httpx.ConnectError("gitea down") monkeypatch.setattr(httpx, "get", boom) monkeypatch.setattr(httpx, "post", boom) status, detail = merge_gate.ensure_open_pr(REPO, BRANCH) assert status == "failed" assert detail # carries a reason def test_tc04_never_raise_on_post_error(monkeypatch): import httpx def boom_post(*a, **k): raise httpx.ConnectError("post exploded") _install_httpx(monkeypatch, get_resp=_Resp(200, []), post_resp=None) monkeypatch.setattr(httpx, "post", boom_post) status, detail = merge_gate.ensure_open_pr(REPO, BRANCH) assert status == "failed" def test_tc04_failed_when_post_non_2xx(monkeypatch): # A plain non-2xx, non-conflict POST -> failed (not silently swallowed). _install_httpx( monkeypatch, get_resp=_Resp(200, []), post_resp=_Resp(500, text="boom") ) status, detail = merge_gate.ensure_open_pr(REPO, BRANCH) assert status == "failed" assert "500" in detail # --------------------------------------------------------------------------- # TC-05 (AC-2 / FR-5): race -> POST returns 409/422 "PR exists" -> re-GET confirms # the existing PR -> ("existed", N), no duplicate. # --------------------------------------------------------------------------- @pytest.mark.parametrize("conflict_code", [409, 422]) def test_tc05_race_post_conflict_confirms_existing(monkeypatch, conflict_code): # First GET: no PR (so we attempt POST). POST: conflict. Re-GET: PR now present. gets = iter([ _Resp(200, []), # first probe: absent _Resp(200, [{"number": 99, "head": {"ref": BRANCH}, "base": {"ref": "main"}}]), ]) _install_httpx( monkeypatch, get_resp=lambda: next(gets), post_resp=_Resp(conflict_code, text="pull request already exists"), ) status, detail = merge_gate.ensure_open_pr(REPO, BRANCH) assert (status, detail) == ("existed", "99")