Add comprehensive tests covering e2e, audio modules, and client utilities

Author: NeuralAudioAI

tests/__init__.py

@@ -0,0 +1,79 @@
+import os
+import time
+import asyncio
+
+import pytest
+from neuralaudio import NeuralAudio
+from neuralaudio.conversational_ai.conversation import Conversation, ClientTools
+from neuralaudio.conversational_ai.default_audio_interface import DefaultAudioInterface
+
+
+@pytest.mark.skipif(os.getenv("CI") == "true", reason="Skip live conversation test in CI environment")
+def test_live_conversation():
+    """Test a live conversation with actual audio I/O"""
+
+    api_key = os.getenv("NEURALAUDIO_API_KEY")
+    if not api_key:
+        raise ValueError("NEURALAUDIO_API_KEY environment variable missing.")
+
+    agent_id = os.getenv("AGENT_ID")
+    if not api_key or not agent_id:
+        raise ValueError("AGENT_ID environment variable missing.")
+
+    client = NeuralAudio(api_key=api_key)
+
+    # Create conversation handlers
+    def on_agent_response(text: str):
+        print(f"Agent: {text}")
+
+    def on_user_transcript(text: str):
+        print(f"You: {text}")
+
+    def on_latency(ms: int):
+        print(f"Latency: {ms}ms")
+
+    # Initialize client tools
+    client_tools = ClientTools()
+
+    def test(parameters):
+        print("Sync tool called with parameters:", parameters)
+        return "Tool called successfully"
+
+    async def test_async(parameters):
+        # Simulate some async work
+        await asyncio.sleep(10)
+        print("Async tool called with parameters:", parameters)
+        return "Tool called successfully"
+
+    client_tools.register("test", test)
+    client_tools.register("test_async", test_async, is_async=True)
+
+    # Initialize conversation
+    conversation = Conversation(
+        client=client,
+        agent_id=agent_id,
+        requires_auth=False,
+        audio_interface=DefaultAudioInterface(),
+        callback_agent_response=on_agent_response,
+        callback_user_transcript=on_user_transcript,
+        callback_latency_measurement=on_latency,
+        client_tools=client_tools,
+    )
+
+    # Start the conversation
+    conversation.start_session()
+
+    # Let it run for 100 seconds
+    time.sleep(100)
+
+    # End the conversation
+    conversation.end_session()
+    conversation.wait_for_session_end()
+
+    # Get the conversation ID for reference
+    conversation_id = conversation._conversation_id
+    print(f"Conversation ID: {conversation_id}")
+
+
+if __name__ == "__main__":
+    test_live_conversation()

tests/e2e_test_convai.py

@@ -0,0 +1,32 @@
+from neuralaudio import play
+from neuralaudio.client import NeuralAudio
+
+from .utils import IN_GITHUB, DEFAULT_VOICE_FILE
+
+
+def test_audio_isolation() -> None:
+    """Test basic audio isolation."""
+    client = NeuralAudio()
+    audio_file = open(DEFAULT_VOICE_FILE, "rb")
+    try:
+        audio_stream = client.audio_isolation.audio_isolation(audio=audio_file)
+        audio = b"".join(chunk for chunk in audio_stream)
+        assert isinstance(audio, bytes), "Combined audio should be bytes"
+        if not IN_GITHUB:
+            play(audio)
+    finally:
+        audio_file.close()
+
+
+def test_audio_isolation_as_stream():
+    """Test audio isolation with streaming."""
+    client = NeuralAudio()
+    audio_file = open(DEFAULT_VOICE_FILE, "rb")
+    try:
+        audio_stream = client.audio_isolation.audio_isolation_stream(audio=audio_file)
+        audio = b"".join(chunk for chunk in audio_stream)
+        assert isinstance(audio, bytes), "Combined audio should be bytes"
+        if not IN_GITHUB:
+            play(audio)
+    finally:
+        audio_file.close()

tests/fixtures/voice_sample.mp3

