"""ORCH-053: tests for the gate-side stuck-task reconciler (F-1) + lifecycle. These cover the F-1 sweeper (``Reconciler.reconcile_gate_once``), the per-stage grace / config (``grace_for_stage``), the no-spam guarantee, the analysis carve- out (AC-16), never-raise isolation, the kill-switch, the unblock observability (AC-12 / F-4) and the restart-safe daemon thread (AC-11). Everything that touches the network (the quality gate, Plane sync, Telegram) is mocked at the src.stage_engine / src.reconciler level so the reconciler runs against a real isolated sqlite DB (same convention as test_stage_engine.py). """ import os import tempfile import pytest # Isolated test DB (set BEFORE importing src.* so settings picks it up). _test_db = os.path.join(tempfile.gettempdir(), "test_orchestrator_reconciler.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, enqueue_job # noqa: E402 from src import stage_engine # noqa: E402 from src import reconciler as reconciler_mod # noqa: E402 from src.reconciler import Reconciler, grace_for_stage # noqa: E402 # --------------------------------------------------------------------------- # Fixtures # --------------------------------------------------------------------------- @pytest.fixture(autouse=True) def fresh_db(monkeypatch): """Fresh isolated DB per test.""" monkeypatch.setattr(_db.settings, "db_path", _test_db) if os.path.exists(_test_db): os.unlink(_test_db) init_db() yield @pytest.fixture(autouse=True) def silence_side_effects(monkeypatch): """No-op every Plane/Telegram/notification side effect in the engine so the real advance_stage runs deterministically and offline.""" for name in ( "notify_stage_change", "notify_qg_failure", "notify_approve_requested", "notify_error", "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(), raising=False) def _make_task(stage, *, repo="enduro-trails", branch="feature/ET-001-x", wi="ET-001", age_s=None): """Insert a task; if age_s is given, backdate updated_at by that many secs.""" 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 if age_s is not None: conn.execute( "UPDATE tasks SET updated_at = datetime('now', ?) WHERE id = ?", (f"-{int(age_s)} seconds", task_id), ) conn.commit() conn.close() return task_id def _stage_of(task_id): conn = get_db() row = conn.execute("SELECT stage FROM tasks WHERE id = ?", (task_id,)).fetchone() conn.close() return row["stage"] def _jobs_for(task_id, agent=None): conn = get_db() if agent: rows = conn.execute( "SELECT * FROM jobs WHERE task_id = ? AND agent = ?", (task_id, agent) ).fetchall() else: rows = conn.execute( "SELECT * FROM jobs WHERE task_id = ?", (task_id,) ).fetchall() conn.close() return [dict(r) for r in rows] def _green_ci(monkeypatch, value=(True, "CI green")): """Patch the check_ci_green entry in QG_CHECKS; return the mock.""" m = MagicMock(return_value=value) monkeypatch.setitem(stage_engine.QG_CHECKS, "check_ci_green", m) return m # --------------------------------------------------------------------------- # TC-01: happy path — stuck development task is advanced to review # --------------------------------------------------------------------------- def test_tc01_advances_stuck_development_task(monkeypatch): _green_ci(monkeypatch) task_id = _make_task("development", age_s=3600) # well past grace Reconciler().reconcile_gate_once() assert _stage_of(task_id) == "review" reviewer_jobs = _jobs_for(task_id, "reviewer") assert len(reviewer_jobs) == 1 # --------------------------------------------------------------------------- # TC-02: source of truth is the gate — advance goes through advance_stage # with finished_agent=None (no own update_task_stage/enqueue_job). # --------------------------------------------------------------------------- def test_tc02_advances_via_advance_stage_finished_agent_none(monkeypatch): _green_ci(monkeypatch) spy = MagicMock(wraps=stage_engine.advance_stage) # advance_if_gate_passed resolves advance_stage as a module global. monkeypatch.setattr(stage_engine, "advance_stage", spy) task_id = _make_task("development", age_s=3600) Reconciler().reconcile_gate_once() assert spy.call_count == 1 # finished_agent must be None (the webhook path). _args, kwargs = spy.call_args assert kwargs.get("finished_agent", "MISSING") is None assert spy.call_args.args[0] == task_id # --------------------------------------------------------------------------- # TC-03: task with an active job is skipped — gate not evaluated, no advance. # --------------------------------------------------------------------------- def test_tc03_active_job_skipped(monkeypatch): ci = _green_ci(monkeypatch) spy = MagicMock(wraps=stage_engine.advance_stage) monkeypatch.setattr(stage_engine, "advance_stage", spy) task_id = _make_task("development", age_s=3600) enqueue_job("reviewer", "enduro-trails", task_id=task_id) # active (queued) Reconciler().reconcile_gate_once() assert _stage_of(task_id) == "development" ci.assert_not_called() spy.assert_not_called() # --------------------------------------------------------------------------- # TC-04: per-stage grace — fresh task untouched, at-threshold task eligible. # --------------------------------------------------------------------------- def test_tc04_grace_boundary(monkeypatch): monkeypatch.setattr(reconciler_mod.settings, "reconcile_grace_default_s", 600) _green_ci(monkeypatch) fresh = _make_task("development", branch="feature/ET-002-fresh", wi="ET-002", age_s=10) # < grace -> untouched stuck = _make_task("development", branch="feature/ET-003-stuck", wi="ET-003", age_s=3600) # >= grace -> advanced Reconciler().reconcile_gate_once() assert _stage_of(fresh) == "development" assert _stage_of(stuck) == "review" # --------------------------------------------------------------------------- # TC-05: grace_for_stage reads overrides JSON; bad JSON -> default, no crash. # --------------------------------------------------------------------------- def test_tc05_grace_for_stage_overrides(monkeypatch): monkeypatch.setattr(reconciler_mod.settings, "reconcile_grace_default_s", 600) monkeypatch.setattr( reconciler_mod.settings, "reconcile_grace_overrides_json", '{"development": 30, "review": 7200}', ) assert grace_for_stage("development") == 30 assert grace_for_stage("review") == 7200 # missing key -> default assert grace_for_stage("testing") == 600 def test_tc05_grace_for_stage_invalid_json_falls_back(monkeypatch): monkeypatch.setattr(reconciler_mod.settings, "reconcile_grace_default_s", 600) monkeypatch.setattr( reconciler_mod.settings, "reconcile_grace_overrides_json", "{not valid json" ) # Must not raise, must fall back to the default. assert grace_for_stage("development") == 600 # --------------------------------------------------------------------------- # TC-06: no spam — a stable-red gate never advances and never notifies, even # across many ticks. # --------------------------------------------------------------------------- def test_tc06_red_gate_no_spam(monkeypatch): _green_ci(monkeypatch, value=(False, "CI red")) task_id = _make_task("development", age_s=3600) rec = Reconciler() for _ in range(5): rec.reconcile_gate_once() assert _stage_of(task_id) == "development" # The QG-failure notification branch inside advance_stage must never fire, # because advance_if_gate_passed returns None on a red gate (no advance call). stage_engine.notify_qg_failure.assert_not_called() stage_engine.plane_notify_qg.assert_not_called() assert rec.unblocked_total == 0 # --------------------------------------------------------------------------- # TC-07: silence when in sync — done / busy / within-grace tasks => no advance. # --------------------------------------------------------------------------- def test_tc07_silence_when_in_sync(monkeypatch): _green_ci(monkeypatch) spy = MagicMock(wraps=stage_engine.advance_stage) monkeypatch.setattr(stage_engine, "advance_stage", spy) _make_task("done", branch="feature/ET-010-done", wi="ET-010", age_s=3600) fresh = _make_task("development", branch="feature/ET-011-fresh", wi="ET-011", age_s=5) busy = _make_task("development", branch="feature/ET-012-busy", wi="ET-012", age_s=3600) enqueue_job("reviewer", "enduro-trails", task_id=busy) rec = Reconciler() rec.reconcile_gate_once() spy.assert_not_called() assert rec.unblocked_total == 0 assert _stage_of(fresh) == "development" # --------------------------------------------------------------------------- # TC-08 (AC-16): F-1 never advances the human analysis gate. # --------------------------------------------------------------------------- def test_tc08_analysis_not_advanced_by_f1(monkeypatch): # Even if the analysis gate would "pass", F-1 must not touch analysis. monkeypatch.setitem( stage_engine.QG_CHECKS, "check_analysis_approved", MagicMock(return_value=(True, "approved")), ) spy = MagicMock(wraps=stage_engine.advance_stage) monkeypatch.setattr(stage_engine, "advance_stage", spy) task_id = _make_task("analysis", age_s=3600) Reconciler().reconcile_gate_once() assert _stage_of(task_id) == "analysis" spy.assert_not_called() # --------------------------------------------------------------------------- # TC-09: never-raise — one task blowing up does not stop the others. # --------------------------------------------------------------------------- def test_tc09_never_raise_isolates_failure(monkeypatch): calls = [] def boom(task_id, stage, repo, wi, branch): calls.append(task_id) raise RuntimeError("boom") monkeypatch.setattr(reconciler_mod, "advance_if_gate_passed", boom) t1 = _make_task("development", branch="feature/ET-020-a", wi="ET-020", age_s=3600) t2 = _make_task("development", branch="feature/ET-021-b", wi="ET-021", age_s=3600) # Must not raise despite both tasks raising inside advance_if_gate_passed. Reconciler().reconcile_gate_once() assert set(calls) == {t1, t2} # both attempted # --------------------------------------------------------------------------- # TC-10: kill-switches. # --------------------------------------------------------------------------- def test_tc10_kill_switch_disables_gate(monkeypatch): monkeypatch.setattr(reconciler_mod.settings, "reconcile_enabled", False) spy = MagicMock(wraps=stage_engine.advance_stage) monkeypatch.setattr(stage_engine, "advance_stage", spy) _green_ci(monkeypatch) task_id = _make_task("development", age_s=3600) Reconciler().reconcile_gate_once() assert _stage_of(task_id) == "development" spy.assert_not_called() def test_tc10_plane_switch_mutes_only_f2(monkeypatch): monkeypatch.setattr(reconciler_mod.settings, "reconcile_enabled", True) monkeypatch.setattr(reconciler_mod.settings, "reconcile_plane_enabled", False) plane_pass = MagicMock() monkeypatch.setattr(reconciler_mod.Reconciler, "_reconcile_plane_project", plane_pass) # F-2 muted -> reconcile_plane_once is a no-op. Reconciler().reconcile_plane_once() plane_pass.assert_not_called() # F-1 still runs. _green_ci(monkeypatch) task_id = _make_task("development", age_s=3600) Reconciler().reconcile_gate_once() assert _stage_of(task_id) == "review" # --------------------------------------------------------------------------- # TC-20: observability — explicit unblock log line + telegram (AC-12 / F-4). # --------------------------------------------------------------------------- def test_tc20_unblock_logs_and_notifies(monkeypatch, caplog): _green_ci(monkeypatch) monkeypatch.setattr(reconciler_mod.settings, "reconcile_notify_unblock", True) tg = MagicMock() monkeypatch.setattr(reconciler_mod, "send_telegram", tg) _make_task("development", wi="ET-042", age_s=3600) rec = Reconciler() with caplog.at_level("INFO", logger="orchestrator.reconciler"): rec.reconcile_gate_once() # Exact AC-12 contract string. assert "reconciler: ET-042 development разблокирована (потерян webhook)" in caplog.text assert rec.unblocked_total == 1 assert rec.last_unblocked == "ET-042" tg.assert_called_once() def test_tc20_no_telegram_when_disabled(monkeypatch): _green_ci(monkeypatch) monkeypatch.setattr(reconciler_mod.settings, "reconcile_notify_unblock", False) tg = MagicMock() monkeypatch.setattr(reconciler_mod, "send_telegram", tg) _make_task("development", wi="ET-043", age_s=3600) Reconciler().reconcile_gate_once() tg.assert_not_called() # --------------------------------------------------------------------------- # TC-21: restart-safe daemon thread — start/stop/idempotent start. # --------------------------------------------------------------------------- def test_tc21_daemon_thread_lifecycle(monkeypatch): # Avoid any real work in the loop: disable both branches, big interval. monkeypatch.setattr(reconciler_mod.settings, "reconcile_enabled", False) rec = Reconciler(interval_s=60) rec.start() assert rec._thread is not None and rec._thread.is_alive() first_thread = rec._thread # Idempotent: a second start does not spawn a new thread. rec.start() assert rec._thread is first_thread rec.stop(timeout=5.0) assert not first_thread.is_alive()