"""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 os import time from pathlib import Path from typing import Any, Dict, List, Optional import pytest from aiohttp import web # 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.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") from hermes_constants import get_hermes_home # conftest redirects home 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 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: @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): 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._active_sessions = {"agent:main:pheby:dm:conv1": asyncio.Event()} # Tool-call dedup state is created lazily via _tool_state() 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 await asyncio.sleep(0) # let ensure_future broadcast run 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"] 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): 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._active_sessions = {"agent:main:pheby:dm:conv1": asyncio.Event()} real.on_post_tool_call( 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): 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" # Resolve: with no real Hermes queue the resolve call fails-open to # accepted=False but the pending entry must be consumed either way. from pheby import hermes_bridge as hb ok = await hb.resolve_approval(req["approval_id"], "deny", None) assert ok is True # pending entry existed; resolution attempted # 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_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 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_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: 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_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"