Files
laya-hook/tests/test_hook.py
2026-09-24 14:27:07 +08:00

137 lines
4.3 KiB
Python

import json
from urllib.error import URLError
from laya_router.hook import handle
def _ok_response():
body = {
"level": "complex",
"level_confidence": 0.81,
"needs_plan": 0.9,
"needs_repo": 0.72,
"needs_web": 0.2,
"needs_deep_reasoning": 0.88,
}
class Response:
def __enter__(self):
return self
def __exit__(self, *args):
return False
def read(self):
return json.dumps(body).encode()
return Response()
def test_cursor_is_a_noop():
text = handle({"prompt": "帮我改这个函数"}, {"CURSOR_VERSION": "1.0"}, "claude")
assert text == "{}"
def test_slash_command_is_skipped():
assert handle({"prompt": "/hooks"}, {}, "codex") == "{}"
def test_filesystem_path_is_not_treated_as_slash_command(monkeypatch, tmp_path):
monkeypatch.setenv("LAYA_ROUTER_HOME", str(tmp_path))
monkeypatch.setattr("laya_router.hook.call_laya", lambda settings, body: _payload())
text = handle({"prompt": "/Users/richard/proj 里的入口在哪"}, {}, "codex")
assert json.loads(text)["hookSpecificOutput"]["hookEventName"] == "UserPromptSubmit"
def test_fail_open_when_server_is_down(monkeypatch, tmp_path):
monkeypatch.setenv("LAYA_ROUTER_HOME", str(tmp_path))
def boom(settings, body):
raise URLError("connection refused")
monkeypatch.setattr("laya_router.hook.call_laya", boom)
assert handle({"prompt": "解释这个报错"}, {}, "claude") == "{}"
assert (tmp_path / "logs" / "hook.log").is_file()
def test_success_output(monkeypatch, tmp_path):
monkeypatch.setenv("LAYA_ROUTER_HOME", str(tmp_path))
monkeypatch.setattr(
"laya_router.hook.call_laya",
lambda settings, body: {
"level": "simple",
"level_confidence": 0.66,
"needs_plan": 0.1,
"needs_repo": 0.2,
"needs_web": 0.05,
"needs_deep_reasoning": 0.08,
},
)
payload = handle(
{"prompt": "把这个变量改名", "session_id": "s1", "cwd": "/tmp/repo"},
{},
"codex",
)
parsed = json.loads(payload)
context = parsed["hookSpecificOutput"]["additionalContext"]
assert "Task complexity: simple" in context
assert "uncalibrated" in context
assert "hint only" in context
def test_pre_tool_deny_and_post_tool_context(monkeypatch, tmp_path):
monkeypatch.setenv("LAYA_ROUTER_HOME", str(tmp_path))
def respond(settings, path, body, timeout):
assert path == "/event"
assert timeout == settings.event_timeout
if body["event"] == "PreToolUse":
return {"action": "deny", "reason": "do not delete", "context": ""}
return {"action": "inject", "context": "replan", "reason": ""}
monkeypatch.setattr("laya_router.hook.post_json", respond)
denied = json.loads(
handle(
{"hook_event_name": "PreToolUse", "tool_name": "Bash", "tool_input": {"command": "rm -rf /"}},
{},
"claude",
)
)
assert denied["hookSpecificOutput"]["permissionDecision"] == "deny"
posted = json.loads(
handle(
{"hook_event_name": "PostToolUse", "tool_name": "Bash", "tool_input": {"command": "pytest"}},
{},
"codex",
)
)
assert posted["hookSpecificOutput"]["hookEventName"] == "PostToolUse"
assert posted["hookSpecificOutput"]["additionalContext"] == "replan"
def test_stop_block_and_fail_open(monkeypatch, tmp_path):
monkeypatch.setenv("LAYA_ROUTER_HOME", str(tmp_path))
monkeypatch.setattr(
"laya_router.hook.post_json",
lambda settings, path, body, timeout: {"action": "block", "reason": "run pytest"},
)
blocked = json.loads(handle({"hook_event_name": "Stop", "stop_hook_active": False}, {}, "codex"))
assert blocked == {"decision": "block", "reason": "run pytest"}
def boom(settings, path, body, timeout):
raise URLError("connection refused")
monkeypatch.setattr("laya_router.hook.post_json", boom)
assert handle({"hook_event_name": "Stop"}, {}, "claude") == "{}"
def _payload():
return {
"level": "normal",
"level_confidence": 0.5,
"needs_plan": 0.4,
"needs_repo": 0.4,
"needs_web": 0.1,
"needs_deep_reasoning": 0.3,
}