orchestrator/tests/test_orch082_ensure_pr.py

"""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")