orchestrator/tests/test_qg_merge_gate.py

"""ORCH-043 / TC-12..17: the merge-gate quality check ``check_branch_mergeable``.

These exercise the COMPOSITION logic in src/qg/checks.check_branch_mergeable —
the deterministic gate the engine runs on the deploy-staging -> deploy edge. The
merge_gate primitives (rebase / re-test / lease) are mocked here; their real-git
behaviour is covered in tests/test_merge_gate.py.

Contract under test (ADR-001 §4):
  * conditionality: merge_gate_enabled=False / repo-out-of-scope -> no-op pass,
    NEVER touching the lease;
  * up-to-date branch -> pass, lease HELD;
  * behind + clean rebase + green re-test -> pass, lease HELD;
  * rebase conflict -> fail, lease RELEASED;
  * red / timeout re-test after rebase -> fail, lease RELEASED;
  * never-raise: an exception inside the gate -> (False, ...) with lease released.
"""

import os

os.environ.setdefault("ORCH_GITEA_TOKEN", "test-token")
os.environ.setdefault("ORCH_PLANE_API_TOKEN", "test-token")

import pytest  # noqa: E402

from src import merge_gate  # noqa: E402
from src.qg import checks as qg  # noqa: E402
from src.qg.checks import check_branch_mergeable  # noqa: E402

_REPO = "orchestrator"
_BRANCH = "feature/ORCH-043-x"
_WI = "ORCH-043"


@pytest.fixture
def lease_spy(monkeypatch):
    """Replace the merge_gate lease primitives with in-memory spies.

    Tracks acquire/release calls and lets each test program the acquire outcome
    so we can assert the gate's lease lifecycle without touching the filesystem.
    """
    state = {
        "acquired": False,
        "released": False,
        "acquire_result": (True, "lease acquired"),
    }

    def _acquire(repo, branch, work_item_id=None, task_id=None):
        ok, reason = state["acquire_result"]
        if ok:
            state["acquired"] = True
        return ok, reason

    def _release(repo, branch=None):
        state["released"] = True

    monkeypatch.setattr(merge_gate, "acquire_merge_lease", _acquire)
    monkeypatch.setattr(merge_gate, "release_merge_lease", _release)
    # Default merge_gate scope: real for the self-hosting orchestrator repo.
    monkeypatch.setattr(qg.settings, "merge_gate_enabled", True)
    monkeypatch.setattr(qg.settings, "merge_gate_repos", "")
    # ORCH-026: these ORCH-043 composition tests assert the ancestor-based
    # short-circuit ("branch up-to-date with main" -> no rebase). That is now the
    # `premerge_rebase_always=False` kill-switch path; pin it OFF here so they
    # keep testing the legacy ORCH-043 behaviour. The new always-rebase default
    # (True) is covered by tests/test_orch026_premerge_rebase.py (TC-A01).
    monkeypatch.setattr(qg.settings, "premerge_rebase_always", False, raising=False)
    return state


# ---------------------------------------------------------------------------
# Conditionality (no-op variants) — must NOT touch the lease.
# ---------------------------------------------------------------------------
def test_tc16_disabled_is_noop(monkeypatch, lease_spy):
    """TC-16 / AC-8: merge_gate_enabled=False -> pass, lease untouched."""
    monkeypatch.setattr(qg.settings, "merge_gate_enabled", False)
    ok, reason = check_branch_mergeable(_REPO, _WI, _BRANCH)
    assert ok is True
    assert reason == "merge-gate disabled"
    assert lease_spy["acquired"] is False
    assert lease_spy["released"] is False


def test_tc17_repo_out_of_scope_is_noop(monkeypatch, lease_spy):
    """TC-17 / AC-8: non-self-hosting repo (empty CSV) -> conditional no-op."""
    ok, reason = check_branch_mergeable("enduro-trails", "ET-1", "feature/ET-1-x")
    assert ok is True
    assert reason == "merge-gate N/A for enduro-trails"
    assert lease_spy["acquired"] is False
    assert lease_spy["released"] is False


def test_csv_scopes_gate_to_listed_repo(monkeypatch, lease_spy):
    """merge_gate_repos CSV makes the gate real for a non-self-hosting repo."""
    monkeypatch.setattr(qg.settings, "merge_gate_repos", "enduro-trails")
    monkeypatch.setattr(merge_gate, "branch_is_behind_main", lambda r, b: False)
    ok, reason = check_branch_mergeable("enduro-trails", "ET-1", "feature/ET-1-x")
    assert ok is True
    assert reason == "branch up-to-date with main"
    assert lease_spy["acquired"] is True  # gate actually ran


# ---------------------------------------------------------------------------
# Lock busy -> DEFER signal (no rollback at this layer).
# ---------------------------------------------------------------------------
def test_lock_busy_returns_defer_signal(monkeypatch, lease_spy):
    """Lease busy -> (False, 'merge-lock busy'); nothing acquired or released."""
    lease_spy["acquire_result"] = (False, "merge-lock busy")
    ok, reason = check_branch_mergeable(_REPO, _WI, _BRANCH)
    assert ok is False
    assert reason == "merge-lock busy"
    assert lease_spy["acquired"] is False
    assert lease_spy["released"] is False  # we never held it


