orchestrator/tests/test_bug_fast_track_routing.py

"""ORCH-019 — advance_stage routing-override (ADR-001 D3).

Covers (04-test-plan.yaml):
  TC-05  bug task: analysis -> development (architecture skipped, developer
         enqueued); non-bug task: analysis -> architecture (architect enqueued).
  TC-06  STAGE_TRANSITIONS is structurally unchanged (set of stages + edges +
         agents + qg byte-for-byte) — the override does NOT mutate the table.
"""
import os
import tempfile

import pytest

_test_db = os.path.join(tempfile.gettempdir(), "test_bug_fast_track_routing.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")

import src.db as db  # noqa: E402
from src.db import init_db, get_db, set_task_track  # noqa: E402
from src import stage_engine  # noqa: E402
from src import config as cfg  # noqa: E402
from src.stage_engine import advance_stage  # noqa: E402


@pytest.fixture(autouse=True)
def fresh_db(monkeypatch, tmp_path):
    dbfile = tmp_path / "r.db"
    monkeypatch.setattr(db.settings, "db_path", str(dbfile))
    monkeypatch.setattr(cfg.settings, "bug_fast_track_enabled", True, raising=False)
    init_db()
    yield


@pytest.fixture(autouse=True)
def silence_side_effects(monkeypatch):
    for name in (
        "notify_stage_change", "notify_qg_failure", "notify_approve_requested",
        "send_telegram", "plane_notify_stage", "plane_notify_qg", "plane_add_comment",
        "set_issue_in_review", "set_issue_needs_input", "set_issue_in_progress",
        "set_issue_blocked", "set_issue_done", "set_issue_analysis",
        "set_issue_awaiting_deploy", "set_issue_deploying", "set_issue_monitoring",
        "set_issue_approved",
    ):
        monkeypatch.setattr(stage_engine, name, lambda *a, **k: None, raising=False)
    yield


def _make_task(work_item_id, stage="analysis", repo="orchestrator"):
    conn = get_db()
    cur = conn.execute(
        "INSERT INTO tasks (plane_id, work_item_id, repo, branch, stage, title) "
        "VALUES (?, ?, ?, ?, ?, ?)",
        (work_item_id, work_item_id, repo, f"feature/{work_item_id}", stage, work_item_id),
    )
    tid = cur.lastrowid
    conn.commit()
    conn.close()
    return tid


# --- TC-05 -----------------------------------------------------------------
def test_tc05_bug_task_skips_architecture():
    tid = _make_task("ORCH-bug", stage="analysis")
    set_task_track(tid, "bug")
    # agent=None -> the webhook Approved-via-status path (gate satisfied, advance).
    res = advance_stage(
        tid, "analysis", "orchestrator", "ORCH-bug", "feature/ORCH-bug",
        finished_agent=None,
    )
    assert res.advanced is True
    assert res.to_stage == "development"
    assert res.enqueued_agent == "developer"
    # DB stage actually advanced past architecture.
    row = db.get_task_by_work_item_id("ORCH-bug")
    assert row["stage"] == "development"


def test_tc05_full_task_keeps_architecture():
    tid = _make_task("ORCH-full", stage="analysis")
    # track defaults to 'full' (no set_task_track call).
    res = advance_stage(
        tid, "analysis", "orchestrator", "ORCH-full", "feature/ORCH-full",
        finished_agent=None,
    )
    assert res.advanced is True
    assert res.to_stage == "architecture"
    assert res.enqueued_agent == "architect"


def test_tc05_killswitch_off_bug_keeps_architecture(monkeypatch):
    monkeypatch.setattr(cfg.settings, "bug_fast_track_enabled", False, raising=False)
    tid = _make_task("ORCH-bugoff", stage="analysis")
    set_task_track(tid, "bug")  # stored, but the flag is off -> inert
    res = advance_stage(
        tid, "analysis", "orchestrator", "ORCH-bugoff", "feature/ORCH-bugoff",
        finished_agent=None,
    )
    assert res.to_stage == "architecture"
    assert res.enqueued_agent == "architect"


def test_tc05_bug_only_affects_analysis_edge():
    """The override is scoped to the analysis-exit edge only — a bug task on
    `development` still routes development -> review (no spurious skips)."""
    tid = _make_task("ORCH-bugdev", stage="development")
    set_task_track(tid, "bug")
    # Make check_ci_green pass deterministically (we only assert routing, not CI).
    import src.stage_engine as se
    orig = se.QG_CHECKS.get("check_ci_green")
    se.QG_CHECKS["check_ci_green"] = lambda *a, **k: (True, "ok")
    try:
        res = advance_stage(
            tid, "development", "orchestrator", "ORCH-bugdev", "feature/ORCH-bugdev",
            finished_agent=None,
        )
    finally:
        if orig is not None:
            se.QG_CHECKS["check_ci_green"] = orig
    assert res.to_stage == "review"


# --- TC-06: STAGE_TRANSITIONS structurally unchanged -----------------------
def test_tc06_stage_transitions_unchanged():
    from src.stages import STAGE_TRANSITIONS
    expected = {
        "created": {"next": "analysis", "agent": "analyst", "qg": None},
        "analysis": {"next": "architecture", "agent": "architect", "qg": "check_analysis_approved"},
        "architecture": {"next": "development", "agent": "developer", "qg": "check_architecture_done"},
        "development": {"next": "review", "agent": "reviewer", "qg": "check_ci_green"},
        "review": {"next": "testing", "agent": "tester", "qg": "check_reviewer_verdict"},
        "testing": {"next": "deploy-staging", "agent": "deployer", "qg": "check_tests_passed"},
        "deploy-staging": {"next": "deploy", "agent": "deployer", "qg": "check_staging_status"},
        "deploy": {"next": "done", "agent": None, "qg": "check_deploy_status"},
        "done": {"next": None, "agent": None, "qg": None},
        "cancelled": {"next": None, "agent": None, "qg": None},
    }
    assert STAGE_TRANSITIONS == expected


def test_tc06_get_next_stage_pure():
    """get_next_stage / get_agent_for_stage stay PURE (no track arg) — the override
    lives in advance_stage, not in stages.py."""
    from src.stages import get_next_stage, get_agent_for_stage
    assert get_next_stage("analysis") == "architecture"
    assert get_agent_for_stage("analysis") == "architect"