developer(ET): auto-commit from developer run_id=192

tests/test_deploy_approve.py

@@ -0,0 +1,160 @@
+"""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},
+    )
+    # 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