orchestrator/tests/test_bug_fast_track_escalation.py

"""ORCH-019 — escalation of a complex bug to the full cycle (FR-5 / AC-5, D5).

Covers (04-test-plan.yaml):
  TC-11  After the escalate endpoint resets track 'bug' -> 'full' (while the task
         is still in `analysis`), the next advance routes analysis -> architecture
         (return to the full cycle with the architect run).
"""
import os
import tempfile

import pytest

_test_db = os.path.join(tempfile.gettempdir(), "test_bug_fast_track_escalation.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  # noqa: E402
from src import stage_engine, 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 / "esc.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", track="bug"):
    conn = get_db()
    cur = conn.execute(
        "INSERT INTO tasks (plane_id, work_item_id, repo, branch, stage, title, track) "
        "VALUES (?, ?, ?, ?, ?, ?, ?)",
        (work_item_id, work_item_id, "orchestrator", f"feature/{work_item_id}",
         stage, work_item_id, track),
    )
    tid = cur.lastrowid
    conn.commit()
    conn.close()
    return tid


def test_tc11_escalate_returns_to_full_cycle(monkeypatch):
    import asyncio
    from src import main

    tid = _make_task("ORCH-cmplx", stage="analysis", track="bug")

    # Operator escalates while the task is still in analysis.
    out = asyncio.run(main.bug_fast_track_escalate(work_item="ORCH-cmplx"))
    assert out["ok"] is True
    assert out["track"] == "full"
    assert out["was"] == "bug"
    assert db.get_task_track(tid) == "full"

    # The next advance now routes back through architecture (full cycle).
    res = advance_stage(
        tid, "analysis", "orchestrator", "ORCH-cmplx", "feature/ORCH-cmplx",
        finished_agent=None,
    )
    assert res.to_stage == "architecture"
    assert res.enqueued_agent == "architect"


def test_tc11_escalate_unknown_work_item():
    import asyncio
    from src import main
    out = asyncio.run(main.bug_fast_track_escalate(work_item="ORCH-nope"))
    assert out["ok"] is False


def test_tc11_escalate_missing_arg():
    import asyncio
    from src import main
    out = asyncio.run(main.bug_fast_track_escalate(work_item=""))
    assert out["ok"] is False


def test_tc11_escalate_idempotent_on_full(monkeypatch):
    import asyncio
    from src import main
    tid = _make_task("ORCH-already", stage="analysis", track="full")
    out = asyncio.run(main.bug_fast_track_escalate(work_item="ORCH-already"))
    assert out["ok"] is True
    assert out["was"] == "full"
    assert db.get_task_track(tid) == "full"