"""Source-control checks for the Hermes leoclean KB bridge files.""" from __future__ import annotations import ast import contextlib import importlib.util import io import re import subprocess from pathlib import Path from types import SimpleNamespace ROOT = Path(__file__).resolve().parents[1] BRIDGE_DIR = ROOT / "hermes-agent" / "leoclean-bin" def test_leoclean_bridge_files_are_present_and_parseable() -> None: cloudsql_tool = BRIDGE_DIR / "cloudsql_memory_tool.py" vps_tool = BRIDGE_DIR / "kb_tool.py" wrapper = BRIDGE_DIR / "teleo-kb" assert cloudsql_tool.exists() assert vps_tool.exists() assert wrapper.exists() ast.parse(cloudsql_tool.read_text()) ast.parse(vps_tool.read_text()) subprocess.run(["bash", "-n", str(wrapper)], check=True) def test_cloudsql_wrapper_supports_expected_review_gated_commands() -> None: wrapper_text = (BRIDGE_DIR / "teleo-kb").read_text() cloudsql_text = (BRIDGE_DIR / "cloudsql_memory_tool.py").read_text() for command in [ "status", "search", "context", "show", "evidence", "edges", "propose-core-change", "propose-edge", "list-proposals", "search-proposals", "show-proposal", "decision-matrix-status", ]: assert command in wrapper_text assert command in cloudsql_text assert "kb_stage.kb_proposals" in cloudsql_text assert "canonical apply" in cloudsql_text.lower() def test_bridge_edge_proposals_stage_strict_apply_payloads() -> None: cloudsql_text = (BRIDGE_DIR / "cloudsql_memory_tool.py").read_text() vps_text = (BRIDGE_DIR / "kb_tool.py").read_text() for text in (cloudsql_text, vps_text): assert "'proposal_type', proposal_type" in text or "select 'add_edge'" in text assert "'apply_payload', jsonb_build_object(" in text assert "'from_claim', from_row.id::text" in text assert "'to_claim', to_row.id::text" in text assert "'edge_type', p.edge_type::text" in text def test_bridge_source_does_not_commit_raw_secret_values() -> None: combined = "\n".join(path.read_text(errors="replace") for path in BRIDGE_DIR.iterdir() if path.is_file()) forbidden_patterns = [ r"bot\d+:[A-Za-z0-9_-]{20,}", r"postgres://[^\\s]+:[^\\s]+@", r"-----BEGIN [A-Z ]*PRIVATE KEY-----", r"sk-[A-Za-z0-9]{20,}", r"xox[baprs]-[A-Za-z0-9-]{20,}", r"AIza[0-9A-Za-z_-]{20,}", ] for pattern in forbidden_patterns: assert not re.search(pattern, combined), pattern def _load_module(path: Path): spec = importlib.util.spec_from_file_location(path.stem, path) assert spec and spec.loader module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) return module def test_vps_bridge_search_proposals_finds_approved_rows_by_payload(monkeypatch) -> None: module = _load_module(BRIDGE_DIR / "kb_tool.py") captured_sql: list[str] = [] def fake_psql_json(_args, sql): captured_sql.append(sql) return [{"id": "a64df080-8502-42e2-98f4-9bbdecb8da73", "status": "approved"}] monkeypatch.setattr(module, "psql_json", fake_psql_json) args = SimpleNamespace(query="Helmer 7 Powers", status="all", limit=20) result = module.search_proposals(args) assert result["proposals"][0]["status"] == "approved" assert {"helmer", "powers"} <= set(result["terms"]) sql = captured_sql[0] assert "payload::text ilike any" in sql assert "coalesce(rationale, '') ilike any" in sql assert "status =" not in sql def test_vps_bridge_proposal_list_prints_rationale_for_non_edge_rows(capsys) -> None: module = _load_module(BRIDGE_DIR / "kb_tool.py") module.print_proposal_list( [ { "id": "a64df080-8502-42e2-98f4-9bbdecb8da73", "proposal_type": "attach_evidence", "status": "approved", "proposed_by_handle": "leo", "channel": "telegram", "created_at": "2026-07-09 00:00:00+00", "reviewed_at": "2026-07-09 00:10:00+00", "applied_at": None, "payload": {"title": "Helmer 7 Powers"}, "rationale": "Revised Helmer packet should remain visible as approved but unapplied.", } ] ) output = capsys.readouterr().out assert "attach_evidence" in output assert "approved" in output assert "applied_at: `-`" in output assert "Revised Helmer packet" in output def test_vps_bridge_decision_matrix_status_checks_schema_tables(monkeypatch) -> None: module = _load_module(BRIDGE_DIR / "kb_tool.py") captured_sql: list[str] = [] def fake_psql_json(_args, sql): captured_sql.append(sql) return [ { "artifact": "teleo_kb_decision_matrix_status", "all_required_tables_present": False, "any_decision_matrix_table_present": False, "decision_matrix_tables": {}, "proposal_status_counts": {}, "guidance": "Decision-matrix approval schema is absent or incomplete.", } ] monkeypatch.setattr(module, "psql_json", fake_psql_json) result = module.decision_matrix_status(SimpleNamespace()) assert result["all_required_tables_present"] is False sql = captured_sql[0] for table in ( "public.matrix_voters", "public.proposal_votes", "public.proposal_decisions", "kb_stage.matrix_voters", "kb_stage.proposal_votes", "kb_stage.proposal_decisions", ): assert table in sql def test_cloudsql_bridge_matches_direct_claim_readback_commands(monkeypatch) -> None: module = _load_module(BRIDGE_DIR / "cloudsql_memory_tool.py") captured_sql: list[str] = [] def fake_psql_json_lines(_args, sql, db=None): captured_sql.append(sql) if "matrix_voters" in sql: return [ { "artifact": "teleo_cloudsql_kb_decision_matrix_status", "all_required_tables_present": False, "any_decision_matrix_table_present": False, "decision_matrix_tables": {}, "proposal_status_counts": {}, "guidance": "Decision-matrix approval schema is absent or incomplete.", } ] return [{"id": "a64df080-8502-42e2-98f4-9bbdecb8da73", "status": "approved"}] monkeypatch.setattr(module, "psql_json_lines", fake_psql_json_lines) args = SimpleNamespace(query="Helmer 7 Powers", status="all", limit=20, canonical_db="teleo") proposal_result = module.search_proposals(args) matrix_result = module.decision_matrix_status(SimpleNamespace(canonical_db="teleo")) assert proposal_result["artifact"] == "teleo_cloudsql_kb_proposal_search" assert proposal_result["proposals"][0]["status"] == "approved" assert matrix_result["artifact"] == "teleo_cloudsql_kb_decision_matrix_status" assert any("payload::text ilike any" in sql for sql in captured_sql) assert any("public.proposal_decisions" in sql for sql in captured_sql) def test_kb_bridges_emit_public_claim_links_for_telegram_rendering() -> None: claim_id = "d3fb892b-3c5a-4700-9512-55e5c680eec1" expected = f"https://leo.livingip.xyz/kb/claims/{claim_id}" for filename in ("kb_tool.py", "cloudsql_memory_tool.py"): module = _load_module(BRIDGE_DIR / filename) assert module.claim_url(claim_id) == expected assert module.markdown_claim_link(claim_id, "d3fb892b") == ( f"[`d3fb892b`]({expected})" ) assert module.markdown_claim_link(claim_id, "claim `with ticks`") == ( f"[`claim 'with ticks'`]({expected})" ) def test_vps_bridge_markdown_links_claim_text_and_edges() -> None: module = _load_module(BRIDGE_DIR / "kb_tool.py") claim_id = "d3fb892b-3c5a-4700-9512-55e5c680eec1" connected_id = "9d5281fe-7ee4-4fe1-b1bf-c54c4d7fb6a5" data = { "query": "strategy kernel", "context_rows": [], "claims": [ { "id": claim_id, "text": "Claims should be easy to scan in Telegram.", "type": "belief", "confidence": 0.8, "tags": ["telegram"], "score": 4, "evidence_count": 2, "edge_count": 1, "evidence": [], "edges": [ { "direction": "outgoing", "edge_type": "supports", "connected_id": connected_id, "connected_text": "Full claim pages expose body, evidence, and graph edges.", } ], } ], } buffer = io.StringIO() with contextlib.redirect_stdout(buffer): module.print_claim_bundle(data) markdown = buffer.getvalue() assert ( f"[`Claims should be easy to scan in Telegram.`](https://leo.livingip.xyz/kb/claims/{claim_id})" in markdown ) assert ( f"- open full claim/body/edges: [`claim page`](https://leo.livingip.xyz/kb/claims/{claim_id})" in markdown ) assert ( f"[`Full claim pages expose body, evidence, and graph edges.`](https://leo.livingip.xyz/kb/claims/{connected_id})" in markdown )