From 3ab601582ece391884b922185ab18f3fd38cfc9c Mon Sep 17 00:00:00 2001 From: Raphael Avocegamou Date: Tue, 9 Jun 2026 20:57:09 -0400 Subject: [PATCH] =?UTF-8?q?feat(sdk-python):=20rewrite=20to=20async=20gene?= =?UTF-8?q?rator=20API=20per=20arch=20doc=20=C2=A712.1,=20fix=20binary=20w?= =?UTF-8?q?ire=20format?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Delete pocketstation/client.py (HTTP-only, no WebSocket, no LEAVE) - Delete tests/test_client.py (tests for deleted module) - Rewrite pocketstation/station.py: single PocketStation class with connect()/listen()/broadcast()/disconnect() and async context manager - broadcast() sends raw binary PCM bytes, not base64 JSON (wire format fix) - listen() propagates WebSocket errors to caller instead of swallowing them - disconnect() sends LEAVE message before closing WebSocket - AudioFrame gains timestamp_ns field (arch doc §4.1 sequence_number + timestamp_ns) - Update __init__.py: remove PocketStationClient/PocketStationSession exports - Burn down FAKE_SCAFFOLD_INVENTORY.md 'close() LEAVE message' partial row - 13 tests passing (9 new GWT-structured tests covering the four behavioral properties) --- docs/standards/FAKE_SCAFFOLD_INVENTORY.md | 1 - pocketstation/__init__.py | 5 +- pocketstation/client.py | 137 ----------- pocketstation/station.py | 235 ++++++++++--------- pocketstation/types.py | 7 +- tests/test_client.py | 126 ---------- tests/test_station.py | 273 ++++++++++++++++++---- 7 files changed, 368 insertions(+), 416 deletions(-) delete mode 100644 pocketstation/client.py delete mode 100644 tests/test_client.py diff --git a/docs/standards/FAKE_SCAFFOLD_INVENTORY.md b/docs/standards/FAKE_SCAFFOLD_INVENTORY.md index c037401..3eafe7a 100644 --- a/docs/standards/FAKE_SCAFFOLD_INVENTORY.md +++ b/docs/standards/FAKE_SCAFFOLD_INVENTORY.md @@ -24,7 +24,6 @@ DEFERRED Intentionally postponed; ADR or phase plan justifies it | Component | Status | File | What's missing | Replace by | Blocked on | |-----------|--------|------|----------------|------------|------------| -| `close()` LEAVE message | PARTIAL | `pocketstation/station.py` | Relay LEAVE message not sent on `close()`. WebSocket closes but relay does not receive LEAVE type. Room state may be stale until TTL. | Phase 5 | None — straightforward fix | | WebRTC transport | DEFERRED | N/A | Python SDK is WebSocket-only. No WebRTC / audio capture planned for this binding. | Not planned | Architecture decision: Python is listener/voice-agent tier only | --- diff --git a/pocketstation/__init__.py b/pocketstation/__init__.py index 3fa8738..362c253 100644 --- a/pocketstation/__init__.py +++ b/pocketstation/__init__.py @@ -1,13 +1,10 @@ -"""PocketStation Python SDK. Phase 5.""" +"""PocketStation Python SDK — spec §12.1.""" from .station import PocketStation -from .client import PocketStation as PocketStationClient, PocketStationSession from .types import AudioFrame, AudioMode, IceServer, PocketStationError, RoomCredentials __version__ = "0.1.0" __all__ = [ "PocketStation", - "PocketStationClient", - "PocketStationSession", "AudioFrame", "AudioMode", "RoomCredentials", diff --git a/pocketstation/client.py b/pocketstation/client.py deleted file mode 100644 index 9600e95..0000000 --- a/pocketstation/client.py +++ /dev/null @@ -1,137 +0,0 @@ -""" -PocketStation Python async client SDK. - -Phase scope: Phase 5. - -Usage:: - - import asyncio - from pocketstation import PocketStation - - async def main(): - async with PocketStation.connect( - api_url="https://api.pocketstation.io", - relay_url="wss://relay.pocketstation.io", - ) as session: - print(f"Connected to room {session.room_id}") - creds = session.credentials - # Use creds.source_token / creds.listener_token for signaling - - asyncio.run(main()) -""" -from __future__ import annotations - -import json -from contextlib import asynccontextmanager -from typing import AsyncGenerator, Optional - -import httpx - -from .types import IceServer, PocketStationError, RoomCredentials - - -class PocketStationSession: - """ - Active PocketStation session. - - Invariant: valid only within the ``async with PocketStation.connect()`` block. - Ownership: created and owned by PocketStation.connect(); do not instantiate directly. - Failure behavior: network errors raise PocketStationError. - - Phase 5: HTTP room creation implemented. WebSocket signaling and WebRTC - publish/subscribe wiring are Phase 5 follow-up (requires native WebRTC - Python binding or aiortc). - """ - - def __init__(self, credentials: RoomCredentials) -> None: - self._credentials = credentials - - @property - def credentials(self) -> RoomCredentials: - """Room credentials including TURN servers (PY-023).""" - return self._credentials - - @property - def room_id(self) -> str: - return self._credentials.room_id - - @property - def source_token(self) -> str: - return self._credentials.source_token - - @property - def listener_token(self) -> str: - return self._credentials.listener_token - - @property - def ice_servers(self) -> list[IceServer]: - """TURN/STUN servers for WebRTC PeerConnection config.""" - return self._credentials.ice_servers - - -class PocketStation: - """ - PocketStation async context manager. - - Creates a room on enter, cleans up on exit. - """ - - @staticmethod - @asynccontextmanager - async def connect( - *, - api_url: str, - relay_url: str, - credentials: Optional[RoomCredentials] = None, - ) -> AsyncGenerator[PocketStationSession, None]: - """ - Async context manager that creates a PocketStation session. - - On entry: creates a new room via POST /v1/rooms (or uses provided credentials). - On exit: releases the session (future: sends LEAVE, closes WebSocket). - - :param api_url: Base URL of the api-server. - :param relay_url: Base URL of the relay (wss://...). - :param credentials: Pre-obtained credentials. If None, a new room is created. - :raises PocketStationError: on network or protocol failure. - """ - if credentials is None: - credentials = await PocketStation._create_room(api_url) - - session = PocketStationSession(credentials) - try: - yield session - finally: - # Phase 5 TODO: send LEAVE via signaling WebSocket. - pass - - @staticmethod - async def create_room(api_url: str) -> RoomCredentials: - """Create a new room and return credentials. Does not start a session.""" - return await PocketStation._create_room(api_url) - - @staticmethod - async def _create_room(api_url: str) -> RoomCredentials: - url = api_url.rstrip("/") + "/v1/rooms" - try: - async with httpx.AsyncClient() as http: - response = await http.post(url, json={}) - except httpx.RequestError as exc: - raise PocketStationError( - f"network error creating room: {exc}", "network_error" - ) from exc - - if not response.is_success: - raise PocketStationError( - f"relay returned HTTP {response.status_code}: {response.text}", - "http_error", - ) - - try: - data = response.json() - except json.JSONDecodeError as exc: - raise PocketStationError( - f"failed to parse room creation response: {exc}", "parse_error" - ) from exc - - return RoomCredentials.from_dict(data) diff --git a/pocketstation/station.py b/pocketstation/station.py index 9a9a03b..da4f7ca 100644 --- a/pocketstation/station.py +++ b/pocketstation/station.py @@ -1,18 +1,28 @@ -"""High-level PocketStation session API (spec §12.1).""" +"""PocketStation session API — spec §12.1. + +Wire format contract: + - broadcast() sends raw binary PCM bytes over the WebSocket, never base64 JSON. + - listen() yields binary WebSocket frames as AudioFrame objects. + - Errors from the WebSocket propagate to the caller; they are never swallowed. + - disconnect() sends a LEAVE message before closing the WebSocket. + +Phase scope: Phase 5 — WebSocket listener / voice-agent mode. +WebRTC transport is intentionally out of scope for this binding (see FAKE_SCAFFOLD_INVENTORY). +""" from __future__ import annotations -import base64 import json -from typing import AsyncIterator, Callable, Optional +import time +from typing import AsyncIterator, Optional import httpx import websockets from .types import AudioFrame, AudioMode, IceServer, PocketStationError, RoomCredentials -_AUDIO_FRAME_MSG_TYPE = "AUDIO_FRAME" -_SUBSCRIBE_MSG_TYPE = "SUBSCRIBE" -_ROOM_STATE_MSG_TYPE = "ROOM_STATE" +_MSG_TYPE_SUBSCRIBE = "SUBSCRIBE" +_MSG_TYPE_LEAVE = "LEAVE" +_MSG_TYPE_ROOM_STATE = "ROOM_STATE" _DEFAULT_API_URL = "http://localhost:8090" _DEFAULT_RELAY_URL = "ws://localhost:8080/v1/signal" @@ -20,91 +30,157 @@ class PocketStation: - """Voice agent / broadcast session manager (spec §12.1). + """Voice agent / broadcast session (spec §12.1). - Construct directly when you already have a room ID:: + Lifecycle:: - station = PocketStation(room_id="abc123", mode=AudioMode.VOICE_AGENT) + station = PocketStation(room_id="abc123", relay_url="wss://...", mode=AudioMode.VOICE_AGENT) + await station.connect() async for frame in station.listen(): - transcript = await stt.transcribe(frame.pcm) await station.broadcast(tts_bytes) - await station.close() + await station.disconnect() - Or create a room and connect in one step:: + Or as a context manager:: - station = await PocketStation.create(relay_url="wss://relay.pocketstation.io") + async with PocketStation(room_id="abc123", relay_url="wss://...") as station: + async for frame in station.listen(): + await station.broadcast(tts_bytes) """ def __init__( self, *, room_id: Optional[str] = None, - mode: AudioMode = AudioMode.VOICE_AGENT, - api_url: str = _DEFAULT_API_URL, relay_url: str = _DEFAULT_RELAY_URL, - opus_frame_duration_ms: int = _DEFAULT_FRAME_DURATION_MS, - on_listener_count: Optional[Callable[[int], None]] = None, - on_packet_loss: Optional[Callable[[], None]] = None, + api_url: str = _DEFAULT_API_URL, + mode: AudioMode = AudioMode.VOICE_AGENT, ) -> None: self.room_id = room_id - self.mode = mode - self.api_url = api_url self.relay_url = relay_url - self.frame_duration_ms = opus_frame_duration_ms - self.on_listener_count = on_listener_count - self.on_packet_loss = on_packet_loss + self.api_url = api_url + self.mode = mode self._ws: Optional[websockets.WebSocketClientProtocol] = None self._credentials: Optional[RoomCredentials] = None - @classmethod - async def create( - cls, - *, - relay_url: str = _DEFAULT_RELAY_URL, - api_url: str = _DEFAULT_API_URL, - mode: AudioMode = AudioMode.VOICE_AGENT, - opus_frame_duration_ms: int = _DEFAULT_FRAME_DURATION_MS, - ) -> "PocketStation": - """Create a new room via the API server and return a ready PocketStation instance. - - :param relay_url: WebSocket URL of the relay. - :param api_url: Base URL of the api-server used to provision the room. - :param mode: Audio session mode. - :param opus_frame_duration_ms: Opus frame duration in milliseconds. - :raises PocketStationError: on network or HTTP failure. + # ------------------------------------------------------------------ + # Context manager + # ------------------------------------------------------------------ + + async def __aenter__(self) -> "PocketStation": + await self.connect() + return self + + async def __aexit__(self, exc_type, exc_val, exc_tb) -> None: + await self.disconnect() + return None + + # ------------------------------------------------------------------ + # Public API + # ------------------------------------------------------------------ + + async def connect(self) -> None: + """POST /v1/rooms, open the WebSocket, send SUBSCRIBE. + + :raises PocketStationError: on HTTP failure. + :raises OSError: if the WebSocket cannot be opened. + """ + credentials = await self._ensure_room() + ws = await websockets.connect(self.relay_url) + self._ws = ws + subscribe = json.dumps({ + "type": _MSG_TYPE_SUBSCRIBE, + "room_id": self.room_id, + "token": credentials.listener_token, + }) + await ws.send(subscribe) + + async def listen(self) -> AsyncIterator[AudioFrame]: + """Yield AudioFrame objects for each binary PCM frame received. + + Errors from the WebSocket propagate to the caller; they are not caught here. + Text (JSON) control frames are consumed silently. + + :raises ConnectionError: (or subclasses) when the WebSocket drops. + :raises websockets.exceptions.WebSocketException: on protocol errors. + """ + if self._ws is None: + raise RuntimeError("call connect() before listen()") + + sequence = 0 + async for message in self._ws: + if isinstance(message, bytes): + yield AudioFrame( + pcm=message, + sequence=sequence, + timestamp_ns=time.monotonic_ns(), + ) + sequence += 1 + # Text frames (e.g. ROOM_STATE JSON) are intentionally skipped here; + # they carry relay control metadata, not audio payload. + + async def broadcast(self, audio: bytes) -> None: + """Send raw PCM bytes to the relay. + + The payload is transmitted as binary WebSocket data — not base64 JSON. + + :param audio: raw PCM bytes (f32-LE 48 kHz mono per PY-013). + :raises RuntimeError: if not connected. + :raises websockets.exceptions.WebSocketException: on send failure. """ - station = cls(api_url=api_url, relay_url=relay_url, mode=mode, - opus_frame_duration_ms=opus_frame_duration_ms) - await station._ensure_room() - return station + if self._ws is None: + raise RuntimeError("call connect() before broadcast()") + await self._ws.send(audio) + + async def disconnect(self) -> None: + """Send LEAVE, then close the WebSocket. + + Safe to call even if already disconnected. + """ + if self._ws is None: + return + ws = self._ws + self._ws = None + try: + leave = json.dumps({ + "type": _MSG_TYPE_LEAVE, + "room_id": self.room_id, + }) + await ws.send(leave) + finally: + await ws.close() + + # ------------------------------------------------------------------ + # Internal helpers + # ------------------------------------------------------------------ async def _ensure_room(self) -> RoomCredentials: - """Create or reuse a room on the API server.""" + """Create or reuse a room via POST /v1/rooms.""" if self._credentials is not None: return self._credentials url = self.api_url.rstrip("/") + "/v1/rooms" try: async with httpx.AsyncClient() as client: - r = await client.post(url, json={}) + response = await client.post(url, json={}) except httpx.RequestError as exc: raise PocketStationError( f"network error creating room: {exc}", "network_error" ) from exc - if not r.is_success: + if not response.is_success: raise PocketStationError( - f"relay returned HTTP {r.status_code}: {r.text}", "http_error" + f"relay returned HTTP {response.status_code}: {response.text}", + "http_error", ) try: - data = r.json() + data = response.json() except Exception as exc: raise PocketStationError( f"failed to parse room creation response: {exc}", "parse_error" ) from exc - creds = RoomCredentials( + credentials = RoomCredentials( room_id=data["room_id"], source_token=data.get("source_token", ""), listener_token=data.get("listener_token", ""), @@ -119,61 +195,6 @@ async def _ensure_room(self) -> RoomCredentials: ], ) if not self.room_id: - self.room_id = creds.room_id - self._credentials = creds - return creds - - async def listen(self) -> AsyncIterator[AudioFrame]: - """Subscribe to audio frames from the relay. - - Connects via WebSocket, sends a SUBSCRIBE message, and yields AudioFrame objects. - Yields nothing if the relay is unreachable. - """ - creds = await self._ensure_room() - subscribe_msg = json.dumps({ - "type": _SUBSCRIBE_MSG_TYPE, - "room_id": self.room_id, - "token": creds.listener_token, - }) - try: - async with websockets.connect(self.relay_url) as ws: - self._ws = ws - await ws.send(subscribe_msg) - seq = 0 - async for raw in ws: - if isinstance(raw, bytes): - yield AudioFrame( - pcm=raw, - sequence=seq, - duration_ms=self.frame_duration_ms, - ) - seq += 1 - elif isinstance(raw, str): - msg = json.loads(raw) - if msg.get("type") == _ROOM_STATE_MSG_TYPE and self.on_listener_count: - self.on_listener_count(msg.get("listener_count", 0)) - except Exception: - return - finally: - self._ws = None - - async def broadcast(self, audio: bytes) -> None: - """Send raw PCM bytes (f32-LE 48 kHz mono) to the relay. - - No-ops silently when not connected. - """ - if self._ws is None: - return - try: - await self._ws.send(json.dumps({ - "type": _AUDIO_FRAME_MSG_TYPE, - "pcm_b64": base64.b64encode(audio).decode(), - })) - except Exception: - pass - - async def close(self) -> None: - """Close the WebSocket connection.""" - if self._ws is not None: - await self._ws.close() - self._ws = None + self.room_id = credentials.room_id + self._credentials = credentials + return credentials diff --git a/pocketstation/types.py b/pocketstation/types.py index 588c8a2..7ff53d3 100644 --- a/pocketstation/types.py +++ b/pocketstation/types.py @@ -19,13 +19,16 @@ class AudioMode(enum.Enum): class AudioFrame: """A single audio frame received from the relay. - pcm: raw PCM bytes, 48 kHz mono f32-LE. + pcm: raw PCM bytes, 48 kHz mono f32-LE (PY-013). + sequence: monotonically increasing frame counter per stream. + timestamp_ns: monotonic nanosecond timestamp at frame receipt. """ pcm: bytes + sequence: int = 0 + timestamp_ns: int = 0 sample_rate: int = 48000 channels: int = 1 duration_ms: int = 20 - sequence: int = 0 @property def samples(self) -> list[float]: diff --git a/tests/test_client.py b/tests/test_client.py deleted file mode 100644 index 27f0e11..0000000 --- a/tests/test_client.py +++ /dev/null @@ -1,126 +0,0 @@ -"""Unit tests for pocketstation.client. Phase 5.""" -from __future__ import annotations - -from unittest.mock import AsyncMock, MagicMock, patch - -import pytest - -from pocketstation import PocketStationClient as PocketStation, PocketStationError, RoomCredentials - - -VALID_ROOM_RESPONSE = { - "room_id": "room-py-001", - "source_token": "src-py", - "listener_token": "lst-py", -} - -ROOM_RESPONSE_WITH_ICE = { - **VALID_ROOM_RESPONSE, - "ice_servers": [ - {"urls": ["stun:relay.example.com:3478"]}, - {"urls": ["turn:relay.example.com:3478"], "username": "u", "credential": "p"}, - ], -} - - -def _mock_http_client(json_response: dict, status_code: int = 200): - """Return a context-manager mock for httpx.AsyncClient.""" - mock_response = MagicMock() - mock_response.is_success = status_code < 400 - mock_response.status_code = status_code - mock_response.text = "" - mock_response.json = MagicMock(return_value=json_response) - - mock_client = AsyncMock() - mock_client.post = AsyncMock(return_value=mock_response) - mock_client.__aenter__ = AsyncMock(return_value=mock_client) - mock_client.__aexit__ = AsyncMock(return_value=None) - return mock_client - - -def _mock_http_client_error(status_code: int = 500, text: str = "internal error"): - """Return a mock httpx.AsyncClient whose response indicates an HTTP error.""" - mock_response = MagicMock() - mock_response.is_success = False - mock_response.status_code = status_code - mock_response.text = text - mock_response.json = MagicMock(return_value={}) - - mock_client = AsyncMock() - mock_client.post = AsyncMock(return_value=mock_response) - mock_client.__aenter__ = AsyncMock(return_value=mock_client) - mock_client.__aexit__ = AsyncMock(return_value=None) - return mock_client - - -@pytest.mark.asyncio -async def test_given_valid_api_server_when_connect_then_session_has_credentials(): - # Given - mock_client = _mock_http_client(VALID_ROOM_RESPONSE, status_code=201) - # When - with patch("pocketstation.client.httpx.AsyncClient", return_value=mock_client): - async with PocketStation.connect( - api_url="http://api.example.com", - relay_url="ws://relay.example.com", - ) as session: - # Then - assert session.room_id == "room-py-001" - assert session.source_token == "src-py" - assert session.listener_token == "lst-py" - assert session.ice_servers == [] - - -@pytest.mark.asyncio -async def test_given_api_server_with_turn_when_connect_then_ice_servers_forwarded(): - # Given - mock_client = _mock_http_client(ROOM_RESPONSE_WITH_ICE, status_code=201) - # When - with patch("pocketstation.client.httpx.AsyncClient", return_value=mock_client): - async with PocketStation.connect( - api_url="http://api.example.com", - relay_url="ws://relay.example.com", - ) as session: - # Then - assert len(session.ice_servers) == 2 - assert session.ice_servers[1].username == "u" - - -@pytest.mark.asyncio -async def test_given_api_server_500_when_connect_then_raises(): - # Given - mock_client = _mock_http_client_error(status_code=500, text="internal error") - # When / Then - with patch("pocketstation.client.httpx.AsyncClient", return_value=mock_client): - with pytest.raises(PocketStationError) as exc_info: - async with PocketStation.connect( - api_url="http://api.example.com", - relay_url="ws://relay.example.com", - ): - pass - assert exc_info.value.code == "http_error" - - -@pytest.mark.asyncio -async def test_given_preexisting_credentials_when_connect_then_no_http_call(): - # Given — no patch; any real HTTP call would fail with PermissionError in sandbox - creds = RoomCredentials.from_dict(VALID_ROOM_RESPONSE) - # When - async with PocketStation.connect( - api_url="http://should-not-be-called.example.com", - relay_url="ws://relay.example.com", - credentials=creds, - ) as session: - # Then — session uses provided credentials without making any HTTP call - assert session.room_id == "room-py-001" - - -@pytest.mark.asyncio -async def test_given_session_when_credentials_property_then_returns_room_credentials(): - creds = RoomCredentials.from_dict(VALID_ROOM_RESPONSE) - async with PocketStation.connect( - api_url="http://unused.example.com", - relay_url="ws://relay.example.com", - credentials=creds, - ) as session: - assert isinstance(session.credentials, RoomCredentials) - assert session.credentials.room_id == "room-py-001" diff --git a/tests/test_station.py b/tests/test_station.py index 284e603..222a32c 100644 --- a/tests/test_station.py +++ b/tests/test_station.py @@ -1,6 +1,7 @@ """Unit tests for pocketstation.station — spec §12.1 voice agent pattern.""" from __future__ import annotations +import json import struct from unittest.mock import AsyncMock, MagicMock, patch @@ -23,37 +24,73 @@ def _make_pcm(n_samples: int = 4) -> bytes: def _mock_http_client(json_response: dict, status_code: int = 200): - """Return a context-manager mock for httpx.AsyncClient that yields a mock response.""" + """Return a context-manager mock for httpx.AsyncClient.""" mock_response = MagicMock() - mock_response.raise_for_status = MagicMock() + mock_response.is_success = status_code < 400 + mock_response.status_code = status_code + mock_response.text = "" mock_response.json = MagicMock(return_value=json_response) mock_client = AsyncMock() mock_client.post = AsyncMock(return_value=mock_response) mock_client.__aenter__ = AsyncMock(return_value=mock_client) mock_client.__aexit__ = AsyncMock(return_value=None) - return mock_client +class _FakeWebSocket: + """Minimal WebSocket stand-in for tests. + + Supports: send (AsyncMock), close (AsyncMock), async iteration over a + fixed message list. Not a MagicMock so dunder methods work correctly. + """ + + def __init__(self, messages: list) -> None: + self._messages = messages + self.send = AsyncMock() + self.close = AsyncMock() + + def __aiter__(self): + return self._gen() + + async def _gen(self): + for msg in self._messages: + yield msg + + +def _make_connect_mock(fake_ws: "_FakeWebSocket") -> AsyncMock: + """ + Return an AsyncMock that, when called and awaited, returns fake_ws. + + Patch usage: + with patch("pocketstation.station.websockets.connect", connect_mock): + ... + Then ``await websockets.connect(url)`` in the implementation resolves to fake_ws. + """ + connect_mock = AsyncMock(return_value=fake_ws) + return connect_mock + + # --------------------------------------------------------------------------- -# AudioMode / construction +# Construction # --------------------------------------------------------------------------- def test_given_voice_agent_mode_when_created_then_fields_set(): # Given / When - station = PocketStation(room_id="abc123", mode=AudioMode.VOICE_AGENT) + station = PocketStation( + room_id="abc123", + relay_url="ws://relay.example.com", + mode=AudioMode.VOICE_AGENT, + ) # Then assert station.room_id == "abc123" assert station.mode is AudioMode.VOICE_AGENT - assert station.frame_duration_ms == 20 assert station._ws is None - assert station._credentials is None # --------------------------------------------------------------------------- -# AudioFrame.samples property +# AudioFrame # --------------------------------------------------------------------------- @@ -82,63 +119,221 @@ def test_given_audio_frame_with_known_values_when_samples_then_decoded_correctly assert abs(actual - expected) < 1e-6 +def test_given_audio_frame_when_constructed_then_timestamp_and_sequence_present(): + # Given / When + frame = AudioFrame(pcm=b"\x00" * 16, sequence=7, timestamp_ns=123456789) + # Then + assert frame.sequence == 7 + assert frame.timestamp_ns == 123456789 + + # --------------------------------------------------------------------------- -# _ensure_room — credential reuse +# connect() — POST /v1/rooms + WebSocket SUBSCRIBE # --------------------------------------------------------------------------- @pytest.mark.asyncio -async def test_given_room_id_when_ensure_room_then_reuses_credentials(): - # Given — first call creates room via mocked HTTP client - mock_client = _mock_http_client(VALID_ROOM_RESPONSE) +async def test_given_voice_agent_mode_when_connect_then_posts_room_and_opens_websocket(): + """ + Given a PocketStation in VOICE_AGENT mode, + When connect() is called, + Then it POSTs to /v1/rooms and sends a SUBSCRIBE message over the WebSocket. + """ + # Given + mock_http = _mock_http_client(VALID_ROOM_RESPONSE, status_code=201) + fake_ws = _FakeWebSocket(messages=[]) + connect_mock = _make_connect_mock(fake_ws) + + station = PocketStation( + room_id="room-station-001", + relay_url="ws://relay.example.com", + mode=AudioMode.VOICE_AGENT, + ) + + with patch("pocketstation.station.httpx.AsyncClient", return_value=mock_http), \ + patch("pocketstation.station.websockets.connect", connect_mock): + # When + await station.connect() + + # Then — HTTP room creation happened + mock_http.post.assert_called_once() + posted_url = mock_http.post.call_args[0][0] + assert "/v1/rooms" in posted_url + + # Then — WebSocket was opened and SUBSCRIBE was sent + connect_mock.assert_called_once_with("ws://relay.example.com") + fake_ws.send.assert_called_once() + sent = json.loads(fake_ws.send.call_args[0][0]) + assert sent["type"] == "SUBSCRIBE" + assert sent["room_id"] == "room-station-001" + assert "token" in sent - station = PocketStation(room_id="existing-room") + await station.disconnect() - with patch("pocketstation.station.httpx.AsyncClient", return_value=mock_client): - # When — call _ensure_room twice - creds1 = await station._ensure_room() - creds2 = await station._ensure_room() - # Then — same object returned; HTTP was called only once - assert creds1 is creds2 - assert creds1.room_id == "room-station-001" - mock_client.post.assert_called_once() +# --------------------------------------------------------------------------- +# broadcast() — must send binary bytes, not base64 JSON +# --------------------------------------------------------------------------- @pytest.mark.asyncio -async def test_given_no_room_id_when_ensure_room_then_room_id_assigned(): +async def test_given_connected_when_broadcast_then_sends_binary_bytes_not_base64(): + """ + Given a connected PocketStation, + When broadcast(audio_bytes) is called, + Then the WebSocket send receives raw bytes, not a base64-encoded JSON string. + """ # Given - mock_client = _mock_http_client(VALID_ROOM_RESPONSE) + mock_http = _mock_http_client(VALID_ROOM_RESPONSE, status_code=201) + fake_ws = _FakeWebSocket(messages=[]) + connect_mock = _make_connect_mock(fake_ws) + + station = PocketStation( + room_id="room-station-001", + relay_url="ws://relay.example.com", + mode=AudioMode.VOICE_AGENT, + ) + audio = _make_pcm(n_samples=16) + + with patch("pocketstation.station.httpx.AsyncClient", return_value=mock_http), \ + patch("pocketstation.station.websockets.connect", connect_mock): + await station.connect() + # Reset after the SUBSCRIBE send that happens in connect() + fake_ws.send.reset_mock() + + # When + await station.broadcast(audio) + + # Then — exactly one send call with the raw bytes payload + fake_ws.send.assert_called_once_with(audio) + sent_arg = fake_ws.send.call_args[0][0] + assert isinstance(sent_arg, bytes), ( + f"broadcast() must send bytes, not {type(sent_arg).__name__}" + ) + # Guard: must not be a JSON / base64 string + assert not isinstance(sent_arg, str) + + await station.disconnect() - station = PocketStation() # no room_id - with patch("pocketstation.station.httpx.AsyncClient", return_value=mock_client): +# --------------------------------------------------------------------------- +# __aenter__ / __aexit__ — context manager sends LEAVE on exit +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_given_context_manager_when_exit_then_sends_leave(): + """ + Given a PocketStation used as an async context manager, + When the block exits normally, + Then a LEAVE message is sent and the WebSocket is closed. + """ + # Given + mock_http = _mock_http_client(VALID_ROOM_RESPONSE, status_code=201) + fake_ws = _FakeWebSocket(messages=[]) + connect_mock = _make_connect_mock(fake_ws) + + with patch("pocketstation.station.httpx.AsyncClient", return_value=mock_http), \ + patch("pocketstation.station.websockets.connect", connect_mock): # When - creds = await station._ensure_room() + async with PocketStation( + room_id="room-station-001", + relay_url="ws://relay.example.com", + ) as station: + pass # nothing inside the block + + # Then — LEAVE was sent exactly once + all_sends = fake_ws.send.call_args_list + leave_sends = [ + c for c in all_sends + if isinstance(c[0][0], str) and json.loads(c[0][0]).get("type") == "LEAVE" + ] + assert len(leave_sends) == 1, ( + f"Expected exactly one LEAVE message, got {all_sends}" + ) + # Then — WebSocket was closed + fake_ws.close.assert_called_once() + + +# --------------------------------------------------------------------------- +# listen() — errors must propagate, not be swallowed +# --------------------------------------------------------------------------- - # Then - assert station.room_id == "room-station-001" - assert creds.room_id == "room-station-001" + +@pytest.mark.asyncio +async def test_given_websocket_closed_when_listen_then_raises_connection_error_not_swallows(): + """ + Given that the WebSocket raises a ConnectionError during iteration, + When station.listen() is consumed, + Then the error propagates to the caller instead of being swallowed. + """ + # Given + mock_http = _mock_http_client(VALID_ROOM_RESPONSE, status_code=201) + + class _FailingWebSocket(_FakeWebSocket): + async def _gen(self): + yield _make_pcm(4) # one good frame before the error + raise ConnectionError("relay dropped connection") + + fake_ws = _FailingWebSocket(messages=[]) + connect_mock = _make_connect_mock(fake_ws) + + station = PocketStation( + room_id="room-station-001", + relay_url="ws://relay.example.com", + ) + + with patch("pocketstation.station.httpx.AsyncClient", return_value=mock_http), \ + patch("pocketstation.station.websockets.connect", connect_mock): + await station.connect() + + # When / Then — ConnectionError propagates; only the first frame arrives + frames: list[AudioFrame] = [] + with pytest.raises(ConnectionError, match="relay dropped connection"): + async for frame in station.listen(): + frames.append(frame) + + # One frame was produced before the error + assert len(frames) == 1 # --------------------------------------------------------------------------- -# listen() — websocket unavailable +# listen() — binary frames are yielded as AudioFrame objects # --------------------------------------------------------------------------- @pytest.mark.asyncio -async def test_given_websocket_unavailable_when_listen_then_yields_nothing(): - # Given — HTTP room creation succeeds, WebSocket connect raises immediately - mock_client = _mock_http_client(VALID_ROOM_RESPONSE) +async def test_given_connected_when_listen_then_yields_audio_frames_from_binary_messages(): + """ + Given a WebSocket that delivers three binary PCM frames, + When station.listen() is iterated, + Then three AudioFrame objects are yielded with monotonically increasing sequences. + """ + # Given + mock_http = _mock_http_client(VALID_ROOM_RESPONSE, status_code=201) + pcm_frames = [_make_pcm(4), _make_pcm(4), _make_pcm(4)] + fake_ws = _FakeWebSocket(messages=pcm_frames) + connect_mock = _make_connect_mock(fake_ws) + + station = PocketStation( + room_id="room-station-001", + relay_url="ws://relay.example.com", + ) - station = PocketStation(room_id="abc123", relay_url="ws://127.0.0.1:1") + with patch("pocketstation.station.httpx.AsyncClient", return_value=mock_http), \ + patch("pocketstation.station.websockets.connect", connect_mock): + await station.connect() - with patch("pocketstation.station.httpx.AsyncClient", return_value=mock_client), \ - patch("pocketstation.station.websockets.connect", side_effect=OSError("unreachable")): # When - frames = [] + frames: list[AudioFrame] = [] async for frame in station.listen(): frames.append(frame) - # Then — no frames, no exception raised - assert frames == [] + # Then + assert len(frames) == 3 + for i, frame in enumerate(frames): + assert isinstance(frame, AudioFrame) + assert frame.pcm == pcm_frames[i] + assert frame.sequence == i + + await station.disconnect()