tests/test_xkb_agent_hook.py
Rama main · réplica de lectura del repositorio original.
from __future__ import annotations
import importlib.util
import json
import sys
import tempfile
import unittest
from pathlib import Path
from unittest import mock
SCRIPTS = Path(__file__).resolve().parents[1] / "scripts"
def load(name: str):
spec = importlib.util.spec_from_file_location(name, SCRIPTS / f"{name}.py")
assert spec and spec.loader
module = importlib.util.module_from_spec(spec)
sys.modules[spec.name] = module
spec.loader.exec_module(module)
return module
hook = load("xkb_agent_hook")
installer = load("xkb_install_agent_hook")
class SessionIdentityTests(unittest.TestCase):
def test_session_id_is_preferred_and_cwd_is_the_fallback(self) -> None:
self.assertEqual(hook.session_key({"session_id": "abc", "cwd": "/tmp"}), "abc")
self.assertEqual(hook.session_key({"cwd": "/work/project"}), "/work/project")
self.assertEqual(hook.session_key({}), "default")
def test_turn_id_is_stable_for_the_same_prompt(self) -> None:
first = hook.turn_id("session", "same prompt")
self.assertEqual(first, hook.turn_id("session", "same prompt"))
self.assertNotEqual(first, hook.turn_id("session", "other prompt"))
class FailOpenTests(unittest.TestCase):
def test_service_being_down_never_blocks_the_agent(self) -> None:
"""A knowledge base that cannot be reached must not stop the user working."""
payload = json.dumps({"hook_event_name": "UserPromptSubmit", "prompt": "hi", "session_id": "s"})
with mock.patch.object(hook, "call", side_effect=OSError("connection refused")), \
mock.patch("sys.stdin", new=__import__("io").StringIO(payload)):
self.assertEqual(hook.main(), 0)
def test_malformed_event_is_ignored(self) -> None:
with mock.patch("sys.stdin", new=__import__("io").StringIO("{not json")):
self.assertEqual(hook.main(), 0)
class EncodingTests(unittest.TestCase):
def test_output_is_utf8_regardless_of_console_codepage(self) -> None:
"""Chinese output must not depend on the console encoding.
print() would encode with the console codepage (cp950 on a zh-TW
Windows box) and raise UnicodeEncodeError, which — being a ValueError —
the fail-open handler swallows, silently disabling the whole hook.
"""
captured = __import__("io").BytesIO()
stdout = mock.Mock()
stdout.buffer = captured
with mock.patch.object(hook.sys, "stdout", stdout):
hook.emit({"text": "語意召回"})
self.assertEqual(json.loads(captured.getvalue().decode("utf-8"))["text"], "語意召回")
class TranscriptTests(unittest.TestCase):
def test_reads_the_last_assistant_message(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
path = Path(tmp) / "t.jsonl"
path.write_text("\n".join([
json.dumps({"message": {"role": "user", "content": "q"}}),
json.dumps({"message": {"role": "assistant", "content": "first"}}),
json.dumps({"message": {"role": "assistant", "content": [{"type": "text", "text": "final"}]}}),
]), encoding="utf-8")
self.assertEqual(hook.last_assistant_message(str(path)), "final")
def test_missing_transcript_is_not_an_error(self) -> None:
self.assertEqual(hook.last_assistant_message("/no/such/file"), "")
class RenderTests(unittest.TestCase):
def test_delivery_distinguishes_quiet_from_degraded_without_injecting_unknown_claims(self) -> None:
for state in ("ready", "degraded"):
for candidates in ([], [{"id": "unverified", "summary": "UNVERIFIED_CLAIM"}]):
packet = {"records": candidates, "quality": {"status": state},
"delivery": {"records": [], "status": "ready"}}
with self.subTest(state=state, candidates=bool(candidates)), \
tempfile.TemporaryDirectory() as tmp, \
mock.patch.object(hook, "STATE_DIR", Path(tmp)), \
mock.patch.object(hook, "call", side_effect=[{"session_id": "fixture"}, {"retrieval": packet}]), \
mock.patch.object(hook, "emit") as emit:
hook.on_prompt({"session_id": "fixture", "prompt": "Prepare the release"}, {"source": "test"})
if state == "ready":
emit.assert_not_called()
else:
context = emit.call_args.args[0]["hookSpecificOutput"]["additionalContext"]
self.assertIn("degraded", context)
self.assertNotIn("UNVERIFIED_CLAIM", context)
def test_prompt_event_injects_conversation_evidence_and_provenance(self) -> None:
record = {"record_type": "conversation_trace", "trace_id": "trace:fixture",
"query": "角色參考圖如何分工？", "answer": "人物、場景、構圖使用各自的參考圖。"}
with tempfile.TemporaryDirectory() as tmp, \
mock.patch.object(hook, "STATE_DIR", Path(tmp)), \
mock.patch.object(hook, "call", side_effect=[
{"session_id": "session-fixture"}, {"retrieval": {"records": [record]}}]), \
mock.patch.object(hook, "emit") as emit:
hook.on_prompt({"session_id": "session-fixture", "prompt": "角色參考圖如何分工？"}, {"source": "test"})
output = emit.call_args.args[0]["hookSpecificOutput"]
self.assertEqual(output["hookEventName"], "UserPromptSubmit")
for text in (record["query"], record["answer"], record["trace_id"], "可能已經過時"):
self.assertIn(text, output["additionalContext"])
def test_recalled_knowledge_is_labelled_as_history(self) -> None:
rendered = hook.render([{"record_type": "knowledge_card", "title": "T", "summary": "S", "source_url": "u"}])
self.assertIn("xkb_recalled_knowledge", rendered)
self.assertIn("可能已經過時", rendered)
self.assertIn("T", rendered)
def test_complete_delivery_keeps_late_conditions_and_budget(self) -> None:
body = 'Step details. ' * 55 + 'Do not proceed unless the review is approved.'
rendered = hook.render([{'title': 'Procedure', 'summary': body, 'source_url': 'source'}])
self.assertIn(body, rendered)
self.assertTrue(rendered.endswith('</xkb_recalled_knowledge>'))
records = [{'title': 'Too long', 'summary': 'x' * (hook.MAX_CONTEXT_CHARS + 1)},
{'title': 'Useful', 'summary': body}]
rendered = hook.render(records)
self.assertIn(body, rendered)
self.assertNotIn('Too long', rendered)
self.assertLessEqual(len(rendered), hook.MAX_CONTEXT_CHARS)
self.assertEqual(hook.render(records[:1]), '')
class InstallerTests(unittest.TestCase):
def setUp(self) -> None:
self.tmp = tempfile.TemporaryDirectory()
self.home = Path(self.tmp.name)
patcher = mock.patch.object(installer, "agent_home", return_value=self.home)
patcher.start()
self.addCleanup(patcher.stop)
self.config = self.home / "hook-config.json"
config_patcher = mock.patch.object(installer, "HOOK_CONFIG", self.config)
config_patcher.start()
self.addCleanup(config_patcher.stop)
self.addCleanup(self.tmp.cleanup)
def settings(self) -> dict:
return json.loads((self.home / "settings.json").read_text(encoding="utf-8"))
def test_install_wires_both_events(self) -> None:
installer.install("http://127.0.0.1:18972", "", "claude-code")
hooks = self.settings()["hooks"]
for event in installer.EVENTS:
commands = [h["command"] for g in hooks[event] for h in g["hooks"]]
self.assertTrue(any(installer.HOOK_SCRIPT.name in c for c in commands))
def test_reinstall_replaces_rather_than_duplicates(self) -> None:
installer.install("http://a", "", "claude-code")
installer.install("http://b", "", "claude-code")
hooks = self.settings()["hooks"]["UserPromptSubmit"]
ours = [h for g in hooks for h in g["hooks"] if installer.HOOK_SCRIPT.name in h["command"]]
self.assertEqual(len(ours), 1)
self.assertEqual(json.loads(self.config.read_text(encoding="utf-8"))["url"], "http://b")
def test_existing_unrelated_hooks_are_preserved(self) -> None:
(self.home / "settings.json").write_text(json.dumps({
"hooks": {"UserPromptSubmit": [{"matcher": "", "hooks": [{"type": "command", "command": "other-tool"}]}]},
"model": "keep-me",
}), encoding="utf-8")
installer.install("http://127.0.0.1:18972", "", "claude-code")
settings = self.settings()
self.assertEqual(settings["model"], "keep-me")
commands = [h["command"] for g in settings["hooks"]["UserPromptSubmit"] for h in g["hooks"]]
self.assertIn("other-tool", commands)
def test_uninstall_removes_only_ours(self) -> None:
(self.home / "settings.json").write_text(json.dumps({
"hooks": {"Stop": [{"matcher": "", "hooks": [{"type": "command", "command": "other-tool"}]}]},
}), encoding="utf-8")
installer.install("http://127.0.0.1:18972", "", "claude-code")
installer.uninstall()
settings = self.settings()
commands = [h["command"] for g in settings.get("hooks", {}).get("Stop", []) for h in g["hooks"]]
self.assertEqual(commands, ["other-tool"])
self.assertNotIn("UserPromptSubmit", settings.get("hooks", {}))
self.assertFalse(self.config.exists())
def test_corrupt_settings_are_never_overwritten(self) -> None:
(self.home / "settings.json").write_text("{not json", encoding="utf-8")
with self.assertRaises(SystemExit):
installer.install("http://127.0.0.1:18972", "", "claude-code")
if __name__ == "__main__":
unittest.main()