diff --git a/pyproject.toml b/pyproject.toml index 7fafb78..a8ecb77 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -28,5 +28,11 @@ build-backend = "uv_build" dev = [ "dotenv>=0.9.9", "pipecat-ai[google,silero]>=1.1.0", + "pytest>=8.0", + "pytest-asyncio>=0.24", "uvicorn>=0.46.0", ] + +[tool.pytest.ini_options] +asyncio_mode = "auto" +testpaths = ["tests"] diff --git a/src/pipecat_asterisk/serializer/serializer.py b/src/pipecat_asterisk/serializer/serializer.py index e125030..4d6c87b 100644 --- a/src/pipecat_asterisk/serializer/serializer.py +++ b/src/pipecat_asterisk/serializer/serializer.py @@ -4,8 +4,7 @@ # SPDX-License-Identifier: BSD-2-Clause # -import inspect -from typing import Awaitable, Callable, Optional, cast +from typing import Awaitable, Callable, Optional from loguru import logger from pipecat.audio.dtmf.types import KeypadEntry from pipecat.audio.utils import create_stream_resampler @@ -77,28 +76,58 @@ def __init__(self, sample_rate: int = 0): sample_rate ) # What sample rate is used in Asterisk websocket channel. If 0, will be populated during setup or from MEDIA_START event + # Pre-built dispatch tables for `serialize`. Keyed by frame class so + # `serialize` does a single dict lookup per frame instead of + # `getattr(self, f"_frame_{type(frame).__name__.lower()}")` + an + # `inspect.isawaitable` check on the hot path. Async handlers are + # kept in a separate table so we know at lookup time whether to + # `await` the result. + self._sync_frame_handlers: dict[type, Callable[[Frame], str | bytes | None]] = { + AsteriskCommandFrame: self._frame_asteriskcommandframe, + EndFrame: self._frame_endframe, + CancelFrame: self._frame_cancelframe, + InterruptionFrame: self._frame_interruptionframe, + } + self._async_frame_handlers: dict[ + type, Callable[[Frame], Awaitable[str | bytes | None]] + ] = { + OutputAudioRawFrame: self._frame_outputaudiorawframe, + } + + # Event handlers are keyed by Asterisk event-name string, matching + # the values produced by the protocol parser. All handlers are sync + # so one table is enough; same goal as the frame tables - skip the + # per-event `getattr` + `f"_ev_{name.lower()}"` lookup. + self._event_handlers: dict[str, Callable[[dict], Frame | None]] = { + "MEDIA_START": self._ev_media_start, + "MEDIA_XOFF": self._ev_media_xoff, + "MEDIA_XON": self._ev_media_xon, + "DTMF_END": self._ev_dtmf_end, + "QUEUE_DRAINED": self._ev_queue_drained, + } + def _handle_event(self, message: dict) -> Frame | None: - """Call the event handler if the handler is defined in the class, otherwise return None. + """Dispatch an Asterisk event through the precomputed handler table. - The handler methods should be named as "_ev_{event_name.lower()}" and should take the event message as a dictionary and return a Frame or None. + The event-handler table is built once in ``__init__`` and keyed by + the exact event-name string Asterisk sends. Unknown events fall + through to the same warning path as the original reflective form. Args: message: The event message as a dictionary. """ - message_type = message.get("event", None) + message_type = message.get("event") if message_type is None: logger.warning( f"Received Asterisk WebSocket message without 'event' field: {message}" ) return None - handler = getattr(self, f"_ev_{message_type.lower()}", None) - if callable(handler): - typed_handler = cast(Callable[[dict], Frame | None], handler) - return typed_handler(message) - else: - logger.info(f"Received unhandled Asterisk WebSocket event: {message}") - return None + handler = self._event_handlers.get(message_type) + if handler is not None: + return handler(message) + logger.info(f"Received unhandled Asterisk WebSocket event: {message}") + return None ### Asterisk Event handlers ### @@ -395,24 +424,29 @@ async def setup(self, frame: StartFrame): async def serialize(self, frame: Frame) -> str | bytes | None: """Convert a frame to its serialized representation suitable for Asterisk WebSocket channel. + Looks up the frame's exact type in the precomputed sync/async + dispatch tables built in ``__init__``. This avoids the per-frame + ``getattr`` + f-string + ``inspect.isawaitable`` overhead the old + reflective form paid, which matters because audio frames go through + this method dozens of times per second. + Args: frame: The frame to serialize. Returns: Serialized frame data as string, bytes, or None if serialization fails. """ - handler = getattr(self, f"_frame_{type(frame).__name__.lower()}", None) - if callable(handler): - result = handler(frame) - if inspect.isawaitable(result): - return cast(str | bytes | None, await result) - else: - return cast(str | bytes | None, result) - else: - logger.trace( - f"Received unhandled frame type in Asterisk WebSocket serializer: {type(frame)}. Frame: {frame}" - ) - return None + frame_type = type(frame) + sync_handler = self._sync_frame_handlers.get(frame_type) + if sync_handler is not None: + return sync_handler(frame) + async_handler = self._async_frame_handlers.get(frame_type) + if async_handler is not None: + return await async_handler(frame) + logger.trace( + f"Received unhandled frame type in Asterisk WebSocket serializer: {frame_type}. Frame: {frame}" + ) + return None async def deserialize(self, data: str | bytes) -> Frame | None: """Convert serialized data from Asterisk's websocket channel to a frame object. diff --git a/src/pipecat_asterisk/transport/transport.py b/src/pipecat_asterisk/transport/transport.py index 9360e3e..f511b29 100644 --- a/src/pipecat_asterisk/transport/transport.py +++ b/src/pipecat_asterisk/transport/transport.py @@ -70,12 +70,12 @@ async def _media_start_handler(self, frame: InputTransportMessageFrame): # Send START_MEDIA_BUFFERING command to Asterisk WebSocket channel to enable audio buffering on the Asterisk side if self._client.is_closing or not self._client.is_connected: logger.warning( - f"Cannot send START_MEDIA_BUFFERING command because the WebSocket client is closing or already closed." + "Cannot send START_MEDIA_BUFFERING command because the WebSocket client is closing or already closed." ) return if not self._params.serializer: logger.error( - f"Cannot send START_MEDIA_BUFFERING command because no serializer is set in the transport parameters." + "Cannot send START_MEDIA_BUFFERING command because no serializer is set in the transport parameters." ) return try: @@ -85,7 +85,7 @@ async def _media_start_handler(self, frame: InputTransportMessageFrame): if cmd: await self._client.send(cmd) logger.info( - f"Sent START_MEDIA_BUFFERING command to Asterisk WebSocket channel to enable audio buffering." + "Sent START_MEDIA_BUFFERING command to Asterisk WebSocket channel to enable audio buffering." ) except Exception as e: logger.error( @@ -126,19 +126,19 @@ async def write_audio_frame(self, frame: OutputAudioRawFrame) -> bool: if self._client.is_closing or not self._client.is_connected: logger.warning( - f"Cannot write audio frame because the WebSocket client is closing or already closed." + "Cannot write audio frame because the WebSocket client is closing or already closed." ) return False if not self._params.serializer: logger.error( - f"Serializer is not set in transport parameters. Cannot write audio frame." + "Serializer is not set in transport parameters. Cannot write audio frame." ) return False if self._flow_controller is None: logger.error( - f"Flow controller is not initialized. Cannot write audio frame." + "Flow controller is not initialized. Cannot write audio frame." ) return False @@ -151,7 +151,7 @@ async def write_audio_frame(self, frame: OutputAudioRawFrame) -> bool: try: payload = await self._params.serializer.serialize(frame) if payload: - if type(payload) == bytes: + if isinstance(payload, bytes): self._flow_controller(payload) return True else: @@ -161,7 +161,7 @@ async def write_audio_frame(self, frame: OutputAudioRawFrame) -> bool: return False else: logger.trace( - f"Serializer returned None or empty payload. Cannot write audio frame." + "Serializer returned None or empty payload. Cannot write audio frame." ) return False except Exception as e: diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/test_serializer.py b/tests/test_serializer.py new file mode 100644 index 0000000..80a6e94 --- /dev/null +++ b/tests/test_serializer.py @@ -0,0 +1,187 @@ +"""Unit tests for AsteriskFrameSerializer dispatch tables. + +These tests lock in the behavior of the dict-based frame/event dispatch +introduced to replace the previous `getattr(...)` reflection. The goal is +to make sure every handler registered in `__init__` is reachable through +the public `serialize` / `_handle_event` entry points, and that unknown +inputs fall through to the same "unhandled" branches as before. +""" + +import pytest + +from pipecat.frames.frames import ( + CancelFrame, + EndFrame, + InterruptionFrame, + InputDTMFFrame, + InputTransportMessageFrame, + StartFrame, +) + +from pipecat_asterisk.serializer.serializer import ( + AsteriskCommandFrame, + AsteriskFrameSerializer, +) + + +# --------------------------------------------------------------------------- +# serialize() — sync handlers +# --------------------------------------------------------------------------- + + +@pytest.mark.parametrize( + "frame, expected_cmd", + [ + (EndFrame(), "HANGUP"), + (CancelFrame(), "HANGUP"), + (InterruptionFrame(), "FLUSH_MEDIA"), + (AsteriskCommandFrame("START_MEDIA_BUFFERING"), "START_MEDIA_BUFFERING"), + (AsteriskCommandFrame("REPORT_QUEUE_DRAINED"), "REPORT_QUEUE_DRAINED"), + ], +) +async def test_serialize_dispatches_sync_handlers(frame, expected_cmd): + """Every entry in `_sync_frame_handlers` must be reachable via serialize(). + + We don't pin the exact wire format here (that's the protocol layer's + contract); we only check that the produced string contains the command + word, so this stays robust if the protocol formatting changes. + """ + serializer = AsteriskFrameSerializer() + result = await serializer.serialize(frame) + assert isinstance(result, str) + assert expected_cmd in result + + +async def test_serialize_returns_none_for_unhandled_frame(): + """Frames not registered in either dispatch table fall through to None. + + `StartFrame` is intentionally not in `_sync_frame_handlers` or + `_async_frame_handlers` — `setup()` handles it separately. + """ + serializer = AsteriskFrameSerializer() + result = await serializer.serialize( + StartFrame(audio_in_sample_rate=16000, audio_out_sample_rate=16000) + ) + assert result is None + + +# --------------------------------------------------------------------------- +# serialize() — async handler (OutputAudioRawFrame) +# --------------------------------------------------------------------------- + + +async def test_serialize_dispatches_async_audio_handler(): + """OutputAudioRawFrame goes through the async handler table. + + The handler short-circuits and returns the raw bytes unchanged when + the pipeline rate matches the Asterisk rate, so this also exercises + the no-resampling fast path. + """ + from pipecat.frames.frames import OutputAudioRawFrame + + serializer = AsteriskFrameSerializer() + # Match rates so the handler returns audio without going through a + # resampler that isn't initialized in this test. + serializer._pipeline_out_sample_rate = 16000 + serializer._asterisk_sample_rate = 16000 + + audio = b"\x00\x01" * 320 # 640 bytes — one slin16 frame + frame = OutputAudioRawFrame(audio=audio, sample_rate=16000, num_channels=1) + result = await serializer.serialize(frame) + assert result == audio + + +# --------------------------------------------------------------------------- +# serialize() — sync handlers must NOT be awaited +# --------------------------------------------------------------------------- + + +async def test_serialize_sync_handler_result_is_not_a_coroutine(): + """Regression guard for the sync/async split. + + The previous implementation called `inspect.isawaitable` on every + return value. After this refactor sync handlers are looked up in a + separate table and must return a plain string, never a coroutine. + """ + import inspect as _inspect + + serializer = AsteriskFrameSerializer() + result = await serializer.serialize(EndFrame()) + assert not _inspect.isawaitable(result) + assert isinstance(result, str) + + +# --------------------------------------------------------------------------- +# _handle_event() — registered events +# --------------------------------------------------------------------------- + + +def test_handle_event_dispatches_dtmf_end(): + serializer = AsteriskFrameSerializer() + frame = serializer._handle_event({"event": "DTMF_END", "digit": "5"}) + assert isinstance(frame, InputDTMFFrame) + + +def test_handle_event_dispatches_queue_drained(): + serializer = AsteriskFrameSerializer() + msg = {"event": "QUEUE_DRAINED"} + frame = serializer._handle_event(msg) + assert isinstance(frame, InputTransportMessageFrame) + assert frame.message == msg + + +def test_handle_event_dispatches_xoff_and_xon_to_none(): + """MEDIA_XOFF and MEDIA_XON handlers log but return None. + + The handlers are registered, so the dispatch must find them and not + fall through to the "unhandled" branch. + """ + serializer = AsteriskFrameSerializer() + assert serializer._handle_event({"event": "MEDIA_XOFF"}) is None + assert serializer._handle_event({"event": "MEDIA_XON"}) is None + + +# --------------------------------------------------------------------------- +# _handle_event() — missing / unknown event +# --------------------------------------------------------------------------- + + +def test_handle_event_missing_event_field_returns_none(): + serializer = AsteriskFrameSerializer() + assert serializer._handle_event({"not_event": "foo"}) is None + + +def test_handle_event_unknown_event_returns_none(): + serializer = AsteriskFrameSerializer() + assert serializer._handle_event({"event": "SOMETHING_NEW"}) is None + + +# --------------------------------------------------------------------------- +# Dispatch table integrity +# --------------------------------------------------------------------------- + + +def test_event_handler_table_keys_are_uppercase_event_names(): + """The dispatch keys must match Asterisk's wire-format event names exactly. + + The chan_websocket protocol uses uppercase event names like + `MEDIA_START`, `DTMF_END`, `QUEUE_DRAINED`. A lowercase key would + silently fall through the dispatch. + """ + serializer = AsteriskFrameSerializer() + assert all(k == k.upper() for k in serializer._event_handlers) + assert "MEDIA_START" in serializer._event_handlers + assert "DTMF_END" in serializer._event_handlers + assert "QUEUE_DRAINED" in serializer._event_handlers + + +def test_frame_handler_tables_have_no_overlap(): + """A frame class registered in both sync and async tables would + produce ambiguous dispatch (sync wins by lookup order). Catch that + here so the registration stays unambiguous. + """ + serializer = AsteriskFrameSerializer() + overlap = set(serializer._sync_frame_handlers) & set( + serializer._async_frame_handlers + ) + assert overlap == set()