95 lines
3.7 KiB
Python
95 lines
3.7 KiB
Python
"""Tests for automatic live database-contract injection into Leo turns."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import importlib.util
|
|
import json
|
|
import subprocess
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
PLUGIN = ROOT / "hermes-agent" / "leoclean-plugins" / "vps" / "leo-db-context" / "__init__.py"
|
|
|
|
|
|
def load_plugin():
|
|
spec = importlib.util.spec_from_file_location("leo_db_context_plugin", PLUGIN)
|
|
assert spec and spec.loader
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
def test_plugin_registers_pre_llm_call_hook() -> None:
|
|
module = load_plugin()
|
|
registered = []
|
|
module.register(SimpleNamespace(register_hook=lambda name, callback: registered.append((name, callback))))
|
|
assert len(registered) == 1
|
|
assert registered[0][0] == "pre_llm_call"
|
|
|
|
|
|
def test_plugin_injects_only_operational_contracts_and_writes_redacted_trace(
|
|
tmp_path: Path, monkeypatch
|
|
) -> None:
|
|
module = load_plugin()
|
|
home = tmp_path / "profile"
|
|
kb_tool = home / "bin" / "kb_tool.py"
|
|
kb_tool.parent.mkdir(parents=True)
|
|
kb_tool.write_text("# fixture\n", encoding="utf-8")
|
|
trace = tmp_path / "trace.jsonl"
|
|
monkeypatch.setenv("LEO_DB_CONTEXT_TRACE_PATH", str(trace))
|
|
payload = {
|
|
"query": "must not be injected",
|
|
"claims": [{"text": "must not be injected"}],
|
|
"operational_contracts": [
|
|
{"id": "reply_budget", "hard_max_words": 220},
|
|
{"id": "source_intake", "capability_tier": "build-local compiler only"},
|
|
],
|
|
}
|
|
calls = []
|
|
|
|
def fake_runner(command, **kwargs):
|
|
calls.append((command, kwargs))
|
|
return subprocess.CompletedProcess(command, 0, stdout=json.dumps(payload), stderr="")
|
|
|
|
context = module.build_database_context(
|
|
"operator secret-shaped but nonsecret message", hermes_home=home, runner=fake_runner
|
|
)
|
|
|
|
assert "reply_budget" in context
|
|
assert "source_intake" in context
|
|
assert "build-local compiler only" in context
|
|
assert "must not be injected" not in context
|
|
assert "operator secret-shaped" not in trace.read_text(encoding="utf-8")
|
|
record = json.loads(trace.read_text(encoding="utf-8"))
|
|
assert record["status"] == "ok"
|
|
assert record["injected"] is True
|
|
assert record["contract_ids"] == ["reply_budget", "source_intake"]
|
|
command, kwargs = calls[0]
|
|
assert command[2:5] == ["--local", "context", "operator secret-shaped but nonsecret message"]
|
|
assert command[-6:] == ["--limit", "0", "--context-limit", "0", "--format", "json"]
|
|
assert kwargs["timeout"] == module.DEFAULT_TIMEOUT_SECONDS
|
|
|
|
|
|
def test_plugin_fails_closed_when_database_contract_lookup_fails(tmp_path: Path) -> None:
|
|
module = load_plugin()
|
|
home = tmp_path / "profile"
|
|
kb_tool = home / "bin" / "kb_tool.py"
|
|
kb_tool.parent.mkdir(parents=True)
|
|
kb_tool.write_text("# fixture\n", encoding="utf-8")
|
|
|
|
def fake_runner(command, **_kwargs):
|
|
return subprocess.CompletedProcess(command, 9, stdout="", stderr="database unavailable")
|
|
|
|
context = module.build_database_context("What is live?", hermes_home=home, runner=fake_runner)
|
|
assert 'status="unavailable"' in context
|
|
assert "Do not infer current schema" in context
|
|
assert "database unavailable" not in context
|
|
|
|
|
|
def test_handler_harness_retains_database_context_proof_fields() -> None:
|
|
source = (ROOT / "scripts" / "run_leo_direct_claim_handler_suite.py").read_text(encoding="utf-8")
|
|
assert "LEO_DB_CONTEXT_TRACE_PATH" in source
|
|
assert '"database_context_trace"' in source
|
|
assert '"database_context_injection_count"' in source
|
|
assert '"database_context_all_ok"' in source
|