orchestrator/tests/test_auto_deploy.py

"""ORCH-089 — autoDeploy врезка in _handle_self_deploy_phase_a.

Covers (04-test-plan.yaml):
  TC-15  autoDeploy + Phase A advance on `deploy` -> Phase B (initiate_deploy)
         is auto-invoked.
  TC-16  no autoDeploy label -> prior Phase A: Awaiting Deploy, wait for Confirm
         Deploy.
  TC-17  idempotent: INITIATED marker already present -> repeat auto-trigger no-op.
  TC-18  non-self repo / out of scope -> no auto (Phase A/B only for self-hosting).
  TC-19  autoDeploy logged + Telegram + Plane comment (transparency AC-7).
"""
import os
import tempfile

import pytest

_test_db = os.path.join(tempfile.gettempdir(), "test_auto_deploy.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_db(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)
    # Pass all edge sub-gates so the deploy-staging -> deploy edge reaches Phase A.
    monkeypatch.setattr(
        stage_engine, "QG_CHECKS",
        {**stage_engine.QG_CHECKS,
         "check_staging_status": _pass,
         "check_branch_mergeable": _pass,
         "check_security_gate": _pass,
         "check_staging_image_fresh": _pass},
    )
    # Default auto-mode flags ON (overridden per-test).
    monkeypatch.setattr(stage_engine.settings, "auto_label_enabled", True, raising=False)
    monkeypatch.setattr(stage_engine.settings, "auto_deploy_label", "autoDeploy", raising=False)
    yield


@pytest.fixture(autouse=True)
def silence(monkeypatch):
    for name in ("notify_stage_change", "notify_qg_failure", "plane_notify_stage",
                 "plane_notify_qg", "set_issue_in_review", "set_issue_awaiting_deploy",
                 "set_issue_deploying"):
        monkeypatch.setattr(stage_engine, name, MagicMock(), raising=False)
    monkeypatch.setattr(stage_engine, "plane_add_comment", MagicMock())
    monkeypatch.setattr(stage_engine, "send_telegram", MagicMock())


def _make_task(stage="deploy-staging", 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 _label(monkeypatch, present=True, applies=True):
    monkeypatch.setattr(labels, "auto_deploy_applies", lambda repo: applies)
    monkeypatch.setattr(labels, "has_label", lambda w, name, p=None: present)


def _advance(tid, repo="orchestrator", wi="ORCH-089"):
    return advance_stage(tid, "deploy-staging", repo, wi,
                         "feature/ORCH-089-x", finished_agent="deployer")


# --- TC-15 -----------------------------------------------------------------
def test_tc15_auto_deploy_initiates_phase_b(monkeypatch):
    _label(monkeypatch, present=True)
    initiate = MagicMock(return_value=(True, "ok"))
    monkeypatch.setattr(stage_engine.self_deploy, "initiate_deploy", initiate)
    tid = _make_task()
    res = _advance(tid)
    # Phase B ran via the same path a human Confirm Deploy takes.
    initiate.assert_called_once()
    assert res.note == "self-deploy-initiated"
    assert self_deploy.has_marker("orchestrator", "ORCH-089", self_deploy.INITIATED)
    # APPROVE_REQUESTED (the human ask) was SKIPPED on the auto path.
    assert not self_deploy.has_marker(
        "orchestrator", "ORCH-089", self_deploy.APPROVE_REQUESTED
    )


# --- TC-16 -----------------------------------------------------------------
def test_tc16_no_label_waits_for_human(monkeypatch):
    _label(monkeypatch, present=False)
    initiate = MagicMock(return_value=(True, "ok"))
    monkeypatch.setattr(stage_engine.self_deploy, "initiate_deploy", initiate)
    tid = _make_task()
    res = _advance(tid)
    # Prior Phase A behaviour: approval-pending, no deploy initiated.
    assert res.note == "self-deploy-approval-pending"
    initiate.assert_not_called()
    assert self_deploy.has_marker(
        "orchestrator", "ORCH-089", self_deploy.APPROVE_REQUESTED
    )
    stage_engine.set_issue_awaiting_deploy.assert_called_once()


# --- TC-17: idempotency ----------------------------------------------------
def test_tc17_idempotent_initiated_marker(monkeypatch):
    """autoDeploy delegates prod-deploy to _handle_self_deploy_phase_b, whose
    INITIATED marker makes a repeat a no-op. Phase A always clears stale state
    first (ADR D4), so the guard that protects against a double prod deploy is the
    INITIATED marker WRITTEN by Phase B — verify the auto path initiates exactly
    once and a subsequent Phase B re-entry (duplicate confirm / reaper re-drive) is
    a no-op."""
    _label(monkeypatch, present=True)
    initiate = MagicMock(return_value=(True, "ok"))
    monkeypatch.setattr(stage_engine.self_deploy, "initiate_deploy", initiate)
    tid = _make_task()
    res = _advance(tid)
    assert res.note == "self-deploy-initiated"
    assert initiate.call_count == 1
    assert self_deploy.has_marker("orchestrator", "ORCH-089", self_deploy.INITIATED)
    # A repeat Phase B (e.g. duplicate Confirm Deploy webhook / reaper re-drive)
    # with INITIATED already set is a no-op — no second prod deploy.
    res2 = stage_engine._handle_self_deploy_phase_b(
        tid, "orchestrator", "ORCH-089", "feature/ORCH-089-x",
        stage_engine.AdvanceResult(from_stage="deploy"),
    )
    assert initiate.call_count == 1  # still exactly one


# --- TC-18: non-self / out of scope ----------------------------------------
def test_tc18_non_self_repo_no_phase_a(monkeypatch):
    # For a non-self repo Phase A is not reached at all (self_deploy_applies False),
    # so autoDeploy is a structural no-op. The edge advances normally to `deploy`.
    _label(monkeypatch, present=True, applies=False)
    initiate = MagicMock(return_value=(True, "ok"))
    monkeypatch.setattr(stage_engine.self_deploy, "initiate_deploy", initiate)
    tid = _make_task(repo="enduro-trails", wi="ET-1")
    res = advance_stage(tid, "deploy-staging", "enduro-trails", "ET-1",
                        "feature/ORCH-089-x", finished_agent="deployer")
    initiate.assert_not_called()
    # No Phase A / Phase B for non-self repo.
    assert res.note != "self-deploy-initiated"


# --- TC-19: transparency ---------------------------------------------------
def test_tc19_transparency_channels(monkeypatch, caplog):
    _label(monkeypatch, present=True)
    monkeypatch.setattr(stage_engine.self_deploy, "initiate_deploy",
                        MagicMock(return_value=(True, "ok")))
    tid = _make_task()
    import logging
    with caplog.at_level(logging.INFO, logger="orchestrator.stage_engine"):
        _advance(tid)
    assert any("auto-confirmed" in r.message.lower() or "autoDeploy" in r.message
               for r in caplog.records)
    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-deploy Plane comment"