Pheby
6be1601ccd
Deliverables now inherit the active assistant draft message ID so clients can render them beside the reply that produced them.
1008 lines
44 KiB
Python
1008 lines
44 KiB
Python
"""Pheby plugin test suite.
|
|
|
|
Run with the Hermes venv's pytest from the repo root:
|
|
|
|
/opt/hermes/.venv/bin/python -m pytest tests/ -o 'addopts=' -q
|
|
|
|
All tests use fakes for the Hermes gateway — no LLM calls, no network beyond
|
|
localhost, no real HERMES_HOME writes (HERMES_HOME is redirected to a tmp dir
|
|
by ``conftest.py``).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import json
|
|
import threading
|
|
import time
|
|
from pathlib import Path
|
|
from typing import Any, Dict, List, Optional
|
|
|
|
import pytest
|
|
|
|
# Make the plugin package importable regardless of install layout.
|
|
import sys
|
|
_PLUGIN_DIR = Path(__file__).resolve().parent.parent / "plugin"
|
|
if str(_PLUGIN_DIR) not in sys.path:
|
|
sys.path.insert(0, str(_PLUGIN_DIR))
|
|
|
|
from pheby import protocol as proto # noqa: E402
|
|
from pheby.attachments import AttachmentStore, constant_time_equals # noqa: E402
|
|
from pheby.adapter import PhebyAdapter # noqa: E402
|
|
from pheby.config import load_config # noqa: E402
|
|
from pheby.conversations import ConversationRouter # noqa: E402
|
|
from pheby.server import PhebyServer # noqa: E402
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════════
|
|
# Fakes
|
|
# ═══════════════════════════════════════════════════════════════════════════
|
|
class FakeWS:
|
|
"""Minimal WebSocketResponse stand-in for server-loop tests."""
|
|
|
|
def __init__(self):
|
|
self.sent: List[str] = []
|
|
self.inbox: "asyncio.Queue[str]" = asyncio.Queue()
|
|
self.closed = False
|
|
self.close_code: Optional[int] = None
|
|
|
|
async def send_str(self, data: str) -> None:
|
|
if self.closed:
|
|
raise ConnectionError("closed")
|
|
self.sent.append(data)
|
|
|
|
async def receive(self, timeout: Optional[float] = None):
|
|
class _Msg:
|
|
def __init__(self, data: str):
|
|
self.type = "text"
|
|
self.data = data
|
|
|
|
try:
|
|
return _Msg(await asyncio.wait_for(self.inbox.get(),
|
|
timeout=timeout))
|
|
except asyncio.TimeoutError:
|
|
raise
|
|
|
|
async def close(self, code: Optional[int] = None, message=None):
|
|
self.closed = True
|
|
self.close_code = code
|
|
|
|
def events(self) -> List[Dict[str, Any]]:
|
|
out = []
|
|
for raw in self.sent:
|
|
try:
|
|
out.append(json.loads(raw))
|
|
except json.JSONDecodeError:
|
|
pass
|
|
return out
|
|
|
|
|
|
class FakeClientConnection:
|
|
"""Wraps FakeWS with the ClientConnection interface the server expects."""
|
|
|
|
def __init__(self):
|
|
self.ws = FakeWS()
|
|
self.conn_id = "test-conn"
|
|
self.authenticated = False
|
|
self.protocol_version = None
|
|
self.connected_at = time.time()
|
|
self.closed = False
|
|
self._send_lock = asyncio.Lock()
|
|
|
|
async def send_json(self, payload: Dict[str, Any]) -> bool:
|
|
if self.closed:
|
|
return False
|
|
try:
|
|
async with self._send_lock:
|
|
await self.ws.send_str(proto.encode_message(payload))
|
|
return True
|
|
except (ConnectionError, RuntimeError, asyncio.CancelledError):
|
|
self.closed = True
|
|
return False
|
|
|
|
|
|
class FakeAdapter:
|
|
"""Adapter stand-in: enough surface for bridge tests."""
|
|
|
|
def __init__(self):
|
|
self.platform = type("P", (), {"value": "pheby"})()
|
|
self.gateway_runner = None
|
|
self._active_sessions: Dict[str, Any] = {}
|
|
self.handled: List[Any] = []
|
|
|
|
def build_source(self, **kwargs):
|
|
from gateway.session import SessionSource # real Hermes type
|
|
return SessionSource(
|
|
platform=self.platform, chat_id=kwargs.get("chat_id", "x"),
|
|
chat_type="dm", user_id="pheby-client")
|
|
|
|
async def handle_message(self, event) -> None:
|
|
self.handled.append(event)
|
|
|
|
async def interrupt_session_activity(self, session_key, chat_id,
|
|
metadata=None):
|
|
self.interrupted = (session_key, chat_id)
|
|
|
|
|
|
class FakeRunner:
|
|
"""Gateway runner stand-in for session-key + interrupt tests."""
|
|
|
|
def __init__(self):
|
|
self.session_store = None
|
|
self._session_db = None
|
|
self._running_agents: Dict[str, Any] = {}
|
|
self.generations: Dict[str, int] = {}
|
|
|
|
def _generate_session_key(self, source):
|
|
return f"agent:main:pheby:dm:{source.chat_id}"
|
|
|
|
def _invalidate_session_run_generation(self, session_key, reason=""):
|
|
self.generations[session_key] = \
|
|
self.generations.get(session_key, 0) + 1
|
|
|
|
|
|
class FakeAgent:
|
|
def __init__(self):
|
|
self.interrupts: List[str] = []
|
|
|
|
def interrupt(self, message=None):
|
|
self.interrupts.append(message or "")
|
|
|
|
|
|
def make_server(tmp_path: Path, **overrides) -> PhebyServer:
|
|
cfg = load_config({
|
|
"secret": "test-secret-abc123",
|
|
"port": overrides.pop("port", 0), # 0 unused in handler tests
|
|
**overrides,
|
|
})
|
|
cfg.secret = overrides.get("secret", cfg.secret or "test-secret-abc123")
|
|
root = Path(tmp_path) / "attachments"
|
|
server = PhebyServer(cfg, adapter=FakeAdapter())
|
|
server.store = AttachmentStore(root=root, retention_days=7)
|
|
hermes_bridge_set(server)
|
|
return server
|
|
|
|
|
|
def hermes_bridge_set(server: PhebyServer) -> None:
|
|
from pheby import hermes_bridge
|
|
hermes_bridge.set_server(server)
|
|
if server.adapter is not None:
|
|
hermes_bridge.set_adapter(server.adapter)
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════════
|
|
# Protocol serialization
|
|
# ═══════════════════════════════════════════════════════════════════════════
|
|
class TestProtocol:
|
|
def test_roundtrip(self):
|
|
msg = {"type": proto.C_CHAT_SEND, "conversation_id": "abc",
|
|
"text": "héllo 🐱", "request_id": "r1"}
|
|
data = proto.encode_message(msg)
|
|
parsed, err = proto.decode_message(data)
|
|
assert err is None and parsed == msg
|
|
|
|
def test_invalid_json_rejected(self):
|
|
for bad in ("{not json", "[]", '"str"', "42", '{"no_type": 1}', ""):
|
|
parsed, err = proto.decode_message(bad)
|
|
assert parsed is None and err == proto.ERR_INVALID_JSON
|
|
|
|
def test_error_event_shape(self):
|
|
ev = proto.error_event(proto.ERR_BAD_REQUEST, "boom", request_id="r9")
|
|
assert ev["type"] == proto.S_ERROR
|
|
assert ev["error"]["code"] == proto.ERR_BAD_REQUEST
|
|
assert ev["request_id"] == "r9"
|
|
|
|
def test_attachment_id_format(self):
|
|
aid = proto.new_id()
|
|
assert len(aid) == 32 and proto.is_valid_attachment_id(aid)
|
|
assert not proto.is_valid_attachment_id("../etc/passwd")
|
|
assert not proto.is_valid_attachment_id("")
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════════
|
|
# Authentication
|
|
# ═══════════════════════════════════════════════════════════════════════════
|
|
class TestAuth:
|
|
@pytest.mark.asyncio
|
|
async def test_hello_success(self, tmp_path):
|
|
server = make_server(tmp_path)
|
|
client = FakeClientConnection()
|
|
client.ws.inbox.put_nowait(proto.encode_message({
|
|
"type": proto.C_HELLO, "secret": "test-secret-abc123",
|
|
"protocol_version": proto.PROTOCOL_VERSION}))
|
|
ok = await server._authenticate(client, "peer1")
|
|
assert ok and client.authenticated
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_hello_wrong_secret(self, tmp_path):
|
|
server = make_server(tmp_path)
|
|
client = FakeClientConnection()
|
|
bad_hello = proto.encode_message(
|
|
{"type": proto.C_HELLO, "secret": "wrong"})
|
|
client.ws.inbox.put_nowait(bad_hello)
|
|
ok = await server._authenticate(client, "peer2")
|
|
assert not ok and not client.authenticated
|
|
# 5 failures → lockout (each attempt needs its own hello frame)
|
|
for _ in range(proto.AUTH_FAILURE_THRESHOLD - 1):
|
|
client.ws.inbox.put_nowait(bad_hello)
|
|
await server._authenticate(client, "peer2")
|
|
assert server._is_locked_out("peer2")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_first_message_not_hello(self, tmp_path):
|
|
server = make_server(tmp_path)
|
|
client = FakeClientConnection()
|
|
client.ws.inbox.put_nowait(proto.encode_message(
|
|
{"type": proto.C_PING}))
|
|
ok = await server._authenticate(client, "peer3")
|
|
assert not ok
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_version_mismatch_refused(self, tmp_path):
|
|
server = make_server(tmp_path)
|
|
client = FakeClientConnection()
|
|
client.ws.inbox.put_nowait(proto.encode_message({
|
|
"type": proto.C_HELLO, "secret": "test-secret-abc123",
|
|
"protocol_version": 99}))
|
|
ok = await server._authenticate(client, "peer4")
|
|
assert not ok and not client.authenticated
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_malformed_version_is_structured_error(self, tmp_path):
|
|
server = make_server(tmp_path)
|
|
client = FakeClientConnection()
|
|
client.ws.inbox.put_nowait(proto.encode_message({
|
|
"type": proto.C_HELLO, "secret": "test-secret-abc123",
|
|
"protocol_version": {"not": "an integer"}}))
|
|
ok = await server._authenticate(client, "peer5")
|
|
assert not ok and not client.authenticated
|
|
error = client.ws.events()[-1]
|
|
assert error["type"] == proto.S_ERROR
|
|
assert error["error"]["code"] == proto.ERR_VERSION_MISMATCH
|
|
|
|
def test_constant_time_equals(self):
|
|
assert constant_time_equals("abc", "abc")
|
|
assert not constant_time_equals("abc", "abd")
|
|
assert not constant_time_equals("abc", "abcd")
|
|
assert not constant_time_equals("", "x")
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════════
|
|
# Conversation operations
|
|
# ═══════════════════════════════════════════════════════════════════════════
|
|
class TestConversations:
|
|
@pytest.mark.asyncio
|
|
async def test_create_list_rename_delete(self, tmp_path):
|
|
server = make_server(tmp_path)
|
|
client = FakeClientConnection()
|
|
client.authenticated = True
|
|
|
|
await server._handle_conversation_create(client, {
|
|
"type": proto.C_CONVERSATION_CREATE, "name": "Project X"}, "r1")
|
|
created = client.ws.events()[-1]
|
|
cid = created["conversation_id"]
|
|
assert created["type"] == proto.S_CONVERSATION_CREATED
|
|
assert created["name"] == "Project X"
|
|
|
|
# list includes it
|
|
await server._handle_conversation_list(client, {
|
|
"type": proto.C_CONVERSATION_LIST}, "r2")
|
|
snap = client.ws.events()[-1]
|
|
assert any(c["conversation_id"] == cid for c in snap["conversations"])
|
|
|
|
# rename
|
|
await server._handle_conversation_rename(client, {
|
|
"type": proto.C_CONVERSATION_RENAME,
|
|
"conversation_id": cid, "name": "Renamed"}, "r3")
|
|
renamed = client.ws.events()[-1]
|
|
assert renamed["type"] == proto.S_CONVERSATION_RENAMED
|
|
assert renamed["name"] == "Renamed"
|
|
|
|
# open (empty history, conversation exists in router)
|
|
await server._handle_conversation_open(client, {
|
|
"type": proto.C_CONVERSATION_OPEN,
|
|
"conversation_id": cid}, "r4")
|
|
hist = client.ws.events()[-1]
|
|
assert hist["type"] == proto.S_CONVERSATION_HISTORY
|
|
assert hist["messages"] == []
|
|
|
|
# delete
|
|
await server._handle_conversation_delete(client, {
|
|
"type": proto.C_CONVERSATION_DELETE,
|
|
"conversation_id": cid}, "r5")
|
|
deleted = client.ws.events()[-1]
|
|
assert deleted["type"] == proto.S_CONVERSATION_DELETED
|
|
|
|
# open after delete → not found
|
|
await server._handle_conversation_open(client, {
|
|
"type": proto.C_CONVERSATION_OPEN,
|
|
"conversation_id": cid}, "r6")
|
|
err = client.ws.events()[-1]
|
|
assert err["type"] == proto.S_ERROR
|
|
assert err["error"]["code"] == proto.ERR_CONVERSATION_NOT_FOUND
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_invalid_id_rejected(self, tmp_path):
|
|
server = make_server(tmp_path)
|
|
client = FakeClientConnection()
|
|
client.authenticated = True
|
|
await server._handle_conversation_open(client, {
|
|
"type": proto.C_CONVERSATION_OPEN,
|
|
"conversation_id": "../../etc"}, "r1")
|
|
ev = client.ws.events()[-1]
|
|
assert ev["error"]["code"] == proto.ERR_BAD_REQUEST
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ids_survive_router_reload(self, tmp_path):
|
|
server = make_server(tmp_path)
|
|
cid = await server.router.new_conversation("Persisted")
|
|
# New router instance (simulates restart) sees the same ID.
|
|
router2 = ConversationRouter()
|
|
assert await router2.get_name(cid) == "Persisted"
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════════
|
|
# Attachments
|
|
# ═══════════════════════════════════════════════════════════════════════════
|
|
class TestAttachments:
|
|
def test_adapter_anchors_deliverable_to_active_assistant_draft(self):
|
|
adapter = object.__new__(PhebyAdapter)
|
|
adapter._drafts = {"conv": {"message_id": "draft-run-1", "text": ""}}
|
|
|
|
assert adapter._active_assistant_message_id("conv") == "draft-run-1"
|
|
assert adapter._active_assistant_message_id("unknown") is None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_register_describe_download_path(self, tmp_path):
|
|
src = Path(tmp_path) / "report.pdf"
|
|
src.write_bytes(b"%PDF-1.4 fake")
|
|
store = AttachmentStore(root=Path(tmp_path) / "att", retention_days=7)
|
|
desc = await store.register_file(str(src), conversation_id="conv1")
|
|
assert desc is not None
|
|
assert desc["filename"] == "report.pdf"
|
|
assert desc["mime_type"] == "application/pdf"
|
|
assert desc["inline_image"] is False
|
|
assert desc["download_path"].startswith("/attachments/")
|
|
# Blob resolves only via the registered ID
|
|
blob = store.resolve_blob(desc["attachment_id"])
|
|
assert blob is not None and blob.read_bytes() == b"%PDF-1.4 fake"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_image_detection(self, tmp_path):
|
|
src = Path(tmp_path) / "pic.png"
|
|
src.write_bytes(b"\x89PNG fake")
|
|
store = AttachmentStore(root=Path(tmp_path) / "att", retention_days=7)
|
|
desc = await store.register_file(str(src), conversation_id="c")
|
|
assert desc["kind"] == "image" and desc["inline_image"] is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_unknown_and_traversal_ids(self, tmp_path):
|
|
store = AttachmentStore(root=Path(tmp_path) / "att", retention_days=7)
|
|
assert store.resolve_blob("f" * 32) is None
|
|
assert store.resolve_blob("../../etc/passwd") is None
|
|
assert store.resolve_blob("../" + "a" * 32) is None
|
|
assert store.resolve_blob("") is None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_seven_day_expiry(self, tmp_path):
|
|
src = Path(tmp_path) / "old.txt"
|
|
src.write_text("expired soon")
|
|
store = AttachmentStore(root=Path(tmp_path) / "att", retention_days=7)
|
|
desc = await store.register_file(str(src), conversation_id="c")
|
|
aid = desc["attachment_id"]
|
|
assert store.resolve_blob(aid) is not None
|
|
# Force age beyond retention.
|
|
store._meta[aid]["created_epoch"] = time.time() - 8 * 86400
|
|
assert store.resolve_blob(aid) is None # expired → unavailable
|
|
removed = await store.cleanup_expired()
|
|
assert removed == 1
|
|
# Blob actually gone from disk; metadata index updated.
|
|
assert store._meta.get(aid) is None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cleanup_never_touches_unrelated_files(self, tmp_path):
|
|
root = Path(tmp_path) / "att"
|
|
store = AttachmentStore(root=root, retention_days=7)
|
|
stranger = root / "blobs" / "zz" / "unrelated.txt"
|
|
stranger.parent.mkdir(parents=True)
|
|
stranger.write_text("keep me")
|
|
await store.cleanup_expired()
|
|
assert stranger.exists() # untouched
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_metadata_survives_restart(self, tmp_path):
|
|
src = Path(tmp_path) / "doc.md"
|
|
src.write_text("# hi")
|
|
root = Path(tmp_path) / "att"
|
|
store1 = AttachmentStore(root=root, retention_days=7)
|
|
desc = await store1.register_file(str(src), conversation_id="c")
|
|
store2 = AttachmentStore(root=root, retention_days=7)
|
|
store2.hydrate_legacy_meta() # no-op for index-file storage
|
|
assert store2.resolve_blob(desc["attachment_id"]) is not None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_missing_source_file(self, tmp_path):
|
|
store = AttachmentStore(root=Path(tmp_path) / "att", retention_days=7)
|
|
desc = await store.register_file(
|
|
str(Path(tmp_path) / "nope.bin"), conversation_id="c")
|
|
assert desc is None
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════════
|
|
# Tool events / approvals / clarifications / cancellation (bridge)
|
|
# ═══════════════════════════════════════════════════════════════════════════
|
|
class TestBridge:
|
|
@pytest.mark.asyncio
|
|
async def test_tool_start_event_is_structured_not_text(self, tmp_path,
|
|
monkeypatch):
|
|
server = make_server(tmp_path)
|
|
client = FakeClientConnection()
|
|
client.authenticated = True
|
|
server._clients["t"] = client
|
|
|
|
# Use the real PhebyAdapter for the tool-event path (FakeAdapter has
|
|
# no format_tool_event; the real one is what we're testing).
|
|
from pheby.adapter import PhebyAdapter
|
|
from gateway.config import PlatformConfig
|
|
real = PhebyAdapter(PlatformConfig(
|
|
enabled=True, extra={"secret": "test-secret-abc123"}))
|
|
real._pcfg = server.config
|
|
real._server = server
|
|
real._loop = asyncio.get_running_loop()
|
|
real._active_sessions = {"agent:main:pheby:dm:conv1": asyncio.Event()}
|
|
monkeypatch.setattr(real, "_conversation_for_session_id",
|
|
lambda _sid: "conv1")
|
|
|
|
from gateway.stream_events import ToolCallChunk
|
|
marker = real.format_tool_event(
|
|
ToolCallChunk(tool_name="web_search", preview="cats",
|
|
args={"query": "cats"}, index=0),
|
|
mode="all")
|
|
assert marker is None # never rendered as chat text
|
|
# The display event is intentionally eaten. The authoritative hook
|
|
# carries the real Hermes session and tool-call IDs.
|
|
real.on_pre_tool_call(
|
|
session_id="session-1", tool_name="web_search",
|
|
tool_call_id="call-1", args={"query": "cats"})
|
|
await asyncio.sleep(0)
|
|
events = client.ws.events()
|
|
tool_events = [e for e in events if e["type"] == proto.S_TOOL_EVENT]
|
|
assert len(tool_events) == 1
|
|
ev = tool_events[0]
|
|
assert ev["tool_name"] == "web_search"
|
|
assert ev["status"] == "running"
|
|
assert ev["tool_call_id"] == "call-1"
|
|
assert ev["conversation_id"] == "conv1"
|
|
# No fake prose leaked into a message event
|
|
assert not any(e.get("type") == proto.S_MESSAGE_COMPLETE
|
|
for e in events)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_post_tool_call_completion(self, tmp_path, monkeypatch):
|
|
server = make_server(tmp_path)
|
|
client = FakeClientConnection()
|
|
client.authenticated = True
|
|
server._clients["t"] = client
|
|
|
|
from pheby.adapter import PhebyAdapter
|
|
from gateway.config import PlatformConfig
|
|
real = PhebyAdapter(PlatformConfig(
|
|
enabled=True, extra={"secret": "test-secret-abc123"}))
|
|
real._server = server
|
|
real._loop = asyncio.get_running_loop()
|
|
real._active_sessions = {"agent:main:pheby:dm:conv1": asyncio.Event()}
|
|
monkeypatch.setattr(real, "_conversation_for_session_id",
|
|
lambda _sid: "conv1")
|
|
|
|
real.on_post_tool_call(
|
|
session_id="session-1", tool_name="terminal", tool_call_id="call_9",
|
|
status="ok", duration_ms=1234)
|
|
await asyncio.sleep(0) # let ensure_future run
|
|
ev = [e for e in client.ws.events()
|
|
if e["type"] == proto.S_TOOL_EVENT][-1]
|
|
assert ev["tool_call_id"] == "call_9"
|
|
assert ev["status"] == "completed"
|
|
assert ev["duration_ms"] == 1234
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_approval_push_and_resolve_roundtrip(self, tmp_path,
|
|
monkeypatch):
|
|
server = make_server(tmp_path)
|
|
client = FakeClientConnection()
|
|
client.authenticated = True
|
|
server._clients["t"] = client
|
|
|
|
from pheby import hermes_bridge
|
|
await hermes_bridge.push_approval(
|
|
{"command": "rm -rf /tmp/x", "description": "Destructive command",
|
|
"allow_permanent": True, "allow_session": True},
|
|
session_key="agent:main:pheby:dm:conv1")
|
|
req = [e for e in client.ws.events()
|
|
if e["type"] == proto.S_APPROVAL_REQUEST][-1]
|
|
assert req["choices"] == ["once", "session", "always", "deny"]
|
|
assert req["description"] == "Destructive command"
|
|
|
|
from pheby import hermes_bridge as hb
|
|
import tools.approval
|
|
monkeypatch.setattr(tools.approval, "resolve_gateway_approval",
|
|
lambda *args, **kwargs: 1)
|
|
ok = await hb.resolve_approval(req["approval_id"], "deny", None)
|
|
assert ok is True
|
|
# Double resolve → not found
|
|
ok2 = await hb.resolve_approval(req["approval_id"], "once", None)
|
|
assert ok2 is False
|
|
|
|
# Client-facing error path via server handler
|
|
await server._handle_approval_respond(client, {
|
|
"type": proto.C_APPROVAL_RESPOND,
|
|
"approval_id": "nope", "choice": "once"}, "r1")
|
|
ev = client.ws.events()[-1]
|
|
assert ev["error"]["code"] == proto.ERR_APPROVAL_NOT_FOUND
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_clarify_push_and_resolve_roundtrip(self, tmp_path):
|
|
server = make_server(tmp_path)
|
|
client = FakeClientConnection()
|
|
client.authenticated = True
|
|
server._clients["t"] = client
|
|
|
|
from pheby import hermes_bridge
|
|
await hermes_bridge.push_clarify(
|
|
"clar1", "sk", "Deploy where?", ["staging", "prod"])
|
|
req = [e for e in client.ws.events()
|
|
if e["type"] == proto.S_CLARIFY_REQUEST][-1]
|
|
assert req["question"] == "Deploy where?"
|
|
assert req["choices"] == ["staging", "prod"]
|
|
assert req["allow_free_text"] is True
|
|
|
|
# Register the clarify in Hermes's real gateway primitive so the
|
|
# full resolve path (tools.clarify_gateway) is exercised.
|
|
from tools import clarify_gateway as cg
|
|
cg.register(clarify_id="clar1", session_key="sk",
|
|
question="Deploy where?", choices=["staging", "prod"])
|
|
from pheby import hermes_bridge as hb
|
|
ok = await hb.resolve_clarify("clar1", "staging")
|
|
assert ok is True
|
|
ok2 = await hb.resolve_clarify("clar1", "staging")
|
|
assert ok2 is False # entry consumed
|
|
cg.clear_session("sk")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_chat_send_creates_message_event(self, tmp_path):
|
|
server = make_server(tmp_path)
|
|
client = FakeClientConnection()
|
|
client.authenticated = True
|
|
server._clients["t"] = client
|
|
|
|
from pheby import hermes_bridge as hb
|
|
await hb.send_chat(server, "conv77", "hello Hermes", client, "r1")
|
|
adapter = server.adapter
|
|
assert len(adapter.handled) == 1
|
|
assert adapter.handled[0].text == "hello Hermes"
|
|
assert adapter.handled[0].source.chat_id == "conv77"
|
|
events = client.ws.events()
|
|
assert events[0]["type"] == proto.S_RUN_ACCEPTED
|
|
assert events[1]["type"] == proto.S_MESSAGE_START
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_chat_handler_passes_server_to_bridge(self, tmp_path):
|
|
server = make_server(tmp_path)
|
|
client = FakeClientConnection()
|
|
client.authenticated = True
|
|
cid = "a" * 32
|
|
|
|
await server._handle_chat_send(client, {
|
|
"type": proto.C_CHAT_SEND,
|
|
"conversation_id": cid,
|
|
"text": "hello through WebSocket",
|
|
}, "request-handler")
|
|
|
|
assert server.adapter.handled[-1].text == "hello through WebSocket"
|
|
assert server.adapter.handled[-1].source.chat_id == cid
|
|
assert client.ws.events()[0]["type"] == proto.S_RUN_ACCEPTED
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_chat_dispatch_failure_closes_run(self, tmp_path,
|
|
monkeypatch):
|
|
server = make_server(tmp_path)
|
|
client = FakeClientConnection()
|
|
client.authenticated = True
|
|
server._clients["t"] = client
|
|
cid = "b" * 32
|
|
|
|
async def reject(_event):
|
|
raise RuntimeError("gateway unavailable")
|
|
|
|
monkeypatch.setattr(server.adapter, "handle_message", reject)
|
|
from pheby import hermes_bridge as hb
|
|
with pytest.raises(RuntimeError, match="gateway unavailable"):
|
|
await hb.send_chat(server, cid, "hello", client, "request")
|
|
|
|
assert hb.active_run(cid) is None
|
|
assert client.ws.events()[-1]["type"] == proto.S_RUN_FINISHED
|
|
assert client.ws.events()[-1]["status"] == "failed"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_removes_route_instead_of_resetting(self, tmp_path,
|
|
monkeypatch):
|
|
server = make_server(tmp_path)
|
|
cid = await server.router.new_conversation("Delete me")
|
|
session_key = f"agent:main:pheby:dm:{cid}"
|
|
|
|
class Entry:
|
|
display_name = "Delete me"
|
|
|
|
class Store:
|
|
def __init__(self):
|
|
self._lock = threading.Lock()
|
|
self._entries = {session_key: Entry()}
|
|
self.saved = False
|
|
|
|
def _ensure_loaded_locked(self):
|
|
return None
|
|
|
|
def _save(self):
|
|
self.saved = True
|
|
|
|
def peek_session_id(self, key):
|
|
return "session-delete" if key == session_key else None
|
|
|
|
class DB:
|
|
deleted = None
|
|
|
|
def delete_session(self, session_id):
|
|
self.deleted = session_id
|
|
return True
|
|
|
|
store, db = Store(), DB()
|
|
from pheby import hermes_bridge as hb
|
|
monkeypatch.setattr(hb, "_session_store", lambda: store)
|
|
monkeypatch.setattr(hb, "_session_db", lambda: db)
|
|
|
|
assert await hb.delete_conversation(cid) is True
|
|
assert db.deleted == "session-delete"
|
|
assert session_key not in store._entries
|
|
assert store.saved is True
|
|
assert await server.router.get_name(cid) is None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_models_snapshot_uses_current_hermes_signature(
|
|
self, tmp_path, monkeypatch):
|
|
make_server(tmp_path)
|
|
captured = {}
|
|
|
|
def fake_list_picker_providers(**kwargs):
|
|
captured.update(kwargs)
|
|
return [{"slug": "test", "models": ["m1"]}]
|
|
|
|
from pheby import hermes_bridge as hb
|
|
import hermes_cli.config as hermes_config
|
|
import hermes_cli.model_switch as model_switch
|
|
monkeypatch.setattr(hb, "_load_cfg", lambda: {
|
|
"model": {"default": "m1", "provider": "test"},
|
|
"providers": {"test": {"base_url": "http://example"}},
|
|
"model_catalog": {"excluded_providers": ["hidden"]},
|
|
})
|
|
monkeypatch.setattr(
|
|
hermes_config, "get_compatible_custom_providers",
|
|
lambda _cfg: [{"name": "test", "base_url": "http://example"}])
|
|
monkeypatch.setattr(
|
|
model_switch, "list_picker_providers", fake_list_picker_providers)
|
|
|
|
snapshot = await hb.models_snapshot()
|
|
assert snapshot["providers"][0]["slug"] == "test"
|
|
assert captured["current_model"] == "m1"
|
|
assert captured["excluded_providers"] == ["hidden"]
|
|
assert captured["custom_providers"][0]["name"] == "test"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cancel_run_interrupts_agent(self, tmp_path):
|
|
server = make_server(tmp_path)
|
|
client = FakeClientConnection()
|
|
client.authenticated = True
|
|
server._clients["t"] = client
|
|
|
|
runner = FakeRunner()
|
|
agent = FakeAgent()
|
|
runner._running_agents["agent:main:pheby:dm:convX"] = agent
|
|
server.adapter.gateway_runner = runner
|
|
|
|
from pheby import hermes_bridge as hb
|
|
hb._ACTIVE_RUNS["convX"] = {"run_id": "run-x", "started": 0}
|
|
ok = await hb.cancel_run("convX", None)
|
|
assert ok is True
|
|
assert agent.interrupts # agent.interrupt called, not thread-kill
|
|
assert runner.generations.get("agent:main:pheby:dm:convX") == 1
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cancel_stale_run_id_rejected(self, tmp_path):
|
|
server = make_server(tmp_path)
|
|
from pheby import hermes_bridge as hb
|
|
server._bridge_active("convY") if hasattr(
|
|
server, "_bridge_active") else None
|
|
hb._ACTIVE_RUNS["convY"] = {"run_id": "run1", "started": 0}
|
|
ok = await hb.cancel_run("convY", "wrong-run")
|
|
assert ok is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_reasoning_set_validation(self, tmp_path):
|
|
from pheby import hermes_bridge as hb
|
|
bad = await hb.set_reasoning("turbo", None)
|
|
assert bad["ok"] is False
|
|
assert bad["code"] == proto.ERR_BAD_REQUEST
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════════
|
|
# Reconnect / recovery semantics
|
|
# ═══════════════════════════════════════════════════════════════════════════
|
|
class TestRecovery:
|
|
@pytest.mark.asyncio
|
|
async def test_history_resync_after_reconnect(self, tmp_path):
|
|
"""Conversation state is authoritative server-side: a fresh client
|
|
connection re-opening a conversation gets the same history."""
|
|
server = make_server(tmp_path)
|
|
cid = await server.router.new_conversation("Sync")
|
|
# Seed transcript via the fake DB path is covered in bridge tests
|
|
# through Hermes; here assert the contract: open is idempotent.
|
|
c1, c2 = FakeClientConnection(), FakeClientConnection()
|
|
for c in (c1, c2):
|
|
c.authenticated = True
|
|
await server._handle_conversation_open(c1, {
|
|
"type": proto.C_CONVERSATION_OPEN, "conversation_id": cid}, "a")
|
|
await server._handle_conversation_open(c2, {
|
|
"type": proto.C_CONVERSATION_OPEN, "conversation_id": cid}, "b")
|
|
h1 = c1.ws.events()[-1]
|
|
h2 = c2.ws.events()[-1]
|
|
assert h1["messages"] == h2["messages"]
|
|
assert h1["conversation_id"] == h2["conversation_id"] == cid
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_open_recovers_runtime_and_attachment_state(self, tmp_path):
|
|
server = make_server(tmp_path)
|
|
cid = await server.router.new_conversation("Recover")
|
|
source = Path(tmp_path) / "recovery.png"
|
|
source.write_bytes(b"\x89PNG recovery")
|
|
attachment = await server.store.register_file(
|
|
str(source), conversation_id=cid)
|
|
|
|
from pheby import hermes_bridge as hb
|
|
hb._ACTIVE_RUNS[cid] = {"run_id": "run-recover", "started": 1}
|
|
hb.record_tool_event(cid, {
|
|
"type": proto.S_TOOL_EVENT,
|
|
"conversation_id": cid,
|
|
"run_id": "run-recover",
|
|
"tool_call_id": "tool-recover",
|
|
"tool_name": "terminal",
|
|
"status": "running",
|
|
})
|
|
await hb.push_approval(
|
|
{"command": "echo hi", "description": "Run command"},
|
|
f"agent:main:pheby:dm:{cid}")
|
|
await hb.push_clarify(
|
|
"clarify-recover", f"agent:main:pheby:dm:{cid}",
|
|
"Continue?", ["yes", "no"])
|
|
|
|
client = FakeClientConnection()
|
|
client.authenticated = True
|
|
await server._handle_conversation_open(client, {
|
|
"type": proto.C_CONVERSATION_OPEN,
|
|
"conversation_id": cid,
|
|
}, "recover")
|
|
snapshot = client.ws.events()[-1]
|
|
assert snapshot["attachments"][0]["attachment_id"] == \
|
|
attachment["attachment_id"]
|
|
assert snapshot["run"]["run_id"] == "run-recover"
|
|
assert snapshot["tools"][0]["tool_call_id"] == "tool-recover"
|
|
assert snapshot["approvals"][0]["conversation_id"] == cid
|
|
assert snapshot["clarifications"][0]["clarify_id"] == \
|
|
"clarify-recover"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_broadcast_reaches_multiple_clients(self, tmp_path):
|
|
server = make_server(tmp_path)
|
|
clients = []
|
|
for i in range(3):
|
|
c = FakeClientConnection()
|
|
c.authenticated = True
|
|
server._clients[f"c{i}"] = c
|
|
clients.append(c)
|
|
await server.broadcast({"type": proto.S_PONG, "ts": "t"})
|
|
for c in clients:
|
|
assert any(e["type"] == proto.S_PONG for e in c.ws.events())
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════════
|
|
# Config
|
|
# ═══════════════════════════════════════════════════════════════════════════
|
|
class TestConfig:
|
|
def test_env_secret_wins(self, monkeypatch):
|
|
monkeypatch.setenv("PHEBY_SECRET", "env-secret")
|
|
cfg = load_config({"secret": "yaml-secret", "port": 9999})
|
|
assert cfg.secret == "env-secret"
|
|
|
|
def test_yaml_fallback_and_defaults(self, monkeypatch):
|
|
monkeypatch.delenv("PHEBY_SECRET", raising=False)
|
|
cfg = load_config({"secret": "yaml-secret"})
|
|
assert cfg.secret == "yaml-secret"
|
|
assert cfg.bind_host == "127.0.0.1"
|
|
assert cfg.port == 8620
|
|
assert cfg.retention_days == 7
|
|
assert cfg.enabled
|
|
|
|
def test_disabled_without_secret(self, monkeypatch):
|
|
monkeypatch.delenv("PHEBY_SECRET", raising=False)
|
|
cfg = load_config({})
|
|
assert not cfg.enabled
|
|
|
|
def test_bad_port_falls_back(self, monkeypatch):
|
|
monkeypatch.delenv("PHEBY_SECRET", raising=False)
|
|
cfg = load_config({"secret": "s", "port": "not-a-port"})
|
|
assert cfg.port == 8620
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════════
|
|
# Live HTTP+WS smoke (localhost only)
|
|
# ═══════════════════════════════════════════════════════════════════════════
|
|
class TestLiveServer:
|
|
@pytest.mark.asyncio
|
|
async def test_health_and_ws_roundtrip(self, tmp_path):
|
|
try:
|
|
import aiohttp
|
|
except ImportError:
|
|
pytest.skip("aiohttp unavailable")
|
|
server = make_server(tmp_path, port=0)
|
|
# Bind on an ephemeral port by patching TCPSite port choice.
|
|
cfg = server.config
|
|
cfg.port = 0 # let OS choose
|
|
ok = await server.start()
|
|
if not ok:
|
|
pytest.skip("could not bind test server")
|
|
try:
|
|
port = server._site._server.sockets[0].getsockname()[1]
|
|
base = f"http://127.0.0.1:{port}"
|
|
async with aiohttp.ClientSession() as http:
|
|
# health: no auth
|
|
async with http.get(f"{base}/health") as resp:
|
|
assert resp.status == 200
|
|
data = await resp.json()
|
|
assert data["status"] == "ok"
|
|
# attachment without auth → 401
|
|
async with http.get(
|
|
f"{base}/attachments/{'a'*32}") as resp:
|
|
assert resp.status == 401
|
|
|
|
# WS handshake with bad secret → server sends error event
|
|
async with http.ws_connect(f"{base}/ws") as ws:
|
|
await ws.send_str(json.dumps(
|
|
{"type": "hello", "secret": "bad"}))
|
|
msg = await ws.receive()
|
|
reply = json.loads(msg.data)
|
|
assert reply["type"] == proto.S_ERROR
|
|
assert reply["error"]["code"] == proto.ERR_UNAUTHORIZED
|
|
finally:
|
|
await server.stop()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_full_ws_flow(self, tmp_path):
|
|
"""hello → ready → ping/pong → conversation create → list."""
|
|
try:
|
|
import aiohttp
|
|
except ImportError:
|
|
pytest.skip("aiohttp unavailable")
|
|
server = make_server(tmp_path)
|
|
cfg = server.config
|
|
cfg.port = 0
|
|
ok = await server.start()
|
|
if not ok:
|
|
pytest.skip("could not bind test server")
|
|
try:
|
|
port = server._site._server.sockets[0].getsockname()[1]
|
|
async with aiohttp.ClientSession() as http:
|
|
async with http.ws_connect(
|
|
f"http://127.0.0.1:{port}/ws") as ws:
|
|
await ws.send_str(json.dumps({
|
|
"type": "hello",
|
|
"secret": "test-secret-abc123",
|
|
"protocol_version": proto.PROTOCOL_VERSION}))
|
|
ready = json.loads((await ws.receive()).data)
|
|
assert ready["type"] == proto.S_READY
|
|
|
|
await ws.send_str(json.dumps({"type": "ping"}))
|
|
pong = json.loads((await ws.receive()).data)
|
|
assert pong["type"] == proto.S_PONG
|
|
|
|
await ws.send_str(json.dumps({
|
|
"type": "conversation.create", "name": "Live",
|
|
"request_id": "r1"}))
|
|
created = json.loads((await ws.receive()).data)
|
|
assert created["type"] == proto.S_CONVERSATION_CREATED
|
|
assert created["request_id"] == "r1"
|
|
cid = created["conversation_id"]
|
|
# The handler also broadcasts a conversation.updated event
|
|
updated = json.loads((await ws.receive()).data)
|
|
assert updated["type"] == proto.S_CONVERSATION_UPDATED
|
|
|
|
await ws.send_str(json.dumps({
|
|
"type": "conversation.list", "request_id": "r2"}))
|
|
snap = json.loads((await ws.receive()).data)
|
|
assert any(c["conversation_id"] == cid
|
|
for c in snap["conversations"])
|
|
|
|
# unknown type → structured error
|
|
await ws.send_str(json.dumps({"type": "bogus.thing"}))
|
|
err = json.loads((await ws.receive()).data)
|
|
assert err["type"] == proto.S_ERROR
|
|
assert err["error"]["code"] == proto.ERR_UNKNOWN_TYPE
|
|
finally:
|
|
await server.stop()
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════════
|
|
# Adapter unit checks
|
|
# ═══════════════════════════════════════════════════════════════════════════
|
|
class TestAdapterUnits:
|
|
@pytest.mark.asyncio
|
|
async def test_stream_preview_keeps_run_open_until_finalize(self, tmp_path):
|
|
from pheby.adapter import PhebyAdapter
|
|
from gateway.config import PlatformConfig
|
|
from pheby import hermes_bridge as hb
|
|
|
|
server = make_server(tmp_path)
|
|
client = FakeClientConnection()
|
|
client.authenticated = True
|
|
server._clients["t"] = client
|
|
adapter = PhebyAdapter(PlatformConfig(
|
|
enabled=True, extra={"secret": "test-secret-abc123"}))
|
|
adapter._server = server
|
|
adapter._loop = asyncio.get_running_loop()
|
|
adapter._drafts["conv"] = {"message_id": "draft-run", "text": ""}
|
|
hb.set_adapter(adapter)
|
|
hb.set_server(server)
|
|
hb._ACTIVE_RUNS["conv"] = {"run_id": "run-1", "started": 0}
|
|
|
|
first = await adapter.send("conv", "hel", metadata={"expect_edits": True})
|
|
assert first.message_id == "draft-run"
|
|
assert hb.active_run_id("conv") == "run-1"
|
|
assert client.ws.events()[-1]["type"] == proto.S_MESSAGE_DELTA
|
|
|
|
final = await adapter.edit_message(
|
|
"conv", "draft-run", "hello", finalize=True)
|
|
await asyncio.sleep(0)
|
|
assert final.message_id == "draft-run"
|
|
assert hb.active_run_id("conv") is None
|
|
assert [e["type"] for e in client.ws.events()][-2:] == [
|
|
proto.S_MESSAGE_COMPLETE, proto.S_RUN_FINISHED]
|
|
|
|
def test_transport_auth_is_gateway_authorization(self):
|
|
from pheby.adapter import PhebyAdapter
|
|
from gateway.config import PlatformConfig
|
|
adapter = PhebyAdapter(PlatformConfig(
|
|
enabled=True, extra={"secret": "test-secret-abc123"}))
|
|
assert adapter.authorization_is_upstream is True
|
|
|
|
def test_redact_args(self):
|
|
from pheby.adapter import _redact_args
|
|
out = _redact_args({"query": "cats", "api_key": "sk-123",
|
|
"token": "t", "long": "x" * 900})
|
|
assert out["api_key"] == "[redacted]"
|
|
assert out["token"] == "[redacted]"
|
|
assert out["query"] == "cats"
|
|
assert out["long"].endswith("…")
|
|
|
|
def test_redact_args_recursively(self):
|
|
from pheby.adapter import _redact_args
|
|
out = _redact_args({
|
|
"headers": {"Authorization": "Bearer secret"},
|
|
"steps": [{"password": "hunter2", "value": "safe"}],
|
|
})
|
|
assert out["headers"]["Authorization"] == "[redacted]"
|
|
assert out["steps"][0]["password"] == "[redacted]"
|
|
assert out["steps"][0]["value"] == "safe"
|
|
|
|
def test_sanitize_filename(self):
|
|
from pheby.attachments import AttachmentStore
|
|
assert AttachmentStore._sanitize_filename("../../etc/passwd") == "passwd"
|
|
# Path separators (either flavor) collapse to the final component.
|
|
assert AttachmentStore._sanitize_filename("a/b\\c.txt") == "c.txt"
|
|
assert AttachmentStore._sanitize_filename("") == "file.bin"
|