diff --git a/src/pipecat/serializers/genesys.py b/src/pipecat/serializers/genesys.py index 48637653b..678d0ef10 100644 --- a/src/pipecat/serializers/genesys.py +++ b/src/pipecat/serializers/genesys.py @@ -1,8 +1,15 @@ -""" -Use with Genesys Audio Connector to connect Genesys Cloud Contact Center with Pipecat pipelines. +"""Genesys AudioHook Serializer for Pipecat. -This connector implements the Genesys AudioHook protocol for bidirectional -audio streaming between Genesys Cloud contact center and Pipecat pipelines. +This module provides a serializer for integrating Pipecat pipelines with +Genesys Cloud Contact Center via the AudioHook protocol. + +Features: +- Bidirectional audio streaming (PCMU μ-law at 8kHz) +- Automatic protocol handshake handling (open/opened, close/closed, ping/pong) +- Input/output variables for Architect flow integration +- DTMF event support +- Barge-in (interruption) events +- Pause/resume support for hold scenarios Protocol Reference: - https://developer.genesys.cloud/devapps/audiohook @@ -10,7 +17,7 @@ Protocol Reference: Audio Format: - PCMU (μ-law) at 8kHz sample rate (preferred) - L16 (16-bit linear PCM) at 8kHz also supported -- Mono or Stereo (external on left, internal on right) +- Mono (external channel) or Stereo (external on left, internal on right) """ import json @@ -76,10 +83,11 @@ class GenesysAudioHookSerializer(FrameSerializer): AudioHook protocol messages. It supports: - Bidirectional audio streaming (PCMU at 8kHz) - - Session lifecycle management (open, close, pause) - - Probe mode for health checks - - Custom configuration passthrough - - Event messaging back to Genesys Cloud + - Automatic protocol handshake (open/opened, close/closed, ping/pong) + - Session lifecycle management with pause/resume support + - Custom input/output variables for Architect flow integration + - DTMF event handling + - Barge-in events for interruption support The AudioHook protocol uses: - Text WebSocket frames for JSON control messages @@ -88,17 +96,30 @@ class GenesysAudioHookSerializer(FrameSerializer): Example usage: ```python serializer = GenesysAudioHookSerializer( - session_id="abc-123", params=GenesysAudioHookSerializer.InputParams( - channel=AudioHookChannel.BOTH, + channel=AudioHookChannel.EXTERNAL, + supported_languages=["en-US", "es-ES"], + selected_language="en-US", ) ) - # Use with WebSocket transport - transport = WebsocketServerTransport( - serializer=serializer, - ... + # Use with FastAPI WebSocket transport + transport = FastAPIWebsocketTransport( + websocket=websocket, + params=FastAPIWebsocketParams( + audio_in_enabled=True, + audio_out_enabled=True, + serializer=serializer, + audio_out_fixed_packet_size=1600, # Important: prevents 429 rate limiting from Genesys + ), ) + + # Access call information after connection + participant = serializer.participant # ani, dnis, etc. + input_vars = serializer.input_variables # Custom vars from Architect + + # Set output variables to return to Architect + serializer.set_output_variables({"intent": "billing", "resolved": True}) ``` Attributes: @@ -110,13 +131,16 @@ class GenesysAudioHookSerializer(FrameSerializer): class InputParams(BaseModel): """Configuration parameters for GenesysAudioHookSerializer. - Parameters: - genesys_sample_rate: Sample rate used by Genesys (8000 Hz). + Attributes: + genesys_sample_rate: Sample rate used by Genesys (default: 8000 Hz). sample_rate: Optional override for pipeline input sample rate. channel: Which audio channels to process (external, internal, both). media_format: Audio format (PCMU or L16). process_external: Whether to process external (customer) audio. process_internal: Whether to process internal (agent) audio. + supported_languages: List of language codes the bot supports (e.g., ["en-US", "es-ES"]). + selected_language: Default language code to use. + start_paused: Whether to start the session in paused state. """ genesys_sample_rate: int = 8000 @@ -125,23 +149,24 @@ class GenesysAudioHookSerializer(FrameSerializer): media_format: AudioHookMediaFormat = AudioHookMediaFormat.PCMU process_external: bool = True process_internal: bool = False + supported_languages: Optional[List[str]] = None + selected_language: Optional[str] = None + start_paused: bool = False def __init__( self, - session_id: Optional[str] = None, params: Optional[InputParams] = None, ): """Initialize the GenesysAudioHookSerializer. Args: - session_id: The AudioHook session ID (received in open message). params: Configuration parameters. """ self._params = params or GenesysAudioHookSerializer.InputParams() - self._session_id = session_id or "" self._genesys_sample_rate = self._params.genesys_sample_rate self._sample_rate = 0 # Pipeline input rate, set in setup() + self._session_id = str(uuid.uuid4()) # Use Pipecat's official resampler if needed (SOXR) # Only used for TTS output (16kHz → 8kHz), input goes without resampling @@ -161,11 +186,12 @@ class GenesysAudioHookSerializer(FrameSerializer): self._custom_config: Optional[Dict[str, Any]] = None self._media_info: Optional[List[Dict[str, Any]]] = None self._input_variables: Optional[Dict[str, Any]] = None # Custom input from Genesys + self._output_variables: Optional[Dict[str, Any]] = None # Custom output to Genesys @property def session_id(self) -> str: - """Get the current session ID.""" + """Get the Genesys AudioHook session ID generated by the serializer.""" return self._session_id @property @@ -193,6 +219,34 @@ class GenesysAudioHookSerializer(FrameSerializer): """Get custom input variables from the open message.""" return self._input_variables + @property + def output_variables(self) -> Optional[Dict[str, Any]]: + """Get custom output variables to send back to Genesys.""" + return self._output_variables + + def set_output_variables(self, variables: Dict[str, Any]) -> None: + """Set custom output variables to send back to Genesys on close. + + These variables will be included in the 'closed' response when Genesys + closes the connection, making them available in the Architect flow. + + Args: + variables: Dictionary of custom variables to send to Genesys. + + Example: + ```python + # During the conversation, collect data and set it + serializer.set_output_variables({ + "intent": "billing_inquiry", + "customer_verified": True, + "summary": "Customer asked about their bill", + "transfer_to": "billing_queue" + }) + ``` + """ + self._output_variables = variables + logger.debug(f"Output variables set: {variables}") + async def setup(self, frame: StartFrame): """Sets up the serializer with pipeline configuration. @@ -279,7 +333,7 @@ class GenesysAudioHookSerializer(FrameSerializer): start_paused: bool = False, supported_languages: Optional[List[str]] = None, selected_language: Optional[str] = None, - ) -> str: + ) -> Dict[str, Any]: """Create an 'opened' response message for the client. This should be sent in response to an 'open' message from Genesys. @@ -290,10 +344,11 @@ class GenesysAudioHookSerializer(FrameSerializer): selected_language: The selected language code. Returns: - JSON string of the opened response message. + Dictionary of the opened response message. """ # Build channels list based on configuration channels: list[str] = [] + if self._params.channel == AudioHookChannel.EXTERNAL: channels = ["external"] elif self._params.channel == AudioHookChannel.INTERNAL: @@ -325,59 +380,88 @@ class GenesysAudioHookSerializer(FrameSerializer): ) self._is_open = True + logger.debug(f"AudioHook session opened: {self._session_id}") - return json.dumps(msg) + return msg - def create_closed_response(self) -> str: + def create_closed_response( + self, + output_variables: Optional[Dict[str, Any]] = None, + ) -> Dict[str, Any]: """Create a 'closed' response message. This should be sent in response to a 'close' message from Genesys. + Args: + output_variables: Optional custom variables to pass back to Genesys. + These will be available in the Architect flow after the AudioHook + action completes. + Returns: - JSON string of the closed response message. + Dictionary of the closed response message. + + Example: + ```python + # Pass custom data back to Genesys + serializer.create_closed_response( + output_variables={ + "intent": "billing_inquiry", + "customer_verified": True, + "summary": "Customer asked about their bill" + } + ) + ``` """ - msg = self._create_message(AudioHookMessageType.CLOSED) + parameters: Optional[Dict[str, Any]] = None + + if output_variables: + parameters = {"outputVariables": output_variables} + + msg = self._create_message( + AudioHookMessageType.CLOSED, + parameters=parameters, + ) self._is_open = False logger.debug(f"AudioHook session closed: {self._session_id}") - return json.dumps(msg) + return msg - def create_pong_response(self) -> str: + def create_pong_response(self) -> Dict[str, Any]: """Create a 'pong' response message. This should be sent in response to a 'ping' message from Genesys. Returns: - JSON string of the pong response message. + Dictionary of the pong response message. """ msg = self._create_message(AudioHookMessageType.PONG) - return json.dumps(msg) + return msg - def create_resumed_response(self) -> str: + def create_resumed_response(self) -> Dict[str, Any]: """Create a 'resumed' response message. This should be sent in response to a 'pause' message when ready to resume. Returns: - JSON string of the resumed response message. + Dictionary of the resumed response message. """ msg = self._create_message(AudioHookMessageType.RESUMED) self._is_paused = False logger.debug(f"AudioHook session resumed: {self._session_id}") - return json.dumps(msg) + return msg - def create_barge_in_event(self) -> str: + def create_barge_in_event(self) -> Dict[str, Any]: """Create a barge-in event message. This notifies Genesys Cloud that the user has interrupted the bot's audio output. Genesys will stop any queued audio playback. Returns: - JSON string of the barge-in event message. + Dictionary of the barge-in event message. """ msg = self._create_message( AudioHookMessageType.EVENT, @@ -390,7 +474,7 @@ class GenesysAudioHookSerializer(FrameSerializer): logger.debug("🔇 Barge-in event sent to Genesys") - return json.dumps(msg) + return msg def create_disconnect_message( self, @@ -398,7 +482,7 @@ class GenesysAudioHookSerializer(FrameSerializer): action: str = "transfer", output_variables: Optional[Dict[str, Any]] = None, info: Optional[str] = None, - ) -> str: + ) -> Dict[str, Any]: """Create a 'disconnect' message to initiate session termination. Args: @@ -408,7 +492,7 @@ class GenesysAudioHookSerializer(FrameSerializer): info: Optional additional information. Returns: - JSON string of the disconnect message. + Dictionary of the disconnect message. """ parameters: Dict[str, Any] = {"reason": reason} @@ -427,14 +511,14 @@ class GenesysAudioHookSerializer(FrameSerializer): ) logger.debug(f"AudioHook disconnect: reason={reason}, action={action}") - return json.dumps(msg) + return msg def create_error_message( self, code: int, message: str, retryable: bool = False, - ) -> str: + ) -> Dict[str, Any]: """Create an 'error' message. Args: @@ -443,7 +527,7 @@ class GenesysAudioHookSerializer(FrameSerializer): retryable: Whether the operation can be retried. Returns: - JSON string of the error message. + Dictionary of the error message. """ parameters = { "code": code, @@ -457,24 +541,26 @@ class GenesysAudioHookSerializer(FrameSerializer): ) logger.error(f"AudioHook error: {code} - {message}") - return json.dumps(msg) + return msg async def serialize(self, frame: Frame) -> str | bytes | None: """Serializes a Pipecat frame to Genesys AudioHook format. Handles conversion of various frame types to AudioHook messages: - - AudioRawFrame -> Binary audio data - - EndFrame/CancelFrame -> Disconnect message + - AudioRawFrame -> Binary PCMU audio data (resampled to 8kHz) + - EndFrame/CancelFrame -> Disconnect message (JSON) + - InterruptionFrame -> Barge-in event (JSON) - OutputTransportMessageFrame -> Pass-through JSON Args: frame: The Pipecat frame to serialize. Returns: - Serialized data as string (JSON) or bytes (audio), or None. + Serialized data as string (JSON) or bytes (audio), or None if + the frame type is not handled or session is not open. """ if isinstance(frame, (EndFrame, CancelFrame)): - return self.create_disconnect_message(reason="completed") + return json.dumps(self.create_disconnect_message(reason="completed")) elif isinstance(frame, AudioRawFrame): if not self._is_open or self._is_paused: @@ -501,7 +587,7 @@ class GenesysAudioHookSerializer(FrameSerializer): return bytes(serialized_data) elif isinstance(frame, InterruptionFrame): - return self.create_barge_in_event() + return json.dumps(self.create_barge_in_event()) elif isinstance(frame, (OutputTransportMessageFrame, OutputTransportMessageUrgentFrame)): # Pass through custom JSON messages @@ -514,18 +600,23 @@ class GenesysAudioHookSerializer(FrameSerializer): """Deserializes Genesys AudioHook data to Pipecat frames. Handles: - - Binary data -> InputAudioRawFrame - - JSON text -> Protocol messages (open, close, ping, pause, etc.) + - Binary data -> InputAudioRawFrame (converted from PCMU to PCM) + - JSON 'open' -> OutputTransportMessageUrgentFrame with 'opened' response + - JSON 'close' -> OutputTransportMessageUrgentFrame with 'closed' response + - JSON 'ping' -> OutputTransportMessageUrgentFrame with 'pong' response + - JSON 'pause' -> Sets is_paused=True, returns None + - JSON 'dtmf' -> InputDTMFFrame + - JSON 'update' -> Updates participant info, returns None + - JSON 'error' -> Logs error, returns None - For control messages (open, close, ping, pause), this method handles - the protocol response internally and logs the events. The application - should monitor session state via the is_open and is_paused properties. + Protocol responses (opened, closed, pong) are returned as urgent frames + to be sent immediately through the transport. Args: - data: The raw WebSocket data from Genesys. + data: The raw WebSocket data from Genesys (binary audio or JSON text). Returns: - A Pipecat frame corresponding to the data, or None if handled internally. + A Pipecat frame to process, or None if handled internally. """ # Binary data = audio if isinstance(data, bytes): @@ -643,13 +734,22 @@ class GenesysAudioHookSerializer(FrameSerializer): async def _handle_open(self, message: Dict[str, Any]) -> Frame | None: """Handle an 'open' message from Genesys. - This initializes the session with metadata from Genesys Cloud. + This initializes the session with metadata from Genesys Cloud and + automatically responds with an 'opened' message using the configured + InputParams (supported_languages, selected_language, start_paused). + + Extracts and stores: + - session_id: The AudioHook session identifier + - conversation_id: The Genesys conversation ID + - participant: Caller info (ani, dnis, etc.) + - input_variables: Custom variables from Architect flow + - media_info: Audio configuration from Genesys Args: - message: The open message. + message: The open message from Genesys. Returns: - None (response should be sent via create_opened_response()). + OutputTransportMessageUrgentFrame with the 'opened' response. """ self._session_id = message.get("id", str(uuid.uuid4())) @@ -686,18 +786,25 @@ class GenesysAudioHookSerializer(FrameSerializer): f"conversation={self._conversation_id}, ani={ani}" ) - # Note: Application should call create_opened_response() to respond - return None + return OutputTransportMessageUrgentFrame(message=self.create_opened_response( + start_paused=self._params.start_paused, + supported_languages=self._params.supported_languages, + selected_language=self._params.selected_language + )) async def _handle_close(self, message: Dict[str, Any]) -> Frame | None: """Handle a 'close' message from Genesys. + Automatically responds with a 'closed' message. If output_variables + were set via set_output_variables(), they will be included in the + response and made available in the Architect flow. + Args: - message: The close message. + message: The close message from Genesys. Returns: - OutputTransportMessageUrgentFrame with the closed response, - or None if no response should be sent. + OutputTransportMessageUrgentFrame with the closed response + (includes outputVariables if set). """ params = message.get("parameters", {}) reason = params.get("reason", "unknown") @@ -709,22 +816,26 @@ class GenesysAudioHookSerializer(FrameSerializer): logger.info(f"Sending closed response to Genesys...") # Return as urgent frame to be sent through pipeline immediately - return OutputTransportMessageUrgentFrame(message=json.loads(self.create_closed_response())) + # Include any output variables that were set during the session + return OutputTransportMessageUrgentFrame( + message=self.create_closed_response(output_variables=self._output_variables) + ) async def _handle_ping(self, message: Dict[str, Any]) -> Frame | None: """Handle a 'ping' message from Genesys. + Automatically responds with a 'pong' message to maintain the connection. + Args: - message: The ping message. + message: The ping message from Genesys. Returns: - None if pong was sent directly via callback, otherwise OutputTransportMessageUrgentFrame with pong response. """ logger.info(f"Sending pong response to Genesys...") # Return as urgent frame to be sent through pipeline immediately - return OutputTransportMessageUrgentFrame(message=json.loads(self.create_pong_response())) + return OutputTransportMessageUrgentFrame(message=self.create_pong_response()) async def _handle_pause(self, message: Dict[str, Any]) -> Frame | None: """Handle a 'pause' message from Genesys. diff --git a/tests/test_genesys_serializer.py b/tests/test_genesys_serializer.py new file mode 100644 index 000000000..65f624c9a --- /dev/null +++ b/tests/test_genesys_serializer.py @@ -0,0 +1,375 @@ +"""Tests for the Genesys AudioHook serializer.""" + +import json +import pytest + +from pipecat.frames.frames import InputDTMFFrame, OutputTransportMessageUrgentFrame + +from pipecat.serializers.genesys import ( + GenesysAudioHookSerializer, + AudioHookChannel +) + + +class TestGenesysAudioHookSerializer: + """Tests for GenesysAudioHookSerializer.""" + + # ==================== Initialization Tests ==================== + + def test_create_serializer_default_params(self): + """Test creating serializer with default parameters.""" + serializer = GenesysAudioHookSerializer() + + # session_id is auto-generated as UUID + assert serializer.session_id != "" + assert len(serializer.session_id) == 36 # UUID format + assert serializer.is_open is False + assert serializer.is_paused is False + + def test_create_serializer_with_custom_params(self): + """Test creating serializer with custom parameters.""" + params = GenesysAudioHookSerializer.InputParams( + channel=AudioHookChannel.BOTH, + sample_rate=16000, + supported_languages=["es-ES", "en-US"], + selected_language="es-ES", + start_paused=True, + ) + serializer = GenesysAudioHookSerializer(params=params) + + assert serializer.session_id != "" + + # ==================== Response Creation Tests ==================== + + def test_create_opened_response(self): + """Test creating an opened response message.""" + serializer = GenesysAudioHookSerializer() + + msg = serializer.create_opened_response() + + assert msg["type"] == "opened" + assert msg["version"] == "2" + assert msg["id"] == serializer.session_id + assert "parameters" in msg + assert serializer.is_open is True + + def test_create_opened_response_with_languages(self): + """Test creating an opened response with language options.""" + serializer = GenesysAudioHookSerializer() + + msg = serializer.create_opened_response( + supported_languages=["es", "en", "fr"], + selected_language="es", + ) + + assert msg["parameters"]["supportedLanguages"] == ["es", "en", "fr"] + assert msg["parameters"]["selectedLanguage"] == "es" + + def test_create_pong_response(self): + """Test creating a pong response message.""" + serializer = GenesysAudioHookSerializer() + + msg = serializer.create_pong_response() + + assert msg["type"] == "pong" + assert msg["id"] == serializer.session_id + + def test_create_closed_response(self): + """Test creating a closed response message.""" + serializer = GenesysAudioHookSerializer() + serializer._is_open = True + + msg = serializer.create_closed_response() + + assert msg["type"] == "closed" + assert serializer.is_open is False + assert "parameters" not in msg # No parameters when no output_variables + + def test_create_closed_response_with_output_variables(self): + """Test creating a closed response with custom output variables.""" + serializer = GenesysAudioHookSerializer() + serializer._is_open = True + + msg = serializer.create_closed_response( + output_variables={ + "intent": "billing_inquiry", + "customer_verified": True, + "summary": "Customer asked about their bill" + } + ) + + assert msg["type"] == "closed" + assert msg["parameters"]["outputVariables"]["intent"] == "billing_inquiry" + assert msg["parameters"]["outputVariables"]["customer_verified"] is True + assert msg["parameters"]["outputVariables"]["summary"] == "Customer asked about their bill" + + def test_create_resumed_response(self): + """Test creating a resumed response message.""" + serializer = GenesysAudioHookSerializer() + serializer._is_paused = True + + msg = serializer.create_resumed_response() + + assert msg["type"] == "resumed" + assert serializer.is_paused is False + + def test_create_disconnect_message(self): + """Test creating a disconnect message.""" + serializer = GenesysAudioHookSerializer() + + msg = serializer.create_disconnect_message( + reason="completed", + action="transfer", + ) + + assert msg["type"] == "disconnect" + assert msg["parameters"]["reason"] == "completed" + assert msg["parameters"]["outputVariables"]["action"] == "transfer" + + def test_create_disconnect_message_with_output_variables(self): + """Test creating a disconnect message with custom output variables.""" + serializer = GenesysAudioHookSerializer() + + msg = serializer.create_disconnect_message( + reason="completed", + action="finished", + output_variables={"result": "success", "code": "123"}, + ) + + assert msg["parameters"]["outputVariables"]["result"] == "success" + assert msg["parameters"]["outputVariables"]["code"] == "123" + + def test_create_error_message(self): + """Test creating an error message.""" + serializer = GenesysAudioHookSerializer() + + msg = serializer.create_error_message( + code=500, + message="Internal error", + retryable=True, + ) + + assert msg["type"] == "error" + assert msg["parameters"]["code"] == 500 + assert msg["parameters"]["message"] == "Internal error" + assert msg["parameters"]["retryable"] is True + + # ==================== Message Handling Tests ==================== + + @pytest.mark.asyncio + async def test_handle_open_message(self, sample_open_message): + """Test handling an open message returns opened frame.""" + serializer = GenesysAudioHookSerializer() + + result = await serializer.deserialize(json.dumps(sample_open_message)) + + # Now returns OutputTransportMessageUrgentFrame with opened response + assert isinstance(result, OutputTransportMessageUrgentFrame) + assert result.message["type"] == "opened" + assert serializer.session_id == "test-session-123" + assert serializer.conversation_id == "conv-456" + + @pytest.mark.asyncio + async def test_handle_open_message_extracts_participant(self, sample_open_message): + """Test that open message extracts participant info.""" + serializer = GenesysAudioHookSerializer() + + await serializer.deserialize(json.dumps(sample_open_message)) + + assert serializer.participant is not None + assert serializer.participant["ani"] == "+1234567890" + assert serializer.participant["dnis"] == "+0987654321" + + @pytest.mark.asyncio + async def test_handle_open_message_uses_params(self, sample_open_message): + """Test that open message uses InputParams for response.""" + params = GenesysAudioHookSerializer.InputParams( + supported_languages=["es-ES", "en-US"], + selected_language="es-ES", + start_paused=True, + ) + serializer = GenesysAudioHookSerializer(params=params) + + result = await serializer.deserialize(json.dumps(sample_open_message)) + + assert isinstance(result, OutputTransportMessageUrgentFrame) + assert result.message["parameters"]["supportedLanguages"] == ["es-ES", "en-US"] + assert result.message["parameters"]["selectedLanguage"] == "es-ES" + assert result.message["parameters"]["startPaused"] is True + + @pytest.mark.asyncio + async def test_handle_open_message_extracts_input_variables(self, sample_open_message_with_input_variables): + """Test that open message extracts inputVariables from Genesys.""" + serializer = GenesysAudioHookSerializer() + + await serializer.deserialize(json.dumps(sample_open_message_with_input_variables)) + + assert serializer.input_variables is not None + assert serializer.input_variables["customer_id"] == "cust-789" + assert serializer.input_variables["queue_name"] == "billing" + assert serializer.input_variables["priority"] == "high" + assert serializer.input_variables["language"] == "es-ES" + + @pytest.mark.asyncio + async def test_handle_ping_message(self, sample_ping_message): + """Test handling a ping message returns pong frame.""" + serializer = GenesysAudioHookSerializer() + + result = await serializer.deserialize(json.dumps(sample_ping_message)) + + assert isinstance(result, OutputTransportMessageUrgentFrame) + assert result.message["type"] == "pong" + + @pytest.mark.asyncio + async def test_handle_close_message(self, sample_close_message): + """Test handling a close message returns closed frame.""" + serializer = GenesysAudioHookSerializer() + serializer._is_open = True + + result = await serializer.deserialize(json.dumps(sample_close_message)) + + assert isinstance(result, OutputTransportMessageUrgentFrame) + assert result.message["type"] == "closed" + assert serializer.is_open is False + + @pytest.mark.asyncio + async def test_handle_close_message_includes_output_variables(self, sample_close_message): + """Test that close response includes output variables when set.""" + serializer = GenesysAudioHookSerializer() + serializer._is_open = True + + # Set output variables before close + serializer.set_output_variables({ + "intent": "support", + "resolved": True, + "transfer_to": "agent_queue" + }) + + result = await serializer.deserialize(json.dumps(sample_close_message)) + + assert isinstance(result, OutputTransportMessageUrgentFrame) + assert result.message["type"] == "closed" + assert result.message["parameters"]["outputVariables"]["intent"] == "support" + assert result.message["parameters"]["outputVariables"]["resolved"] is True + assert result.message["parameters"]["outputVariables"]["transfer_to"] == "agent_queue" + + # ==================== Output Variables Tests ==================== + + def test_set_output_variables(self): + """Test setting output variables.""" + serializer = GenesysAudioHookSerializer() + + assert serializer.output_variables is None + + serializer.set_output_variables({ + "intent": "billing", + "score": 0.95 + }) + + assert serializer.output_variables is not None + assert serializer.output_variables["intent"] == "billing" + assert serializer.output_variables["score"] == 0.95 + + def test_set_output_variables_overwrites(self): + """Test that setting output variables overwrites previous values.""" + serializer = GenesysAudioHookSerializer() + + serializer.set_output_variables({"first": "value"}) + serializer.set_output_variables({"second": "value"}) + + assert "first" not in serializer.output_variables + assert serializer.output_variables["second"] == "value" + + @pytest.mark.asyncio + async def test_handle_pause_message(self, sample_pause_message): + """Test handling a pause message.""" + serializer = GenesysAudioHookSerializer() + serializer._is_open = True + + result = await serializer.deserialize(json.dumps(sample_pause_message)) + + assert result is None # Pause is handled internally + assert serializer.is_paused is True + + @pytest.mark.asyncio + async def test_handle_update_message(self, sample_update_message): + """Test handling an update message.""" + serializer = GenesysAudioHookSerializer() + + result = await serializer.deserialize(json.dumps(sample_update_message)) + + assert result is None # Update is handled internally + assert serializer.participant is not None + assert serializer.participant["name"] == "John Doe" + + @pytest.mark.asyncio + async def test_handle_error_message(self, sample_error_message): + """Test handling an error message.""" + serializer = GenesysAudioHookSerializer() + + result = await serializer.deserialize(json.dumps(sample_error_message)) + + assert result is None # Error is logged but returns None + + # ==================== DTMF Tests ==================== + + @pytest.mark.asyncio + async def test_handle_dtmf_digit(self, sample_dtmf_message): + """Test handling a DTMF digit message.""" + serializer = GenesysAudioHookSerializer() + + result = await serializer.deserialize(json.dumps(sample_dtmf_message)) + + assert isinstance(result, InputDTMFFrame) + assert result.button.value == "5" + + @pytest.mark.asyncio + async def test_handle_dtmf_star(self, sample_dtmf_star_message): + """Test handling a DTMF star (*) message.""" + serializer = GenesysAudioHookSerializer() + + result = await serializer.deserialize(json.dumps(sample_dtmf_star_message)) + + assert isinstance(result, InputDTMFFrame) + assert result.button.value == "*" + + @pytest.mark.asyncio + async def test_handle_dtmf_hash(self, sample_dtmf_hash_message): + """Test handling a DTMF hash (#) message.""" + serializer = GenesysAudioHookSerializer() + + result = await serializer.deserialize(json.dumps(sample_dtmf_hash_message)) + + assert isinstance(result, InputDTMFFrame) + assert result.button.value == "#" + + @pytest.mark.asyncio + async def test_handle_dtmf_empty_digit(self): + """Test handling a DTMF message without digit.""" + serializer = GenesysAudioHookSerializer() + + dtmf_msg = { + "version": "2", + "type": "dtmf", + "seq": 6, + "id": "test-session-123", + "parameters": {}, + } + + result = await serializer.deserialize(json.dumps(dtmf_msg)) + + assert result is None # No digit provided + + # ==================== Sequence Number Tests ==================== + + def test_sequence_numbers_increment(self): + """Test that sequence numbers increment correctly.""" + serializer = GenesysAudioHookSerializer() + + response1 = serializer.create_pong_response() + response2 = serializer.create_pong_response() + response3 = serializer.create_pong_response() + + assert response1["seq"] == 1 + assert response2["seq"] == 2 + assert response3["seq"] == 3