tanmay11k
6d73f74e0b
Major changes: - Config-driven adapters: all channels (Slack, Discord, Telegram, WebChat, Webhooks) controlled via config.json with enabled flags and token auto-detection, no CLI flags required - Runtime engine field: runtime.engine selects opencode/claude from config - Interactive install script: 8-phase setup wizard with AI runtime detection/installation, token setup, identity file personalization (personality presets), aetheel CLI command, background service (launchd/systemd) - Live runtime switching: /engine, /model, /provider commands hot-swap the AI runtime from chat without restart, changes persisted to config.json - Usage tracking: per-request cost extraction from Claude Code JSON output, cumulative stats via /usage command - Auto-failover: rate limit detection on both runtimes, automatic switch to other engine on quota errors with user notification - Chat commands work without / prefix (Slack intercepts / in channels), commands: engine, model, provider, config, usage, reload, cron, subagents, status, help - /config set for editing config.json from chat with dotted key notation - Security audit saved to docs/security-audit.md - Full command reference in docs/commands.md - Future changes doc with NanoClaw agent teams analysis - Logo added to README and WebChat UI - README fully rewritten with all features documented
173 lines
5.7 KiB
Python
173 lines
5.7 KiB
Python
"""
|
|
Tests for the Hook system (HookManager, HookEvent).
|
|
"""
|
|
|
|
import os
|
|
|
|
import pytest
|
|
|
|
from hooks.hooks import HookEvent, HookEntry, HookManager
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Fixtures
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.fixture
|
|
def workspace(tmp_path):
|
|
return str(tmp_path)
|
|
|
|
|
|
@pytest.fixture
|
|
def manager(workspace):
|
|
return HookManager(workspace_dir=workspace)
|
|
|
|
|
|
def _create_hook(workspace, name, events, enabled=True, handler_code=None):
|
|
"""Helper to create a hook directory with HOOK.md and optional handler.py."""
|
|
hook_dir = os.path.join(workspace, "hooks", name)
|
|
os.makedirs(hook_dir, exist_ok=True)
|
|
|
|
enabled_str = "true" if enabled else "false"
|
|
events_str = "[" + ", ".join(events) + "]"
|
|
hook_md = f"""---
|
|
name: {name}
|
|
description: Test hook {name}
|
|
events: {events_str}
|
|
enabled: {enabled_str}
|
|
---
|
|
# {name}
|
|
Test hook documentation.
|
|
"""
|
|
with open(os.path.join(hook_dir, "HOOK.md"), "w") as f:
|
|
f.write(hook_md)
|
|
|
|
if handler_code:
|
|
with open(os.path.join(hook_dir, "handler.py"), "w") as f:
|
|
f.write(handler_code)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tests: HookEvent
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestHookEvent:
|
|
def test_event_key(self):
|
|
event = HookEvent(type="gateway", action="startup")
|
|
assert event.event_key == "gateway:startup"
|
|
|
|
def test_messages_default_empty(self):
|
|
event = HookEvent(type="command", action="new")
|
|
assert event.messages == []
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tests: HookManager discovery
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestHookManagerDiscovery:
|
|
def test_discover_empty_workspace(self, manager):
|
|
hooks = manager.discover()
|
|
assert hooks == []
|
|
|
|
def test_discover_finds_hooks(self, workspace, manager):
|
|
_create_hook(workspace, "my-hook", ["gateway:startup"])
|
|
hooks = manager.discover()
|
|
assert len(hooks) == 1
|
|
assert hooks[0].name == "my-hook"
|
|
assert hooks[0].events == ["gateway:startup"]
|
|
|
|
def test_discover_skips_disabled(self, workspace, manager):
|
|
_create_hook(workspace, "disabled-hook", ["gateway:startup"], enabled=False)
|
|
hooks = manager.discover()
|
|
assert len(hooks) == 0
|
|
|
|
def test_discover_multiple_hooks(self, workspace, manager):
|
|
_create_hook(workspace, "hook-a", ["gateway:startup"])
|
|
_create_hook(workspace, "hook-b", ["command:new"])
|
|
hooks = manager.discover()
|
|
assert len(hooks) == 2
|
|
|
|
def test_discover_multiple_events(self, workspace, manager):
|
|
_create_hook(workspace, "multi", ["gateway:startup", "command:new"])
|
|
hooks = manager.discover()
|
|
assert hooks[0].events == ["gateway:startup", "command:new"]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tests: HookManager trigger
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestHookManagerTrigger:
|
|
def test_trigger_calls_handler(self, workspace, manager):
|
|
handler_code = """
|
|
results = []
|
|
def handle(event):
|
|
results.append(event.event_key)
|
|
"""
|
|
_create_hook(workspace, "test-hook", ["gateway:startup"], handler_code=handler_code)
|
|
manager.discover()
|
|
event = HookEvent(type="gateway", action="startup")
|
|
manager.trigger(event)
|
|
# Handler was called without error — that's the test
|
|
|
|
def test_trigger_no_matching_hooks(self, manager):
|
|
event = HookEvent(type="gateway", action="startup")
|
|
messages = manager.trigger(event)
|
|
assert messages == []
|
|
|
|
def test_trigger_handler_error_doesnt_crash(self, workspace, manager):
|
|
handler_code = """
|
|
def handle(event):
|
|
raise RuntimeError("boom")
|
|
"""
|
|
_create_hook(workspace, "bad-hook", ["gateway:startup"], handler_code=handler_code)
|
|
manager.discover()
|
|
event = HookEvent(type="gateway", action="startup")
|
|
# Should not raise
|
|
manager.trigger(event)
|
|
|
|
def test_trigger_messages(self, workspace, manager):
|
|
handler_code = """
|
|
def handle(event):
|
|
event.messages.append("hello from hook")
|
|
"""
|
|
_create_hook(workspace, "msg-hook", ["gateway:startup"], handler_code=handler_code)
|
|
manager.discover()
|
|
event = HookEvent(type="gateway", action="startup")
|
|
messages = manager.trigger(event)
|
|
assert "hello from hook" in messages
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tests: Programmatic hooks
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestProgrammaticHooks:
|
|
def test_register_and_trigger(self, manager):
|
|
received = []
|
|
manager.register("gateway:startup", lambda e: received.append(e.event_key))
|
|
event = HookEvent(type="gateway", action="startup")
|
|
manager.trigger(event)
|
|
assert received == ["gateway:startup"]
|
|
|
|
def test_unregister(self, manager):
|
|
received = []
|
|
handler = lambda e: received.append(e.event_key)
|
|
manager.register("gateway:startup", handler)
|
|
manager.unregister("gateway:startup", handler)
|
|
manager.trigger(HookEvent(type="gateway", action="startup"))
|
|
assert received == []
|
|
|
|
def test_type_level_handler(self, manager):
|
|
received = []
|
|
manager.register("gateway", lambda e: received.append(e.action))
|
|
manager.trigger(HookEvent(type="gateway", action="startup"))
|
|
manager.trigger(HookEvent(type="gateway", action="shutdown"))
|
|
assert received == ["startup", "shutdown"]
|