orchestrator/tests/test_auto_label_combinations.py

"""ORCH-089 — label independence + kill-switch (AC-8/AC-9).

Covers (04-test-plan.yaml):
  TC-20  only autoApprove: BRD auto, deploy waits for a human (AC-9).
  TC-21  only autoDeploy: BRD waits for a human, deploy auto (AC-9).
  TC-22  auto_label_enabled=False: both gates manual even with both labels (AC-8).
"""
import os
import tempfile

import pytest

_test_db = os.path.join(tempfile.gettempdir(), "test_auto_combos.db")
os.environ["ORCH_DB_PATH"] = _test_db
os.environ["ORCH_REPOS_DIR"] = tempfile.gettempdir()
os.environ.setdefault("ORCH_GITEA_TOKEN", "test-token")
os.environ.setdefault("ORCH_PLANE_API_TOKEN", "test-token")

from unittest.mock import MagicMock  # noqa: E402

import src.db as _db  # noqa: E402
from src.db import init_db, get_db  # noqa: E402
from src import stage_engine  # noqa: E402
from src import self_deploy  # noqa: E402
from src import labels  # noqa: E402
from src.stage_engine import advance_stage  # noqa: E402


def _pass(*a, **k):
    return (True, "ok")


@pytest.fixture(autouse=True)
def fresh(monkeypatch, tmp_path):
    monkeypatch.setattr(_db.settings, "db_path", _test_db)
    if os.path.exists(_test_db):
        os.unlink(_test_db)
    init_db()
    monkeypatch.setattr(self_deploy.settings, "repos_dir", str(tmp_path))
    monkeypatch.setattr(self_deploy.settings, "host_repos_dir", str(tmp_path))
    monkeypatch.setattr(stage_engine.settings, "deploy_require_manual_approve", True)
    monkeypatch.setattr(
        stage_engine, "QG_CHECKS",
        {**stage_engine.QG_CHECKS,
         "check_analysis_complete": _pass,
         "check_staging_status": _pass,
         "check_branch_mergeable": _pass,
         "check_security_gate": _pass,
         "check_staging_image_fresh": _pass},
    )
    monkeypatch.setattr(stage_engine, "_build_analyst_ready_comment",
                        lambda *a, **k: "ready", raising=False)
    # Real auto-mode scope/flags (kill-switch exercised per-test).
    monkeypatch.setattr(stage_engine.settings, "auto_label_enabled", True, raising=False)
    monkeypatch.setattr(stage_engine.settings, "auto_label_repos", "", raising=False)
    monkeypatch.setattr(stage_engine.settings, "auto_approve_label", "autoApprove", raising=False)
    monkeypatch.setattr(stage_engine.settings, "auto_deploy_label", "autoDeploy", raising=False)
    monkeypatch.setattr(stage_engine.self_deploy, "initiate_deploy",
                        MagicMock(return_value=(True, "ok")))
    for name in ("notify_stage_change", "notify_qg_failure", "plane_notify_stage",
                 "plane_notify_qg", "set_issue_in_review", "set_issue_needs_input",
                 "set_issue_awaiting_deploy", "set_issue_deploying", "set_issue_approved",
                 "notify_approve_requested"):
        monkeypatch.setattr(stage_engine, name, MagicMock(), raising=False)
    monkeypatch.setattr(stage_engine, "plane_add_comment", MagicMock())
    monkeypatch.setattr(stage_engine, "send_telegram", MagicMock())
    yield


def _make_task(stage, repo="orchestrator", branch="feature/ORCH-089-x", wi="ORCH-089"):
    conn = get_db()
    cur = conn.execute(
        "INSERT INTO tasks (plane_id, work_item_id, repo, branch, stage, brd_review_started_at) "
        "VALUES (?, ?, ?, ?, ?, datetime('now'))",
        (f"plane-{wi}", wi, repo, branch, stage),
    )
    tid = cur.lastrowid
    conn.commit()
    conn.close()
    return tid


def _stage_of(task_id):
    conn = get_db()
    row = conn.execute("SELECT stage FROM tasks WHERE id=?", (task_id,)).fetchone()
    conn.close()
    return row[0]


def _present_labels(monkeypatch, names):
    """has_label True only for the given normalized label names (real applies())."""
    want = {n.casefold() for n in names}
    monkeypatch.setattr(labels, "has_label",
                        lambda w, name, p=None: name.casefold() in want)


def _run_brd(wi="ORCH-089"):
    tid = _make_task("analysis", wi=wi)
    res = advance_stage(tid, "analysis", "orchestrator", wi,
                        "feature/ORCH-089-x", finished_agent="analyst")
    return tid, res


def _run_deploy(wi="ORCH-089"):
    tid = _make_task("deploy-staging", wi=wi)
    res = advance_stage(tid, "deploy-staging", "orchestrator", wi,
                        "feature/ORCH-089-x", finished_agent="deployer")
    return tid, res


# --- TC-20: only autoApprove -----------------------------------------------
def test_tc20_only_auto_approve(monkeypatch):
    _present_labels(monkeypatch, ["autoApprove"])
    tid_brd, res_brd = _run_brd()
    assert res_brd.note == "auto-approved-via-label"
    assert _stage_of(tid_brd) == "architecture"
    # Deploy gate still manual (autoDeploy absent).
    _tid_dep, res_dep = _run_deploy()
    assert res_dep.note == "self-deploy-approval-pending"


# --- TC-21: only autoDeploy ------------------------------------------------
def test_tc21_only_auto_deploy(monkeypatch):
    _present_labels(monkeypatch, ["autoDeploy"])
    tid_brd, res_brd = _run_brd()
    # BRD gate still manual (autoApprove absent).
    assert res_brd.note == "analysis-in-review"
    assert _stage_of(tid_brd) == "analysis"
    # Deploy auto-confirmed.
    _tid_dep, res_dep = _run_deploy()
    assert res_dep.note == "self-deploy-initiated"


# --- TC-22: kill-switch -> both manual -------------------------------------
def test_tc22_killswitch_both_manual(monkeypatch):
    monkeypatch.setattr(stage_engine.settings, "auto_label_enabled", False, raising=False)
    # Both labels "present", but the kill-switch makes applies() False FIRST, so
    # has_label is never consulted -> both gates stay manual.
    spy = MagicMock(return_value=True)
    monkeypatch.setattr(labels, "has_label", spy)
    tid_brd, res_brd = _run_brd()
    assert res_brd.note == "analysis-in-review"
    assert _stage_of(tid_brd) == "analysis"
    _tid_dep, res_dep = _run_deploy()
    assert res_dep.note == "self-deploy-approval-pending"
    spy.assert_not_called()  # zero network — AC-8