orchestrator/tests/test_auto_labels_integration.py

"""ORCH-089 — integration: end-to-end auto-pass across pipeline edges.

Covers (04-test-plan.yaml):
  TC-23  both labels + all tech-gates green -> analysis -> deploy with no manual
         clicks (AC-3).
  TC-24  autoDeploy + a RED edge sub-gate -> Phase A not reached, Phase B not
         initiated (AC-5).
  TC-25  regression: no labels -> both gates manual exactly as before ORCH-089
         (AC-4).
"""
import os
import tempfile

import pytest

_test_db = os.path.join(tempfile.gettempdir(), "test_auto_integ.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.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, "_build_analyst_ready_comment",
                        lambda *a, **k: "ready", 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",
                 "set_issue_blocked", "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 _gates(monkeypatch, **overrides):
    base = {
        "check_analysis_complete": _pass,
        "check_staging_status": _pass,
        "check_branch_mergeable": _pass,
        "check_security_gate": _pass,
        "check_staging_image_fresh": _pass,
    }
    base.update(overrides)
    monkeypatch.setattr(stage_engine, "QG_CHECKS", {**stage_engine.QG_CHECKS, **base})


def _make_task(stage, wi="ORCH-089", repo="orchestrator"):
    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, "feature/ORCH-089-x", 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]


# --- TC-23: both labels, all green -> autonomous ---------------------------
def test_tc23_both_labels_autonomous(monkeypatch):
    _gates(monkeypatch)
    monkeypatch.setattr(labels, "has_label", lambda w, name, p=None: True)

    # BRD edge: analyst finished -> auto-approve -> architecture.
    brd = _make_task("analysis")
    res_brd = advance_stage(brd, "analysis", "orchestrator", "ORCH-089",
                            "feature/ORCH-089-x", finished_agent="analyst")
    assert res_brd.note == "auto-approved-via-label"
    assert _stage_of(brd) == "architecture"

    # Deploy edge: staging-deployer finished -> Phase A advances to deploy -> auto
    # Phase B initiates the prod deploy. No human Approved nor Confirm Deploy.
    dep = _make_task("deploy-staging", wi="ORCH-089b")
    res_dep = advance_stage(dep, "deploy-staging", "orchestrator", "ORCH-089b",
                            "feature/ORCH-089-x", finished_agent="deployer")
    assert res_dep.note == "self-deploy-initiated"
    assert stage_engine.self_deploy.initiate_deploy.called
    assert _stage_of(dep) == "deploy"


# --- TC-24: red sub-gate blocks autoDeploy ---------------------------------
def test_tc24_red_staging_blocks_auto_deploy(monkeypatch):
    # staging RED -> the edge fails BEFORE Phase A -> autoDeploy cannot fire.
    _gates(monkeypatch, check_staging_status=lambda *a, **k: (False, "FAILED"))
    monkeypatch.setattr(labels, "has_label", lambda w, name, p=None: True)

    dep = _make_task("deploy-staging")
    res = advance_stage(dep, "deploy-staging", "orchestrator", "ORCH-089",
                        "feature/ORCH-089-x", finished_agent="deployer")
    # Phase B never initiated despite the autoDeploy label.
    assert not stage_engine.self_deploy.initiate_deploy.called
    assert res.note != "self-deploy-initiated"
    assert _stage_of(dep) != "deploy"  # did not advance onto deploy


# --- TC-25: regression — no labels -> manual gates -------------------------
def test_tc25_no_labels_manual(monkeypatch):
    _gates(monkeypatch)
    monkeypatch.setattr(labels, "has_label", lambda w, name, p=None: False)

    brd = _make_task("analysis")
    res_brd = advance_stage(brd, "analysis", "orchestrator", "ORCH-089",
                            "feature/ORCH-089-x", finished_agent="analyst")
    assert res_brd.note == "analysis-in-review"
    assert _stage_of(brd) == "analysis"

    dep = _make_task("deploy-staging", wi="ORCH-089b")
    res_dep = advance_stage(dep, "deploy-staging", "orchestrator", "ORCH-089b",
                            "feature/ORCH-089-x", finished_agent="deployer")
    assert res_dep.note == "self-deploy-approval-pending"
    assert not stage_engine.self_deploy.initiate_deploy.called