orchestrator/tests/test_orch026_conditionality.py

"""ORCH-026 conditionality / self-hosting safety (TC-A06, TC-A07).

TC-A06  kill-switch / out-of-scope: with the flag off (or for a repo outside the
        merge-gate scope) the merge path behaves 1:1 as before ORCH-026 — no-op.
TC-A07  self-hosting safety: the new Level-A logic never pushes to main; the only
        force op stays --force-with-lease on the task branch; STAGE_TRANSITIONS
        and the QG_CHECKS registry are unchanged.
"""
import os
import tempfile

os.environ.setdefault("ORCH_DB_PATH", os.path.join(tempfile.gettempdir(), "test_orch026_cond.db"))
os.environ.setdefault("ORCH_GITEA_TOKEN", "test-token")
os.environ.setdefault("ORCH_PLANE_API_TOKEN", "test-token")

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


# ----------------------------------------------------------------- TC-A06
def test_out_of_scope_repo_is_noop_even_with_flag_on(monkeypatch):
    """A repo outside merge_gate scope -> N/A pass, regardless of premerge flag."""
    monkeypatch.setattr(checks.settings, "merge_gate_enabled", True, raising=False)
    monkeypatch.setattr(checks.settings, "merge_gate_repos", "orchestrator", raising=False)
    monkeypatch.setattr(checks.settings, "premerge_rebase_always", True, raising=False)

    # enduro-trails is NOT in the scope -> no lease, no rebase, just N/A.
    called = {"acquire": 0, "rebase": 0}
    monkeypatch.setattr(merge_gate, "acquire_merge_lease",
                        lambda *a, **k: (called.__setitem__("acquire", called["acquire"] + 1), (True, "x"))[1],
                        raising=False)
    monkeypatch.setattr(merge_gate, "auto_rebase_onto_main",
                        lambda *a, **k: (called.__setitem__("rebase", called["rebase"] + 1), (True, "x"))[1],
                        raising=False)
    ok, reason = checks.check_branch_mergeable("enduro-trails", "ET-1", "feature/e")
    assert ok is True
    assert "N/A" in reason
    assert called["acquire"] == 0 and called["rebase"] == 0


def test_task_deps_kill_switch_omits_gate(monkeypatch):
    """task_deps_enabled=False -> claim_next_job query is the ORCH-1 query (no gate)."""
    import src.db as db
    monkeypatch.setattr(db.settings, "task_deps_enabled", False, raising=False)
    # Inspect the SQL the claim builds by stubbing the connection.
    captured = {}

    class _FakeConn:
        def execute(self, sql, *a):
            captured.setdefault("sql", sql)

            class _R:
                def fetchone(self_inner):
                    return None
            return _R()

        def commit(self):
            pass

        def close(self):
            pass

    monkeypatch.setattr(db, "get_db", lambda: _FakeConn())
    db.claim_next_job()
    assert "NOT EXISTS" not in captured["sql"], "gate must be omitted when disabled"


def test_task_deps_enabled_adds_gate(monkeypatch):
    import src.db as db
    monkeypatch.setattr(db.settings, "task_deps_enabled", True, raising=False)
    captured = {}

    class _FakeConn:
        def execute(self, sql, *a):
            captured.setdefault("sql", sql)

            class _R:
                def fetchone(self_inner):
                    return None
            return _R()

        def commit(self):
            pass

        def close(self):
            pass

    monkeypatch.setattr(db, "get_db", lambda: _FakeConn())
    db.claim_next_job()
    assert "NOT EXISTS" in captured["sql"], "gate must be present when enabled"
    assert "job_deps" in captured["sql"]


# ----------------------------------------------------------------- TC-A07
def test_stage_transitions_unchanged():
    """ORCH-026 must not touch the state machine (AC-A5)."""
    from src.stages import STAGE_TRANSITIONS
    # The canonical happy-path edges must still exist exactly.
    assert STAGE_TRANSITIONS["deploy-staging"]["next"] == "deploy"
    assert STAGE_TRANSITIONS["deploy"]["next"] == "done"
    assert STAGE_TRANSITIONS["development"]["next"] == "review"


def test_qg_registry_has_no_new_dep_gate():
    """The dependency gate is врезка in claim_next_job, NOT a registered QG."""
    from src.qg.checks import QG_CHECKS
    joined = " ".join(QG_CHECKS.keys())
    assert "task_dep" not in joined and "dependency" not in joined


def test_premerge_only_force_with_lease_on_branch():
    """auto_rebase_onto_main never pushes to main; force is --force-with-lease only."""
    import inspect
    src = inspect.getsource(merge_gate.auto_rebase_onto_main)
    assert "--force-with-lease" in src
    # No raw 'push origin main' / force-push to main in the rebase path.
    assert "push origin main" not in src
    assert "--force " not in src  # plain --force (not -with-lease) is forbidden