adds tests

tests/streaming/action/test_dtmf.py

@@ -1,5 +1,10 @@
+import asyncio
+import base64
+import json
+
 import pytest
 from aioresponses import aioresponses
+from pytest_mock import MockerFixture
 
 from tests.fakedata.id import generate_uuid
 from vocode.streaming.action.dtmf import (
@@ -12,9 +17,15 @@
     TwilioPhoneConversationActionInput,
     VonagePhoneConversationActionInput,
 )
+from vocode.streaming.models.audio import AudioEncoding
 from vocode.streaming.models.telephony import VonageConfig
+from vocode.streaming.output_device.twilio_output_device import TwilioOutputDevice
 from vocode.streaming.utils import create_conversation_id
-from vocode.streaming.utils.state_manager import VonagePhoneConversationStateManager
+from vocode.streaming.utils.dtmf_utils import generate_dtmf_tone
+from vocode.streaming.utils.state_manager import (
+    TwilioPhoneConversationStateManager,
+    VonagePhoneConversationStateManager,
+)
 
 
 @pytest.mark.asyncio
@@ -59,12 +70,74 @@ async def test_vonage_dtmf_press_digits(mocker, mock_env):
         assert action_output.response.success is True
 
 
+@pytest.fixture
+def mock_twilio_output_device(mocker: MockerFixture):
+    output_device = TwilioOutputDevice()
+    output_device.ws = mocker.AsyncMock()
+    output_device.stream_sid = "stream_sid"
+    return output_device
+
+
+@pytest.fixture
+def mock_twilio_phone_conversation(
+    mocker: MockerFixture, mock_twilio_output_device: TwilioOutputDevice
+):
+    twilio_phone_conversation_mock = mocker.MagicMock()
+    twilio_phone_conversation_mock.output_device = mock_twilio_output_device
+    return twilio_phone_conversation_mock
+
+
 @pytest.mark.asyncio
-async def test_twilio_dtmf_press_digits(mocker, mock_env):
+async def test_twilio_dtmf_press_digits(
+    mocker, mock_env, mock_twilio_phone_conversation, mock_twilio_output_device: TwilioOutputDevice
+):
     action = TwilioDTMF(action_config=DTMFVocodeActionConfig())
     digits = "1234"
     twilio_sid = "twilio_sid"
 
+    action.attach_conversation_state_manager(
+        TwilioPhoneConversationStateManager(mock_twilio_phone_conversation)
+    )
+
+    action_output = await action.run(
+        action_input=TwilioPhoneConversationActionInput(
+            action_config=DTMFVocodeActionConfig(),
+            conversation_id=create_conversation_id(),
+            params=DTMFParameters(buttons=digits),
+            twilio_sid=twilio_sid,
+        )
+    )
+
+    mock_twilio_output_device.start()
+    while not mock_twilio_output_device._twilio_events_queue.empty():
+        await asyncio.sleep(0.1)
+
+    assert action_output.response.success
+    mock_twilio_output_device.terminate()
+
+    for digit, call in zip(digits, mock_twilio_output_device.ws.send_text.call_args_list):
+        expected_dtmf = generate_dtmf_tone(
+            digit, sampling_rate=8000, audio_encoding=AudioEncoding.MULAW
+        )
+        media_message = json.loads(call[0][0])
+        assert media_message["streamSid"] == mock_twilio_output_device.stream_sid
+        assert media_message["media"] == {
+            "payload": base64.b64encode(expected_dtmf).decode("utf-8")
+        }
+
+
+@pytest.mark.asyncio
+async def test_twilio_dtmf_failure(
+    mocker, mock_env, mock_twilio_phone_conversation, mock_twilio_output_device: TwilioOutputDevice
+):
+    action = TwilioDTMF(action_config=DTMFVocodeActionConfig())
+    digits = "****"
+    twilio_sid = "twilio_sid"
+
+    action.attach_conversation_state_manager(
+        TwilioPhoneConversationStateManager(mock_twilio_phone_conversation)
+    )
+
     action_output = await action.run(
         action_input=TwilioPhoneConversationActionInput(
             action_config=DTMFVocodeActionConfig(),
@@ -74,4 +147,4 @@ async def test_twilio_dtmf_press_digits(mocker, mock_env):
         )
     )
 
-    assert action_output.response.success is False  # Twilio does not support DTMF
+    assert not action_output.response.success

vocode/streaming/action/dtmf.py

@@ -92,5 +92,5 @@ async def run(self, action_input: ActionInput[DTMFParameters]) -> ActionOutput[D
         )
         return ActionOutput(
             action_type=action_input.action_config.type,
-            response=DTMFResponse(success=False),
+            response=DTMFResponse(success=True),
         )

vocode/streaming/utils/dtmf_utils.py

@@ -6,6 +6,7 @@
 from vocode.streaming.models.audio import AudioEncoding
 
 DEFAULT_DTMF_TONE_LENGTH_SECONDS = 0.3
+MAX_INT = 32767
 
 
 class KeypadEntry(str, Enum):
@@ -45,7 +46,7 @@ def generate_dtmf_tone(
     t = np.linspace(0, duration_seconds, int(sampling_rate * duration_seconds), endpoint=False)
     tone = np.sin(2 * np.pi * f1 * t) + np.sin(2 * np.pi * f2 * t)
     tone = tone / np.max(np.abs(tone))  # Normalize to [-1, 1]
-    pcm = (tone * 32767).astype(np.int16).tobytes()
+    pcm = (tone * MAX_INT).astype(np.int16).tobytes()
     if audio_encoding == AudioEncoding.MULAW:
         return audioop.lin2ulaw(pcm, 2)
     else: