wx44wx/pipecat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

tests(genesys): fix formatting

Browse Source
This commit is contained in:
Aleix Conchillo Flaqué
2026-01-28 09:15:42 -08:00
parent 01bbf61e0d
commit 55dadc9118
3 changed files with 77 additions and 73 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
tests/genesys/__init__.py
View File

@@ -1,6 +1 @@
# Copyright (c) 2026, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
"""Tests for Genesys AudioHook serializer."""

6
tests/genesys/conftest.py
View File

@@ -1,3 +1,9 @@
#
# Copyright (c) 2024-2026, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
"""Pytest fixtures for Genesys AudioHook serializer tests.
These fixtures provide sample AudioHook protocol messages for testing

139
tests/genesys/test_genesys_serializer.py
View File

@@ -1,3 +1,9 @@
#
# Copyright (c) 2024-2026, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
"""Tests for the Genesys AudioHook serializer."""
import json
@@ -16,7 +22,7 @@ class TestGenesysAudioHookSerializer:
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
@@ -33,7 +39,7 @@ class TestGenesysAudioHookSerializer:
start_paused=True,
)
serializer = GenesysAudioHookSerializer(params=params)
assert serializer.session_id != ""
# ==================== Response Creation Tests ====================
@@ -41,9 +47,9 @@ class TestGenesysAudioHookSerializer:
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
@@ -53,21 +59,21 @@ class TestGenesysAudioHookSerializer:
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
@@ -75,9 +81,9 @@ class TestGenesysAudioHookSerializer:
"""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
@@ -86,15 +92,15 @@ class TestGenesysAudioHookSerializer:
"""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"
"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
@@ -104,21 +110,21 @@ class TestGenesysAudioHookSerializer:
"""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"
@@ -126,26 +132,26 @@ class TestGenesysAudioHookSerializer:
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"
@@ -157,9 +163,9 @@ class TestGenesysAudioHookSerializer:
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"
@@ -170,9 +176,9 @@ class TestGenesysAudioHookSerializer:
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"
@@ -186,21 +192,23 @@ class TestGenesysAudioHookSerializer:
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):
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"
@@ -211,9 +219,9 @@ class TestGenesysAudioHookSerializer:
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"
@@ -222,9 +230,9 @@ class TestGenesysAudioHookSerializer:
"""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
@@ -234,16 +242,14 @@ class TestGenesysAudioHookSerializer:
"""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"
})
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"
@@ -255,14 +261,11 @@ class TestGenesysAudioHookSerializer:
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
})
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
@@ -270,10 +273,10 @@ class TestGenesysAudioHookSerializer:
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"
@@ -282,9 +285,9 @@ class TestGenesysAudioHookSerializer:
"""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
@@ -292,9 +295,9 @@ class TestGenesysAudioHookSerializer:
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"
@@ -303,9 +306,9 @@ class TestGenesysAudioHookSerializer:
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 ====================
@@ -314,9 +317,9 @@ class TestGenesysAudioHookSerializer:
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"
@@ -324,9 +327,9 @@ class TestGenesysAudioHookSerializer:
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 == "*"
@@ -334,9 +337,9 @@ class TestGenesysAudioHookSerializer:
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 == "#"
@@ -344,7 +347,7 @@ class TestGenesysAudioHookSerializer:
async def test_handle_dtmf_empty_digit(self):
"""Test handling a DTMF message without digit."""
serializer = GenesysAudioHookSerializer()
dtmf_msg = {
"version": "2",
"type": "dtmf",
@@ -352,9 +355,9 @@ class TestGenesysAudioHookSerializer:
"id": "test-session-123",
"parameters": {},
}
result = await serializer.deserialize(json.dumps(dtmf_msg))
assert result is None # No digit provided
# ==================== Sequence Number Tests ====================
@@ -362,11 +365,11 @@ class TestGenesysAudioHookSerializer:
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