"""Unit tests for scripts/apply_proposal.py. Pure tests: exercise the SQL builders, payload validation, and dispatch without a live database. Runnable under pytest or standalone (`python3 tests/test_apply_proposal.py`). """ 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_proposal as ap # noqa: E402 # --- literals ------------------------------------------------------------- # def test_sql_literal_escapes_single_quotes(): assert ap.sql_literal("O'Brien") == "'O''Brien'" def test_sql_literal_none_is_null(): assert ap.sql_literal(None) == "null" def test_sql_literal_bool_and_number(): assert ap.sql_literal(True) == "true" assert ap.sql_literal(0.5) == "0.5" # --- revise_strategy ------------------------------------------------------ # def _revise_payload(): return { "apply_payload": { "agent_id": "11111111-1111-1111-1111-111111111111", "strategy": { "diagnosis": "Species-level inflection.", "guiding_policy": "Build Teleo as a multi-agent public intelligence.", "proximate_objectives": ["route before creating", "keep KB calibrated"], }, "strategy_nodes": [ {"node_type": "diagnosis", "title": "Inflection", "body": "b1", "rank": 1}, {"node_type": "policy", "title": "Multi-agent PI", "body": "b2", "rank": 1}, ], } } def test_revise_strategy_sql_shape(): sql = ap.build_revise_strategy_sql(_revise_payload()["apply_payload"], "pid-1", "m3ta") assert sql.startswith("begin;") assert sql.rstrip().endswith("commit;") # deactivate old, insert versioned active, retire nodes, insert new nodes assert "update public.strategies" in sql assert "active = false" in sql assert "coalesce(max(version), 0) + 1" in sql assert "update public.strategy_nodes" in sql assert "insert into public.strategy_nodes" in sql # ledger + invariant assert "status = 'applied'" in sql assert "exactly one active strategy" in sql # proximate_objectives rendered as jsonb assert "::jsonb" in sql def test_revise_strategy_requires_agent_id(): payload = _revise_payload()["apply_payload"] del payload["agent_id"] with pytest.raises(ValueError): ap.build_revise_strategy_sql(payload, "pid-1", None) def test_revise_strategy_rejects_bad_node_type(): payload = _revise_payload()["apply_payload"] payload["strategy_nodes"][0]["node_type"] = "manifesto" with pytest.raises(ValueError): ap.build_revise_strategy_sql(payload, "pid-1", None) # --- add_edge ------------------------------------------------------------- # def test_add_edge_sql_dedup_guard(): payload = {"from_claim": "aaaa", "to_claim": "bbbb", "edge_type": "supersedes", "weight": 0.9} sql = ap.build_add_edge_sql(payload, "pid-2", None) assert "insert into public.claim_edges" in sql assert "not exists" in sql assert "::edge_type" in sql assert "status = 'applied'" in sql def test_add_edge_rejects_self_loop(): payload = {"from_claim": "same", "to_claim": "same", "edge_type": "supports"} with pytest.raises(ValueError): ap.build_add_edge_sql(payload, "pid-2", None) # --- attach_evidence ------------------------------------------------------ # def test_attach_evidence_sql_with_source_id(): payload = {"evidence": [{"claim_id": "cccc", "source_id": "dddd", "role": "grounds", "weight": 0.78}]} sql = ap.build_attach_evidence_sql(payload, "pid-3", None) assert "insert into public.claim_evidence" in sql assert "::evidence_role" in sql assert "not exists" in sql def test_attach_evidence_requires_source_id(): payload = {"evidence": [{"claim_id": "cccc"}]} with pytest.raises(ValueError): ap.build_attach_evidence_sql(payload, "pid-3", None) # --- dispatch + guards ---------------------------------------------------- # def test_build_apply_sql_requires_apply_payload(): proposal = {"id": "pid", "proposal_type": "add_edge", "payload": {"rationale": "x"}} with pytest.raises(ValueError): ap.build_apply_sql(proposal, None) def test_build_apply_sql_rejects_unsupported_type(): proposal = {"id": "pid", "proposal_type": "reject_claim", "payload": {"apply_payload": {}}} with pytest.raises(ValueError): ap.build_apply_sql(proposal, None) def test_assert_applyable_blocks_pending(): with pytest.raises(SystemExit): ap.assert_applyable({"id": "pid", "status": "pending_review"}) def test_assert_applyable_blocks_already_applied(): with pytest.raises(SystemExit): ap.assert_applyable({"id": "pid", "status": "applied"}) def test_assert_applyable_allows_approved(): ap.assert_applyable({"id": "pid", "status": "approved"}) # no raise # --- ledger flip: concurrency guard + FK stamp --------------------------- # def test_ledger_flip_asserts_rowcount_one(): # The flip must guard against a concurrent double-apply by asserting exactly # one 'approved' row moved, not by re-reading status afterwards. sql = ap.build_add_edge_sql({"from_claim": "a", "to_claim": "b", "edge_type": "supports"}, "pid", None) assert "get diagnostics" in sql assert "flipped <> 1" in sql assert "and status = 'approved'" in sql def test_ledger_flip_stamps_agent_fk(): sql = ap.build_add_edge_sql({"from_claim": "a", "to_claim": "b", "edge_type": "supports"}, "pid", "kb-apply") # Hard resolve into a variable + NOT-NULL assert, never a silent inline # subselect that would stamp NULL on an unresolved handle. assert "select id into resolved_agent_id" in sql assert "resolved_agent_id is null then" in sql assert "applied_by_agent_id = resolved_agent_id" in sql assert "applied_by_handle = 'kb-apply'" in sql def test_build_apply_sql_defaults_applied_by_to_service_agent(): proposal = { "id": "pid", "proposal_type": "add_edge", "payload": {"apply_payload": {"from_claim": "a", "to_claim": "b", "edge_type": "supports"}}, } sql = ap.build_apply_sql(proposal, None) assert f"applied_by_handle = '{ap.SERVICE_AGENT_HANDLE}'" in sql 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)