orchestrator/tests/test_orch026_dep_cycles.py

"""ORCH-026 Level B — dependency cycle / deadlock detection (TC-B03, TC-B04).

TC-B03  detect_cycle is deterministic: A->B->A (and longer) is detected; an
        acyclic graph yields None. Pure function (edges injected).
TC-B04  a detected cycle escalates: set_issue_blocked + a Telegram alert, with
        no worker crash and no blocking of other tasks (never-raise).
"""
import os
import tempfile

import pytest

os.environ["ORCH_DB_PATH"] = os.path.join(tempfile.gettempdir(), "test_orch026_cycles.db")
os.environ.setdefault("ORCH_GITEA_TOKEN", "test-token")
os.environ.setdefault("ORCH_PLANE_API_TOKEN", "test-token")

import src.db as db  # noqa: E402
from src.db import init_db, get_db  # noqa: E402
from src import task_deps  # noqa: E402


@pytest.fixture(autouse=True)
def fresh_db(tmp_path, monkeypatch):
    dbfile = tmp_path / "cycles.db"
    monkeypatch.setattr(db.settings, "db_path", str(dbfile))
    monkeypatch.setattr(db.settings, "task_deps_enabled", True, raising=False)
    init_db()
    yield


# ----------------------------------------------------------------- TC-B03
def test_detect_two_node_cycle():
    # edge (B, A) means "B depends on A"; 1->2 and 2->1 is a 2-cycle.
    edges = [(1, 2), (2, 1)]
    cyc = task_deps.detect_cycle(1, edges=edges)
    assert cyc is not None
    assert cyc[0] == cyc[-1]  # closed cycle
    assert set(cyc) == {1, 2}


def test_detect_longer_cycle():
    edges = [(1, 2), (2, 3), (3, 1)]
    cyc = task_deps.detect_cycle(1, edges=edges)
    assert cyc is not None
    assert set(cyc) >= {1, 2, 3}


def test_acyclic_graph_has_no_cycle():
    edges = [(1, 2), (2, 3), (1, 3)]  # DAG
    assert task_deps.detect_cycle(1, edges=edges) is None
    assert task_deps.find_any_cycle(edges=edges) is None


def test_find_any_cycle_scans_whole_graph():
    # A disconnected cycle 10<->11 not reachable from node 1.
    edges = [(1, 2), (10, 11), (11, 10)]
    assert task_deps.detect_cycle(1, edges=edges) is None
    cyc = task_deps.find_any_cycle(edges=edges)
    assert cyc is not None
    assert set(cyc) == {10, 11}


def test_detect_cycle_never_raises_on_garbage():
    assert task_deps.detect_cycle(None) is None
    # Malformed edge list -> swallowed -> None.
    assert task_deps.detect_cycle(1, edges="not-a-list") is None


# ----------------------------------------------------------------- TC-B04
def _make_task(work_item_id, stage="development"):
    conn = get_db()
    cur = conn.execute(
        "INSERT INTO tasks (plane_id, work_item_id, repo, branch, stage) "
        "VALUES (?, ?, ?, ?, ?)",
        (work_item_id, work_item_id, "orchestrator", f"feature/{work_item_id}", stage),
    )
    tid = cur.lastrowid
    conn.commit()
    conn.close()
    return tid


def test_handle_cycle_blocks_and_alerts(monkeypatch):
    a = _make_task("ORCH-60")
    b = _make_task("ORCH-61")
    db.add_dependency(a, b)
    db.add_dependency(b, a)  # cycle a<->b

    blocked = []
    alerts = []
    import src.plane_sync as plane_sync
    import src.notifications as notifications
    monkeypatch.setattr(plane_sync, "set_issue_blocked",
                        lambda wi, *a, **k: blocked.append(wi), raising=False)
    monkeypatch.setattr(notifications, "send_telegram",
                        lambda text, *a, **k: alerts.append(text), raising=False)

    cyc = task_deps.detect_cycle(a)
    assert cyc is not None
    ok = task_deps.handle_cycle(cyc)
    assert ok is True
    assert set(blocked) == {"ORCH-60", "ORCH-61"}
    assert len(alerts) == 1
    assert "ORCH-60" in alerts[0] and "ORCH-61" in alerts[0]


def test_handle_cycle_never_raises_when_notify_fails(monkeypatch):
    a = _make_task("ORCH-70")
    b = _make_task("ORCH-71")
    db.add_dependency(a, b)
    db.add_dependency(b, a)
    import src.plane_sync as plane_sync
    import src.notifications as notifications

    def _boom(*a, **k):
        raise RuntimeError("plane down")

    monkeypatch.setattr(plane_sync, "set_issue_blocked", _boom, raising=False)
    monkeypatch.setattr(notifications, "send_telegram", _boom, raising=False)
    cyc = task_deps.detect_cycle(a)
    # Must not propagate the exception (AC-G1).
    assert task_deps.handle_cycle(cyc) in (True, False)


def test_declare_dependency_escalates_cycle(monkeypatch):
    """declare_dependency surfaces a freshly-introduced cycle at declaration."""
    a = _make_task("ORCH-80")
    b = _make_task("ORCH-81")
    handled = []
    monkeypatch.setattr(task_deps, "handle_cycle",
                        lambda cyc: handled.append(cyc), raising=False)
    assert task_deps.declare_dependency(a, b) is True
    assert handled == []  # no cycle yet
    # Closing the loop -> handle_cycle invoked.
    assert task_deps.declare_dependency(b, a) is True
    assert len(handled) == 1