@@ -0,0 +1,165 @@
+from unittest.mock import MagicMock, patch
+from neuralaudio.conversational_ai.conversation import Conversation, AudioInterface, ConversationInitiationData
+import json
+import time
+
+
+class MockAudioInterface(AudioInterface):
+    def start(self, input_callback):
+        print("Audio interface started")
+        self.input_callback = input_callback
+
+    def stop(self):
+        print("Audio interface stopped")
+
+    def output(self, audio):
+        print(f"Would play audio of length: {len(audio)} bytes")
+
+    def interrupt(self):
+        print("Audio interrupted")
+
+
+# Add test constants and helpers at module level
+TEST_CONVERSATION_ID = "test123"
+TEST_AGENT_ID = "test_agent"
+
+
+def create_mock_websocket(messages=None):
+    """Helper to create a mock websocket with predefined responses"""
+    mock_ws = MagicMock()
+
+    if messages is None:
+        messages = [
+            {
+                "type": "conversation_initiation_metadata",
+                "conversation_initiation_metadata_event": {"conversation_id": TEST_CONVERSATION_ID},
+            },
+            {"type": "agent_response", "agent_response_event": {"agent_response": "Hello there!"}},
+        ]
+
+    def response_generator():
+        for msg in messages:
+            yield json.dumps(msg)
+        while True:
+            yield '{"type": "keep_alive"}'
+
+    mock_ws.recv = MagicMock(side_effect=response_generator())
+    return mock_ws
+
+
+def test_conversation_basic_flow():
+    # Mock setup
+    mock_ws = create_mock_websocket()
+    mock_client = MagicMock()
+    agent_response_callback = MagicMock()
+
+    # Setup the conversation
+    conversation = Conversation(
+        client=mock_client,
+        agent_id=TEST_AGENT_ID,
+        requires_auth=False,
+        audio_interface=MockAudioInterface(),
+        callback_agent_response=agent_response_callback,
+    )
+
+    # Run the test
+    with patch("neuralaudio.conversational_ai.conversation.connect") as mock_connect:
+        mock_connect.return_value.__enter__.return_value = mock_ws
+        conversation.start_session()
+
+        # Add a wait for the callback to be called
+        timeout = 5  # 5 seconds timeout
+        start_time = time.time()
+        while not agent_response_callback.called and time.time() - start_time < timeout:
+            time.sleep(0.1)
+
+        conversation.end_session()
+        conversation.wait_for_session_end()
+
+    # Assertions
+    expected_init_message = {
+        "type": "conversation_initiation_client_data",
+        "custom_llm_extra_body": {},
+        "conversation_config_override": {},
+        "dynamic_variables": {},
+    }
+    mock_ws.send.assert_any_call(json.dumps(expected_init_message))
+    agent_response_callback.assert_called_once_with("Hello there!")
+    assert conversation._conversation_id == TEST_CONVERSATION_ID
+
+
+def test_conversation_with_auth():
+    # Mock setup
+    mock_client = MagicMock()
+    mock_client.conversational_ai.get_signed_url.return_value.signed_url = "wss://signed.url"
+    mock_ws = create_mock_websocket(
+        [
+            {
+                "type": "conversation_initiation_metadata",
+                "conversation_initiation_metadata_event": {"conversation_id": TEST_CONVERSATION_ID},
+            }
+        ]
+    )
+
+    conversation = Conversation(
+        client=mock_client,
+        agent_id=TEST_AGENT_ID,
+        requires_auth=True,
+        audio_interface=MockAudioInterface(),
+    )
+
+    # Run the test
+    with patch("neuralaudio.conversational_ai.conversation.connect") as mock_connect:
+        mock_connect.return_value.__enter__.return_value = mock_ws
+        conversation.start_session()
+        conversation.end_session()
+        conversation.wait_for_session_end()
+
+    # Assertions
+    mock_client.conversational_ai.get_signed_url.assert_called_once_with(agent_id=TEST_AGENT_ID)
+
+def test_conversation_with_dynamic_variables():
+    # Mock setup
+    mock_ws = create_mock_websocket()
+    mock_client = MagicMock()
+    agent_response_callback = MagicMock()
+
+    dynamic_variables = {"name": "angelo"}
+    config = ConversationInitiationData(dynamic_variables=dynamic_variables)
+
+    # Setup the conversation
+    conversation = Conversation(
+        client=mock_client,
+        config=config,
+        agent_id=TEST_AGENT_ID,
+        requires_auth=False,
+        audio_interface=MockAudioInterface(),
+        callback_agent_response=agent_response_callback,
+    )
+
+    # Run the test
+    with patch("neuralaudio.conversational_ai.conversation.connect") as mock_connect:
+        mock_connect.return_value.__enter__.return_value = mock_ws
+        conversation.start_session()
+
+        # Add a wait for the callback to be called
+        timeout = 5  # 5 seconds timeout
+        start_time = time.time()
+        while not agent_response_callback.called and time.time() - start_time < timeout:
+            time.sleep(0.1)
+
+        conversation.end_session()
+        conversation.wait_for_session_end()
+
+    # Assertions
+    expected_init_message = {
+        "type": "conversation_initiation_client_data",
+        "custom_llm_extra_body": {},
+        "conversation_config_override": {},
+        "dynamic_variables": {
+            "name": "angelo"
+        },
+    }
+    mock_ws.send.assert_any_call(json.dumps(expected_init_message))
+    agent_response_callback.assert_called_once_with("Hello there!")
+    assert conversation._conversation_id == TEST_CONVERSATION_ID

