orchestrator/tests/test_auto_approve_brd.py

"""ORCH-089 — autoApprove врезка in _handle_analysis_approved_flow.

Covers (04-test-plan.yaml):
  TC-10  autoApprove + artifacts ready -> auto-advance analysis->architecture,
         Approved set, brd_review_ended clock closed.
  TC-11  no autoApprove label -> prior behaviour: In Review, return w/o advance.
  TC-12  autoApprove but artifacts missing (check_analysis_complete FAIL) -> NO
         advance (AC-5 for BRD).
  TC-13  autoApprove goes through the SAME advance path as a manual Approved (no
         duplicated transition logic; idempotent — stage lands on architecture).
  TC-14  autoApprove logged + Telegram + Plane comment (transparency AC-7).
"""
import os
import tempfile

import pytest

_test_db = os.path.join(tempfile.gettempdir(), "test_auto_approve_brd.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 labels  # noqa: E402
from src.stage_engine import advance_stage  # noqa: E402


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


def _files_fail(*a, **k):
    return (False, "missing artifacts")


@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()
    # Silence Plane/Telegram side effects; capture the transparency channels.
    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_approved", "notify_approve_requested"):
        monkeypatch.setattr(stage_engine, name, MagicMock(), raising=False)
    # Avoid worktree access in the analyst "ready" comment builder.
    monkeypatch.setattr(stage_engine, "_build_analyst_ready_comment",
                        lambda *a, **k: "ready", raising=False)
    monkeypatch.setattr(stage_engine, "plane_add_comment", MagicMock())
    monkeypatch.setattr(stage_engine, "send_telegram", MagicMock())
    yield


def _make_task(stage="analysis", 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) "
        "VALUES (?, ?, ?, ?, ?)",
        (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 _brd_ended(task_id):
    conn = get_db()
    row = conn.execute(
        "SELECT brd_review_ended_at FROM tasks WHERE id=?", (task_id,)
    ).fetchone()
    conn.close()
    return row[0]


def _patch_complete_gate(monkeypatch, ok=True):
    gate = _files_ok if ok else _files_fail
    monkeypatch.setattr(
        stage_engine, "QG_CHECKS",
        {**stage_engine.QG_CHECKS, "check_analysis_complete": gate},
    )


def _label(monkeypatch, present=True, applies=True):
    monkeypatch.setattr(labels, "auto_approve_applies", lambda repo: applies)
    monkeypatch.setattr(labels, "has_label", lambda w, name, p=None: present)


# --- TC-10 -----------------------------------------------------------------
def test_tc10_auto_approve_advances(monkeypatch):
    _patch_complete_gate(monkeypatch, ok=True)
    _label(monkeypatch, present=True)
    tid = _make_task()
    # The BRD-review clock was started when the task entered In Review; the
    # advance closes it (mark_brd_review_ended only stamps when a start exists).
    conn = get_db()
    conn.execute(
        "UPDATE tasks SET brd_review_started_at=datetime('now') WHERE id=?", (tid,)
    )
    conn.commit()
    conn.close()
    res = advance_stage(tid, "analysis", "orchestrator", "ORCH-089",
                        "feature/ORCH-089-x", finished_agent="analyst")
    assert res.note == "auto-approved-via-label"
    assert res.advanced is True
    assert _stage_of(tid) == "architecture"
    assert _brd_ended(tid) is not None  # clock closed by mark_brd_review_ended
    stage_engine.set_issue_approved.assert_called_once()  # Approved indication


# --- TC-11 -----------------------------------------------------------------
def test_tc11_no_label_waits_for_human(monkeypatch):
    _patch_complete_gate(monkeypatch, ok=True)
    _label(monkeypatch, present=False)
    tid = _make_task()
    res = advance_stage(tid, "analysis", "orchestrator", "ORCH-089",
                        "feature/ORCH-089-x", finished_agent="analyst")
    assert res.note == "analysis-in-review"
    assert res.advanced is False
    assert _stage_of(tid) == "analysis"  # still waiting for a human
    stage_engine.set_issue_in_review.assert_called_once()
    stage_engine.set_issue_approved.assert_not_called()


# --- TC-12 -----------------------------------------------------------------
def test_tc12_missing_artifacts_no_auto(monkeypatch):
    # autoApprove present, but artifacts incomplete -> files_ok False -> the
    # autoApprove block (inside `if files_ok`) is never reached.
    _patch_complete_gate(monkeypatch, ok=False)
    _label(monkeypatch, present=True)
    tid = _make_task()
    res = advance_stage(tid, "analysis", "orchestrator", "ORCH-089",
                        "feature/ORCH-089-x", finished_agent="analyst")
    assert res.advanced is False
    assert _stage_of(tid) == "analysis"
    assert res.note != "auto-approved-via-label"
    stage_engine.set_issue_approved.assert_not_called()


# --- TC-13: same advance path / idempotent ---------------------------------
def test_tc13_same_advance_path_idempotent(monkeypatch):
    _patch_complete_gate(monkeypatch, ok=True)
    _label(monkeypatch, present=True)
    tid = _make_task()
    res = advance_stage(tid, "analysis", "orchestrator", "ORCH-089",
                        "feature/ORCH-089-x", finished_agent="analyst")
    # The advance went through the unified path -> architect enqueued exactly once.
    assert res.enqueued_agent == "architect"
    conn = get_db()
    n = conn.execute(
        "SELECT COUNT(*) FROM jobs WHERE task_id=? AND agent='architect'", (tid,)
    ).fetchone()[0]
    conn.close()
    assert n == 1
    # A later real Approved (webhook path, finished_agent=None) sees architecture,
    # not analysis -> it cannot re-run the analysis advance (idempotent).
    assert _stage_of(tid) == "architecture"


# --- TC-14: transparency ---------------------------------------------------
def test_tc14_transparency_channels(monkeypatch, caplog):
    _patch_complete_gate(monkeypatch, ok=True)
    _label(monkeypatch, present=True)
    tid = _make_task()
    import logging
    with caplog.at_level(logging.INFO, logger="orchestrator.stage_engine"):
        advance_stage(tid, "analysis", "orchestrator", "ORCH-089",
                      "feature/ORCH-089-x", finished_agent="analyst")
    # (a) log mentions the label + auto-approve.
    assert any("auto-approved" in r.message.lower() or "autoApprove" in r.message
               for r in caplog.records)
    # (b) Telegram fired; (c) a Plane comment authored by analyst about the auto-pass.
    assert stage_engine.send_telegram.called
    comment_calls = [c for c in stage_engine.plane_add_comment.call_args_list
                     if "авто-подтверждён" in c.args[1]]
    assert comment_calls, "expected an auto-approve Plane comment"