Deshraj YadavandClaude Fable 5
e11d2a4a06
Claude Code auto-discovers a plugin's hooks at <plugin>/hooks/hooks.json. The generator was writing to hooks/generated/, so the manifest existed but was never loaded and no hook would have fired on install. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
142 lines
5.0 KiB
Python
142 lines
5.0 KiB
Python
"""CLI-level behavior: what the editor's hooks actually invoke."""
|
|
|
|
import io
|
|
import json
|
|
import sys
|
|
|
|
import pytest
|
|
|
|
from mem0_agent import cli
|
|
from mem0_agent.settings import SessionState, Settings
|
|
|
|
|
|
class Args:
|
|
def __init__(self, **kw):
|
|
self.session_id = "sess-cli"
|
|
for k, v in kw.items():
|
|
setattr(self, k, v)
|
|
|
|
|
|
class FakeCtx:
|
|
def __init__(self, tmp_path, ready=True):
|
|
self.api = None
|
|
self.settings = Settings(data={"capture": "balanced", "retrieval": "balanced"},
|
|
path=tmp_path / "s.json")
|
|
self.state = SessionState("sess-cli", root=tmp_path / "sessions")
|
|
self.user_id, self.app_id = "dev", "acme-repo"
|
|
self.session_id, self.branch = "sess-cli", "main"
|
|
self.ready, self.reason = ready, "" if ready else "no API key"
|
|
|
|
def provenance(self, mtype):
|
|
return {"type": mtype}
|
|
|
|
def log(self, *a, **k):
|
|
pass
|
|
|
|
|
|
@pytest.fixture
|
|
def ctx(tmp_path, monkeypatch):
|
|
c = FakeCtx(tmp_path)
|
|
monkeypatch.setattr(cli, "build", lambda *a, **k: c)
|
|
return c
|
|
|
|
|
|
def run(fn, args, stdin=""):
|
|
"""Invoke a command with a controlled stdin/stdout."""
|
|
old_in, old_out = sys.stdin, sys.stdout
|
|
sys.stdin = io.StringIO(stdin)
|
|
sys.stdout = out = io.StringIO()
|
|
try:
|
|
code = fn(args)
|
|
finally:
|
|
sys.stdin, sys.stdout = old_in, old_out
|
|
return code, out.getvalue()
|
|
|
|
|
|
def test_hook_input_tolerates_garbage(monkeypatch):
|
|
monkeypatch.setattr(sys, "stdin", io.StringIO("not json at all"))
|
|
assert cli.hook_input() == {}
|
|
|
|
|
|
def test_hook_input_parses_payload(monkeypatch):
|
|
monkeypatch.setattr(sys, "stdin", io.StringIO(json.dumps({"session_id": "abc"})))
|
|
assert cli.hook_input()["session_id"] == "abc"
|
|
|
|
|
|
def test_queued_context_is_delivered_once(ctx):
|
|
"""The detached error assist queues; the next prompt hook drains it exactly once."""
|
|
cli.queue_context(ctx, "<mem0-recall>- [insight] restart pgbouncer</mem0-recall>")
|
|
first = cli.drain_context(ctx)
|
|
second = cli.drain_context(ctx)
|
|
assert "pgbouncer" in first
|
|
assert second == "", "a queued block must not be delivered twice"
|
|
|
|
|
|
def test_observe_emits_queued_recall(ctx, tmp_path):
|
|
cli.queue_context(ctx, "<mem0-recall>- [insight] the fix</mem0-recall>")
|
|
code, out = run(cli.cmd_observe, Args(transcript=None, source="prompt"),
|
|
stdin=json.dumps({"session_id": "sess-cli", "prompt": "why did that fail?"}))
|
|
assert code == 0
|
|
assert "the fix" in out
|
|
|
|
|
|
def test_commands_are_noops_without_credentials(tmp_path, monkeypatch):
|
|
c = FakeCtx(tmp_path, ready=False)
|
|
monkeypatch.setattr(cli, "build", lambda *a, **k: c)
|
|
for fn, args in [
|
|
(cli.cmd_context, Args(force=False, stats=False)),
|
|
(cli.cmd_observe, Args(transcript=None, source="prompt")),
|
|
(cli.cmd_flush, Args(transcript=None, reason="stop", json=False)),
|
|
(cli.cmd_assist_error, Args(text="boom", emit=True)),
|
|
]:
|
|
code, out = run(fn, args)
|
|
assert code == 0, "a hook must never exit non-zero"
|
|
assert out == ""
|
|
|
|
|
|
def test_main_never_propagates_an_exception(monkeypatch):
|
|
def explode(*a, **k):
|
|
raise RuntimeError("kaboom")
|
|
|
|
monkeypatch.setattr(cli, "cmd_health", explode)
|
|
assert cli.main(["health"]) == 0, "a crash in memory must not break the session"
|
|
|
|
|
|
def test_config_reports_and_updates(tmp_path, monkeypatch):
|
|
settings = Settings(data=dict(capture="balanced", retrieval="balanced",
|
|
memory_mode="dual"), path=tmp_path / "s.json")
|
|
monkeypatch.setattr(cli.Settings, "load", classmethod(lambda cls, *a, **k: settings))
|
|
code, out = run(cli.cmd_config, Args(capture="conservative", retrieval=None, mode=None))
|
|
assert code == 0
|
|
assert settings.get("capture") == "conservative"
|
|
assert "capture = conservative" in out
|
|
|
|
|
|
def test_every_subcommand_is_registered():
|
|
"""The generated hook manifest invokes these by name; a rename must fail loudly."""
|
|
for cmd in ("setup", "onboard", "context", "observe", "flush", "assist-error",
|
|
"remember", "forget", "maintain", "health", "stats", "config"):
|
|
with pytest.raises(SystemExit) as e:
|
|
cli.main([cmd, "--help"])
|
|
assert e.value.code == 0
|
|
|
|
|
|
def test_hook_manifest_commands_all_exist():
|
|
"""Guards against the manifest and the CLI drifting apart."""
|
|
import pathlib
|
|
|
|
manifest = pathlib.Path(__file__).resolve().parents[1] / "hooks/hooks.json"
|
|
data = json.loads(manifest.read_text())
|
|
known = {"setup", "onboard", "context", "observe", "flush", "assist-error",
|
|
"remember", "forget", "maintain", "health", "stats", "config"}
|
|
found = 0
|
|
for entries in data["hooks"].values():
|
|
for entry in entries:
|
|
for hook in entry.get("hooks", []):
|
|
cmd = hook["command"]
|
|
assert "mem0-agent " in cmd
|
|
sub = cmd.split("mem0-agent ", 1)[1].split()[0]
|
|
assert sub in known, f"manifest invokes unknown subcommand {sub!r}"
|
|
found += 1
|
|
assert found >= 6
|