tests/test_audio_isolation.py

@@ -0,0 +1,8 @@
+from neuralaudio import GetSpeechHistoryResponse, NeuralAudio
+
+
+def test_history():
+    client = NeuralAudio()
+    page_size = 5
+    history = client.history.get_all(page_size=page_size)
+    assert isinstance(history, GetSpeechHistoryResponse)

tests/test_convai.py

@@ -0,0 +1,9 @@
+from neuralaudio import Model
+from neuralaudio.client import NeuralAudio
+
+
+def test_models_get_all():
+    client = NeuralAudio()
+    models = client.models.get_all()
+    assert len(models) > 0
+    assert isinstance(models[0], Model)

tests/test_history.py

@@ -0,0 +1,31 @@
+from neuralaudio import play
+from neuralaudio.client import NeuralAudio
+
+from .utils import IN_GITHUB, DEFAULT_VOICE, DEFAULT_VOICE_FILE
+
+
+def test_sts() -> None:
+    """Test basic speech-to-speech generation."""
+    client = NeuralAudio()
+    audio_file = open(DEFAULT_VOICE_FILE, "rb")
+    try:
+        audio_stream = client.speech_to_speech.convert(voice_id=DEFAULT_VOICE, audio=audio_file)
+        audio = b"".join(chunk for chunk in audio_stream)
+        assert isinstance(audio, bytes), "Combined audio should be bytes"
+        if not IN_GITHUB:
+            play(audio)
+    finally:
+        audio_file.close()
+
+
+def test_sts_as_stream():
+    client = NeuralAudio()
+    audio_file = open(DEFAULT_VOICE_FILE, "rb")
+    try:
+        audio_stream = client.speech_to_speech.convert_as_stream(voice_id=DEFAULT_VOICE, audio=audio_file)
+        audio = b"".join(chunk for chunk in audio_stream)
+        assert isinstance(audio, bytes), "Combined audio should be bytes"
+        if not IN_GITHUB:
+            play(audio)
+    finally:
+        audio_file.close()

tests/test_models.py

@@ -0,0 +1,44 @@
+import pytest
+from neuralaudio.client import AsyncNeuralAudio, NeuralAudio
+
+from .utils import DEFAULT_VOICE_FILE
+
+DEFAULT_EXT_AUDIO = "https://storage.googleapis.com/neuralaudio-public-cdn/audio/marketing/nicole.mp3"
+
+
+
+@pytest.mark.asyncio
+async def test_stt_convert():
+    """Test basic speech-to-text conversion."""
+    client = NeuralAudio()
+    
+    audio_file = open(DEFAULT_VOICE_FILE, "rb")
+    
+    transcription = client.speech_to_text.convert(
+        file=audio_file,
+        model_id="scribe_v1"
+    )
+    
+    assert isinstance(transcription.text, str)
+    assert len(transcription.text) > 0
+    assert isinstance(transcription.words, list)
+    assert len(transcription.words) > 0
+
+@pytest.mark.asyncio
+async def test_stt_convert_as_stream():
+    """Test speech-to-text conversion as stream."""
+    client = AsyncNeuralAudio()
+    
+    audio_file = open(DEFAULT_VOICE_FILE, "rb")
+    
+    stream = client.speech_to_text.convert_as_stream(
+        file=audio_file,
+        model_id="scribe_v1"
+    )
+    
+    transcription_text = ""
+    async for chunk in stream:
+        assert isinstance(chunk.text, str)
+        transcription_text += chunk.text
+    
+    assert len(transcription_text) > 0