teleo-infrastructure/tests/test_apply_worker.py

"""Unit tests for scripts/apply_worker.py.

Pure tests: query building, candidate parsing, render-hook expansion, and the
report-only enablement gate -- no live database. Runnable under pytest or
standalone (`python3 tests/test_apply_worker.py`).
"""

import argparse
from pathlib import Path
import sys

try:
    import pytest
except ImportError:  # standalone fallback so the file runs without pytest installed
    import contextlib
    import types

    pytest = types.SimpleNamespace()

    @contextlib.contextmanager
    def _raises(exc):
        try:
            yield
        except exc:
            return
        raise AssertionError(f"expected {exc.__name__} to be raised")

    pytest.raises = _raises

REPO_ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(REPO_ROOT / "scripts"))

import apply_worker as w  # noqa: E402


# --- candidate query ------------------------------------------------------ #
def test_candidate_query_filters_on_approved():
    sql = w.build_candidate_query()
    # The structural guarantee: worker can only ever act on approved proposals.
    assert "status = 'approved'" in sql
    # Never selects pending_review (no auto-approve possible).
    assert "pending_review" not in sql


def test_candidate_query_requires_apply_payload():
    sql = w.build_candidate_query()
    assert "payload ? 'apply_payload'" in sql


def test_candidate_query_lists_worker_types_only():
    sql = w.build_candidate_query()
    for t in w.WORKER_TYPES:
        assert f"'{t}'" in sql
    # a non-applyable type must not appear
    assert "reject_claim" not in sql


# --- candidate parsing ---------------------------------------------------- #
def test_parse_candidates_skips_blank_lines():
    out = '{"id": "a", "proposal_type": "add_edge", "agent_id": null}\n\n'
    rows = w.parse_candidates(out)
    assert len(rows) == 1
    assert rows[0]["id"] == "a"


# --- render hook ---------------------------------------------------------- #
def test_render_command_none_when_unset():
    assert w.build_render_command("", "agent-1") is None


def test_render_command_none_without_agent():
    assert w.build_render_command("python3 render_soul.py --agent-id {agent_id}", None) is None


def test_render_command_expands_agent_id():
    cmd = w.build_render_command("python3 render_soul.py --agent-id {agent_id}", "abc")
    assert cmd == ["python3", "render_soul.py", "--agent-id", "abc"]


# --- enablement gate ------------------------------------------------------ #
def _args(**over):
    base = dict(
        enable=False, limit=20, max_per_tick=1, max_attempts=3,
        failure_state_file=None, applied_by="kb-apply",
        apply_script="apply_proposal.py", render_cmd="",
        secrets_file="x", container="teleo-pg", db="teleo",
        host="127.0.0.1", role="kb_apply",
    )
    base.update(over)
    return argparse.Namespace(**base)


def test_report_only_gate_does_not_apply(monkeypatch=None):
    calls = []
    w.ap.load_password = lambda _f: "pw"
    w.fetch_candidates = lambda a, p: [{"id": "p1", "proposal_type": "revise_strategy", "agent_id": "ag"}]
    w.apply_one = lambda a, pid: calls.append(pid)

    rc = w.run(_args(enable=False))
    assert rc == 0
    assert calls == []  # disabled worker performs NO writes


def test_enabled_worker_applies_and_renders():
    applied, rendered = [], []
    w.ap.load_password = lambda _f: "pw"
    w.fetch_candidates = lambda a, p: [{"id": "p1", "proposal_type": "revise_strategy", "agent_id": "ag"}]
    w.apply_one = lambda a, pid: applied.append(pid)
    w.render_one = lambda a, agent_id: rendered.append(agent_id) or "ok"

    rc = w.run(_args(enable=True))
    assert rc == 0
    assert applied == ["p1"]
    assert rendered == ["ag"]  # revise_strategy triggers a render


def test_enabled_worker_skips_render_for_add_edge():
    applied, rendered = [], []
    w.ap.load_password = lambda _f: "pw"
    w.fetch_candidates = lambda a, p: [{"id": "e1", "proposal_type": "add_edge", "agent_id": None}]
    w.apply_one = lambda a, pid: applied.append(pid)
    w.render_one = lambda a, agent_id: rendered.append(agent_id) or "ok"

    rc = w.run(_args(enable=True))
    assert rc == 0
    assert applied == ["e1"]
    assert rendered == []  # add_edge changes the graph, not the rendered self


# --- per-tick cap + poison-pill ceiling ----------------------------------- #
def _cands(*ids):
    return [{"id": i, "proposal_type": "add_edge", "agent_id": None} for i in ids]


def test_partition_caps_to_max_per_tick():
    split = w.partition_candidates(_cands("a", "b", "c"), {}, max_attempts=3, max_per_tick=1)
    assert [c["id"] for c in split["to_apply"]] == ["a"]  # only 1 applied per tick
    assert split["poisoned"] == []


def test_partition_skips_poison_pill_over_ceiling():
    split = w.partition_candidates(_cands("a", "b"), {"a": 3}, max_attempts=3, max_per_tick=5)
    assert [c["id"] for c in split["poisoned"]] == ["a"]  # a is poisoned, never retried
    assert [c["id"] for c in split["to_apply"]] == ["b"]  # b still eligible


def test_enabled_worker_caps_applies_at_one_per_tick():
    applied = []
    w.ap.load_password = lambda _f: "pw"
    w.fetch_candidates = lambda a, p: _cands("e1", "e2", "e3")
    w.apply_one = lambda a, pid: applied.append(pid)

    rc = w.run(_args(enable=True, max_per_tick=1))
    assert rc == 0
    assert applied == ["e1"]  # capped to 1 even with 3 approved candidates


def test_failure_state_roundtrip_and_ceiling(tmp_path=None):
    import tempfile

    d = tempfile.mkdtemp()
    path = str(Path(d) / "failures.json")
    w.save_failure_state(path, {"p1": 2})
    assert w.load_failure_state(path) == {"p1": 2}
    # p1 at 2 with max_attempts 3 is still eligible; at 3 it poisons
    split = w.partition_candidates(_cands("p1"), {"p1": 3}, max_attempts=3, max_per_tick=1)
    assert [c["id"] for c in split["poisoned"]] == ["p1"]


def test_apply_failure_increments_persisted_count():
    import tempfile

    path = str(Path(tempfile.mkdtemp()) / "failures.json")
    w.ap.load_password = lambda _f: "pw"
    w.fetch_candidates = lambda a, p: _cands("bad")

    def _boom(a, pid):
        raise RuntimeError("deterministic failure")

    w.apply_one = _boom
    rc = w.run(_args(enable=True, failure_state_file=path, max_per_tick=1))
    assert rc == 1
    assert w.load_failure_state(path).get("bad") == 1  # count persisted for next tick


if __name__ == "__main__":
    import traceback

    failures = 0
    for name, fn in sorted(globals().items()):
        if name.startswith("test_") and callable(fn):
            try:
                fn()
                print(f"PASS {name}")
            except Exception:  # noqa: BLE001
                failures += 1
                print(f"FAIL {name}")
                traceback.print_exc()
    sys.exit(1 if failures else 0)