150 lines
4.9 KiB
Python
150 lines
4.9 KiB
Python
"""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
|
|
|
|
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",
|
|
"show-proposal",
|
|
]:
|
|
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_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
|
|
)
|