"""ORCH-036 TC-07/08/09: self vs non-self deploy routing (AC-2, AC-11). * TC-07 — ``is_self_hosting_repo``/``self_deploy_applies`` recognise the orchestrator repo and reject any other (no regression). * TC-08 — for the self repo the restart is launched as a DETACHED host process (ssh + setsid + background), never synchronously inside the agent. * TC-09 — for a non-self repo (enduro-trails) the deploy keeps the legacy path: the self-deploy Phase A/B logic does NOT apply. """ import os import tempfile import pytest _test_db = os.path.join(tempfile.gettempdir(), "test_orch_deploy_routing.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.qg.checks import is_self_hosting_repo # 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() 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, branch, wi): 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-07: routing predicate # --------------------------------------------------------------------------- def test_tc07_is_self_hosting_repo_only_orchestrator(): assert is_self_hosting_repo("orchestrator") is True assert is_self_hosting_repo("ORCHESTRATOR") is True # case-insensitive assert is_self_hosting_repo("enduro-trails") is False assert is_self_hosting_repo("") is False assert is_self_hosting_repo(None) is False def test_tc07_self_deploy_applies_mirrors_routing(monkeypatch): monkeypatch.setattr(self_deploy.settings, "self_deploy_enabled", True) monkeypatch.setattr(self_deploy.settings, "self_deploy_repos", "") assert self_deploy.self_deploy_applies("orchestrator") is True assert self_deploy.self_deploy_applies("enduro-trails") is False # Global kill-switch wins. monkeypatch.setattr(self_deploy.settings, "self_deploy_enabled", False) assert self_deploy.self_deploy_applies("orchestrator") is False # --------------------------------------------------------------------------- # TC-08: self repo -> DETACHED host process (ssh + setsid + background) # --------------------------------------------------------------------------- def test_tc08_self_repo_launches_detached_host_process(monkeypatch): """The deploy command must be an ssh invocation that detaches the hook via setsid and backgrounds it (`&`), so it survives the prod container restart — i.e. NOT a synchronous in-agent call.""" monkeypatch.setattr(self_deploy.settings, "deploy_ssh_user", "slin") monkeypatch.setattr(self_deploy.settings, "deploy_ssh_host", "mva154") cmd = self_deploy.build_deploy_command("orchestrator", "ORCH-036", "feature/ORCH-036-x") assert cmd[0] == "ssh" assert "slin@mva154" in cmd remote = cmd[-1] assert "setsid" in remote # detached session assert remote.rstrip().endswith("&") # backgrounded assert " legacy path, self-deploy logic does not apply # --------------------------------------------------------------------------- def test_tc09_non_self_repo_uses_legacy_path(monkeypatch): """enduro-trails on the deploy-staging -> deploy edge: no Phase A interception, the deployer is enqueued for the deploy stage exactly as before ORCH-036.""" 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 left REAL -> N/A for non-self repo # Spy: self-deploy must not be initiated for a non-self repo. initiate = MagicMock() monkeypatch.setattr(stage_engine.self_deploy, "initiate_deploy", initiate) task_id = _make_task("deploy-staging", "enduro-trails", "feature/ET-009-x", "ET-009") res = advance_stage( task_id, "deploy-staging", "enduro-trails", "ET-009", "feature/ET-009-x", finished_agent="deployer", ) assert res.advanced is True assert _stage(task_id) == "deploy" assert res.note != "self-deploy-approval-pending" initiate.assert_not_called() # Legacy path enqueues the deployer for the deploy stage. jobs = _jobs() assert len(jobs) == 1 assert jobs[0]["agent"] == "deployer" # No self-deploy marker for the non-self repo. assert not self_deploy.has_marker("enduro-trails", "ET-009", self_deploy.APPROVE_REQUESTED)