"""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"