"""ORCH-036 TC-04/05/06: the manual-approve gate for the executable self-deploy. Contract (AC-5, AC-12): * TC-04 — ``deploy_require_manual_approve`` defaults to True in settings. * TC-05 — flag true + NO human approve -> the prod hook is NEVER called; the deploy-staging -> deploy edge only advances the STAGE and requests an approve (Phase A). ``initiate_deploy`` / ssh subprocess must not be touched. * TC-06 — flag true + a human Approved -> the prod hook is launched EXACTLY once (Phase B), idempotent on a repeated Approved (the ``initiated`` marker guards). """ import os import tempfile import pytest _test_db = os.path.join(tempfile.gettempdir(), "test_orch_deploy_approve.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.stage_engine import advance_stage # noqa: E402 @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() # Isolate the sentinel state dirs to a per-test tmp dir. monkeypatch.setattr(self_deploy.settings, "repos_dir", str(tmp_path)) monkeypatch.setattr(self_deploy.settings, "host_repos_dir", str(tmp_path)) 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", ): monkeypatch.setattr(stage_engine, name, MagicMock()) def _make_task(stage, repo="orchestrator", branch="feature/ORCH-036-x", wi="ORCH-036"): 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), ) task_id = cur.lastrowid conn.commit() conn.close() return task_id def _stage(task_id): conn = get_db() row = conn.execute("SELECT stage FROM tasks WHERE id=?", (task_id,)).fetchone() conn.close() return row[0] def _jobs(): conn = get_db() rows = conn.execute("SELECT agent, repo, task_id FROM jobs ORDER BY id").fetchall() conn.close() return [dict(r) for r in rows] def _pass(*a, **k): return (True, "ok") # --------------------------------------------------------------------------- # TC-04: default flag value # --------------------------------------------------------------------------- def test_tc04_manual_approve_default_true(): """The fresh, un-overridden settings default must be True (safe-by-default).""" from src.config import Settings assert Settings().deploy_require_manual_approve is True # --------------------------------------------------------------------------- # TC-05: flag true, no approve -> prod hook NOT called (Phase A only) # --------------------------------------------------------------------------- def test_tc05_no_approve_does_not_call_prod_hook(monkeypatch): monkeypatch.setattr(stage_engine.settings, "deploy_require_manual_approve", True) monkeypatch.setattr( stage_engine, "QG_CHECKS", {**stage_engine.QG_CHECKS, "check_staging_status": _pass, "check_branch_mergeable": _pass, "check_staging_image_fresh": _pass}, ) # Spy: the deploy launcher must never run on the staging->deploy edge. initiate = MagicMock() monkeypatch.setattr(stage_engine.self_deploy, "initiate_deploy", initiate) ssh_run = MagicMock() monkeypatch.setattr(self_deploy.subprocess, "run", ssh_run) task_id = _make_task("deploy-staging") res = advance_stage( task_id, "deploy-staging", "orchestrator", "ORCH-036", "feature/ORCH-036-x", finished_agent="deployer", ) # Phase A: advanced the STAGE to deploy, but requested approve — no prod hook. assert res.advanced is True assert res.to_stage == "deploy" assert _stage(task_id) == "deploy" assert res.note == "self-deploy-approval-pending" initiate.assert_not_called() ssh_run.assert_not_called() # No deployer job: the human Approved (Phase B) is what triggers the deploy. assert _jobs() == [] # The restart-safe approve-requested marker was written. assert self_deploy.has_marker("orchestrator", "ORCH-036", self_deploy.APPROVE_REQUESTED) # --------------------------------------------------------------------------- # TC-06: flag true + Approved -> prod hook called exactly once (idempotent) # --------------------------------------------------------------------------- def test_tc06_approved_calls_prod_hook_exactly_once(monkeypatch): monkeypatch.setattr(stage_engine.settings, "deploy_require_manual_approve", True) monkeypatch.setattr(stage_engine.settings, "deploy_ssh_host", "mva154") # Real initiate_deploy, but the ssh subprocess is mocked (rc=0 -> dispatched). ssh_run = MagicMock(return_value=MagicMock(returncode=0, stdout="", stderr="")) monkeypatch.setattr(self_deploy.subprocess, "run", ssh_run) task_id = _make_task("deploy") # already on deploy, awaiting Approved # 1st human Approved -> Phase B initiates the detached deploy. res1 = advance_stage( task_id, "deploy", "orchestrator", "ORCH-036", "feature/ORCH-036-x", finished_agent=None, ) assert res1.note == "self-deploy-initiated" assert ssh_run.call_count == 1 # The finalizer was enqueued. assert any(j["agent"] == "deploy-finalizer" for j in _jobs()) assert self_deploy.has_marker("orchestrator", "ORCH-036", self_deploy.INITIATED) # 2nd (duplicate) Approved -> idempotent no-op, hook NOT called again. res2 = advance_stage( task_id, "deploy", "orchestrator", "ORCH-036", "feature/ORCH-036-x", finished_agent=None, ) assert res2.note == "self-deploy-already-initiated" assert ssh_run.call_count == 1 # still exactly one prod deploy