# ---------------------------------------------------------------------------
# TC-12: branch already up-to-date -> pass, lease HELD.
# ---------------------------------------------------------------------------
def test_tc12_up_to_date_passes_lease_held(monkeypatch, lease_spy):
    monkeypatch.setattr(merge_gate, "branch_is_behind_main", lambda r, b: False)
    # If these were called the test would wrongly proceed — guard with raisers.
    monkeypatch.setattr(
        merge_gate, "auto_rebase_onto_main",
        lambda r, b: pytest.fail("must not rebase an up-to-date branch"),
    )
    ok, reason = check_branch_mergeable(_REPO, _WI, _BRANCH)
    assert ok is True
    assert reason == "branch up-to-date with main"
    assert lease_spy["acquired"] is True
    assert lease_spy["released"] is False  # lease HELD until the merge


# ---------------------------------------------------------------------------
# TC-13: behind + clean rebase + green re-test -> pass, lease HELD.
# ---------------------------------------------------------------------------
def test_tc13_behind_clean_rebase_green_passes_lease_held(monkeypatch, lease_spy):
    monkeypatch.setattr(merge_gate, "branch_is_behind_main", lambda r, b: True)
    monkeypatch.setattr(
        merge_gate, "auto_rebase_onto_main",
        lambda r, b: (True, "rebased onto origin/main"),
    )
    monkeypatch.setattr(merge_gate, "retest_branch", lambda r, b: (True, "re-test green"))
    ok, reason = check_branch_mergeable(_REPO, _WI, _BRANCH)
    assert ok is True
    assert reason == "rebased onto main, re-test green"
    assert lease_spy["acquired"] is True
    assert lease_spy["released"] is False  # lease HELD


# ---------------------------------------------------------------------------
# TC-14: rebase conflict -> fail, lease RELEASED.
# ---------------------------------------------------------------------------
def test_tc14_rebase_conflict_fails_lease_released(monkeypatch, lease_spy):
    monkeypatch.setattr(merge_gate, "branch_is_behind_main", lambda r, b: True)
    monkeypatch.setattr(
        merge_gate, "auto_rebase_onto_main",
        lambda r, b: (False, "rebase conflict: src/db.py"),
    )
    monkeypatch.setattr(
        merge_gate, "retest_branch",
        lambda r, b: pytest.fail("must not re-test after a failed rebase"),
    )
    ok, reason = check_branch_mergeable(_REPO, _WI, _BRANCH)
    assert ok is False
    assert reason == "rebase conflict: src/db.py"
    assert lease_spy["released"] is True


# ---------------------------------------------------------------------------
# TC-15: red / timeout re-test after rebase -> fail, lease RELEASED.
# ---------------------------------------------------------------------------
def test_tc15_red_retest_fails_lease_released(monkeypatch, lease_spy):
    monkeypatch.setattr(merge_gate, "branch_is_behind_main", lambda r, b: True)
    monkeypatch.setattr(
        merge_gate, "auto_rebase_onto_main",
        lambda r, b: (True, "rebased onto origin/main"),
    )
    monkeypatch.setattr(
        merge_gate, "retest_branch",
        lambda r, b: (False, "re-test failed: ...1 failed, 5 passed"),
    )
    ok, reason = check_branch_mergeable(_REPO, _WI, _BRANCH)
    assert ok is False
    assert reason.startswith("re-test failed after rebase:")
    assert "1 failed, 5 passed" in reason
    assert lease_spy["released"] is True


def test_tc15_retest_timeout_passes_reason_through(monkeypatch, lease_spy):
    """AC-6: a re-test timeout keeps its distinct reason and releases the lease."""
    monkeypatch.setattr(merge_gate, "branch_is_behind_main", lambda r, b: True)
    monkeypatch.setattr(
        merge_gate, "auto_rebase_onto_main",
        lambda r, b: (True, "rebased onto origin/main"),
    )
    monkeypatch.setattr(
        merge_gate, "retest_branch",
        lambda r, b: (False, "re-test timeout after 600s"),
    )
    ok, reason = check_branch_mergeable(_REPO, _WI, _BRANCH)
    assert ok is False
    assert reason == "re-test timeout after 600s"
    assert lease_spy["released"] is True


# ---------------------------------------------------------------------------
# Never-raise: an exception inside the gate -> (False, ...) + lease released.
# ---------------------------------------------------------------------------
def test_never_raise_releases_lease_on_internal_error(monkeypatch, lease_spy):
    """AC-9: a blowing-up primitive is caught; the gate returns and releases."""
    def _boom(r, b):
        raise RuntimeError("git exploded")

    monkeypatch.setattr(merge_gate, "branch_is_behind_main", _boom)
    ok, reason = check_branch_mergeable(_REPO, _WI, _BRANCH)
    assert ok is False
    assert "merge-gate error" in reason
    assert lease_spy["released"] is True  # held then released on the error path