test(bidirectional): Add integ test

.gitignore

@@ -13,3 +13,4 @@ dist
 repl_state
 .kiro
 uv.lock
+.audio_cache

tests_integ/bidirectional_streaming/__init__.py

@@ -0,0 +1 @@
+"""Integration tests for bidirectional streaming agents."""

tests_integ/bidirectional_streaming/conftest.py

@@ -0,0 +1,28 @@
+"""Pytest fixtures for bidirectional streaming integration tests."""
+
+import logging
+
+import pytest
+
+from .utils.audio_generator import AudioGenerator
+
+logger = logging.getLogger(__name__)
+
+
+@pytest.fixture(scope="session")
+def audio_generator():
+    """Provide AudioGenerator instance for tests."""
+    return AudioGenerator(region="us-east-1")
+
+
+@pytest.fixture(autouse=True)
+def setup_logging():
+    """Configure logging for tests."""
+    logging.basicConfig(
+        level=logging.DEBUG,
+        format="%(levelname)s | %(name)s | %(message)s",
+    )
+    # Reduce noise from some loggers
+    logging.getLogger("boto3").setLevel(logging.WARNING)
+    logging.getLogger("botocore").setLevel(logging.WARNING)
+    logging.getLogger("urllib3").setLevel(logging.WARNING)

tests_integ/bidirectional_streaming/test_bidirectional_agent.py

@@ -0,0 +1,202 @@
+"""Parameterized integration tests for bidirectional streaming.
+
+Tests fundamental functionality across multiple model providers (Nova Sonic, OpenAI, etc.)
+including multi-turn conversations, audio I/O, text transcription, and tool execution.
+
+This demonstrates the provider-agnostic design of the bidirectional streaming system.
+"""
+
+import asyncio
+import logging
+import os
+
+import pytest
+from strands_tools import calculator
+
+from strands.experimental.bidirectional_streaming.agent.agent import BidirectionalAgent
+from strands.experimental.bidirectional_streaming.models.novasonic import NovaSonicBidirectionalModel
+from strands.experimental.bidirectional_streaming.models.openai import OpenAIRealtimeBidirectionalModel
+from strands.experimental.bidirectional_streaming.models.gemini_live import GeminiLiveBidirectionalModel
+
+from .utils.test_context import BidirectionalTestContext
+
+logger = logging.getLogger(__name__)
+
+
+# Provider configurations
+PROVIDER_CONFIGS = {
+    "nova_sonic": {
+        "model_class": NovaSonicBidirectionalModel,
+        "model_kwargs": {"region": "us-east-1"},
+        "silence_duration": 2.5,  # Nova Sonic needs 2+ seconds of silence
+        "env_vars": ["AWS_ACCESS_KEY_ID", "AWS_SECRET_ACCESS_KEY"],
+        "skip_reason": "AWS credentials not available",
+    },
+    "openai": {
+        "model_class": OpenAIRealtimeBidirectionalModel,
+        "model_kwargs": {
+            "model": "gpt-4o-realtime-preview-2024-12-17",
+            "session": {
+                "output_modalities": ["audio"],  # OpenAI only supports audio OR text, not both
+                "audio": {
+                    "input": {
+                        "format": {"type": "audio/pcm", "rate": 24000},
+                        "turn_detection": {
+                            "type": "server_vad",
+                            "threshold": 0.5,
+                            "silence_duration_ms": 700,
+                        },
+                    },
+                    "output": {"format": {"type": "audio/pcm", "rate": 24000}, "voice": "alloy"},
+                },
+            },
+        },
+        "silence_duration": 1.0,  # OpenAI has faster VAD
+        "env_vars": ["OPENAI_API_KEY"],
+        "skip_reason": "OPENAI_API_KEY not available",
+    },
+    # NOTE: Gemini Live is temporarily disabled in parameterized tests
+    # Issue: Transcript events are not being properly emitted alongside audio events
+    # The model responds with audio but the test infrastructure expects text/transcripts
+    # TODO: Fix Gemini Live event emission to yield both transcript and audio events
+    # "gemini_live": {
+    #     "model_class": GeminiLiveBidirectionalModel,
+    #     "model_kwargs": {
+    #         "model_id": "gemini-2.5-flash-native-audio-preview-09-2025",
+    #         "params": {
+    #             "response_modalities": ["AUDIO"],
+    #             "output_audio_transcription": {},
+    #             "input_audio_transcription": {},
+    #         },
+    #     },
+    #     "silence_duration": 3.0,
+    #     "env_vars": ["GOOGLE_AI_API_KEY"],
+    #     "skip_reason": "GOOGLE_AI_API_KEY not available",
+    # },
+}
+
+
+def check_provider_available(provider_name: str) -> tuple[bool, str]:
+    """Check if a provider's credentials are available.
+
+    Args:
+        provider_name: Name of the provider to check.
+
+    Returns:
+        Tuple of (is_available, skip_reason).
+    """
+    config = PROVIDER_CONFIGS[provider_name]
+    env_vars = config["env_vars"]
+
+    missing_vars = [var for var in env_vars if not os.getenv(var)]
+
+    if missing_vars:
+        return False, f"{config['skip_reason']}: {', '.join(missing_vars)}"
+
+    return True, ""
+
+
+@pytest.fixture(params=list(PROVIDER_CONFIGS.keys()))
+def provider_config(request):
+    """Provide configuration for each model provider.
+
+    This fixture is parameterized to run tests against all available providers.
+    """
+    provider_name = request.param
+    config = PROVIDER_CONFIGS[provider_name]
+
+    # Check if provider is available
+    is_available, skip_reason = check_provider_available(provider_name)
+    if not is_available:
+        pytest.skip(skip_reason)
+
+    return {
+        "name": provider_name,
+        **config,
+    }
+
+
+@pytest.fixture
+def agent_with_calculator(provider_config):
+    """Provide bidirectional agent with calculator tool for the given provider.
+
+    Note: Session lifecycle (start/end) is handled by BidirectionalTestContext.
+    """
+    model_class = provider_config["model_class"]
+    model_kwargs = provider_config["model_kwargs"]
+
+    model = model_class(**model_kwargs)
+    return BidirectionalAgent(
+        model=model,
+        tools=[calculator],
+        system_prompt="You are a helpful assistant with access to a calculator tool. Keep responses brief.",
+    )
+
+@pytest.mark.asyncio
+async def test_bidirectional_agent(agent_with_calculator, audio_generator, provider_config):
+    """Test multi-turn conversation with follow-up questions across providers.
+
+    This test runs against all configured providers (Nova Sonic, OpenAI, etc.)
+    to validate provider-agnostic functionality.
+
+    Validates:
+    - Session lifecycle (start/end via context manager)
+    - Audio input streaming
+    - Speech-to-text transcription
+    - Tool execution (calculator)
+    - Multi-turn conversation flow
+    - Text-to-speech audio output
+    """
+    provider_name = provider_config["name"]
+    silence_duration = provider_config["silence_duration"]
+
+    logger.info(f"Testing provider: {provider_name}")
+
+    async with BidirectionalTestContext(agent_with_calculator, audio_generator) as ctx:
+        # Turn 1: Simple greeting to test basic audio I/O
+        await ctx.say("Hello, can you hear me?")
+        # Wait for silence to trigger provider's VAD/silence detection
+        await asyncio.sleep(silence_duration)
+        await ctx.wait_for_response()
+
+        text_outputs_turn1 = ctx.get_text_outputs()
+        all_text_turn1 = " ".join(text_outputs_turn1).lower()
+
+        # Validate turn 1 - just check we got a response
+        assert len(text_outputs_turn1) > 0, (
+            f"[{provider_name}] No text output received in turn 1"
+        )
+
+        logger.info(f"[{provider_name}] ✓ Turn 1 complete: received response")
+        logger.info(f"[{provider_name}]   Response: {text_outputs_turn1[0][:100]}...")
+
+        # Turn 2: Follow-up to test multi-turn conversation
+        await ctx.say("What's your name?")
+        # Wait for silence to trigger provider's VAD/silence detection
+        await asyncio.sleep(silence_duration)
+        await ctx.wait_for_response()
+
+        text_outputs_turn2 = ctx.get_text_outputs()
+
+        # Validate turn 2 - check we got more responses
+        assert len(text_outputs_turn2) > len(text_outputs_turn1), (
+            f"[{provider_name}] No new text output in turn 2"
+        )
+
+        logger.info(f"[{provider_name}] ✓ Turn 2 complete: multi-turn conversation works")
+        logger.info(f"[{provider_name}]   Total responses: {len(text_outputs_turn2)}")
+
+        # Validate full conversation
+        # Validate audio outputs
+        audio_outputs = ctx.get_audio_outputs()
+        assert len(audio_outputs) > 0, f"[{provider_name}] No audio output received"
+        total_audio_bytes = sum(len(audio) for audio in audio_outputs)
+
+        # Summary
+        logger.info("=" * 60)
+        logger.info(f"[{provider_name}] ✓ Multi-turn conversation test PASSED")
+        logger.info(f"  Provider: {provider_name}")
+        logger.info(f"  Total events: {len(ctx.get_events())}")
+        logger.info(f"  Text responses: {len(text_outputs_turn2)}")
+        logger.info(f"  Audio chunks: {len(audio_outputs)} ({total_audio_bytes:,} bytes)")
+        logger.info("=" * 60)

tests_integ/bidirectional_streaming/utils/__init__.py

@@ -0,0 +1 @@
+"""Utilities for bidirectional streaming integration tests."""

tests_integ/bidirectional_streaming/utils/audio_generator.py

@@ -0,0 +1,154 @@
+"""Audio generation utilities using Amazon Polly for test audio input.
+
+Provides text-to-speech conversion for generating realistic audio test data
+without requiring physical audio devices or pre-recorded files.
+"""
+
+import hashlib
+import logging
+from pathlib import Path
+from typing import Literal
+
+import boto3
+
+logger = logging.getLogger(__name__)
+
+# Audio format constants matching Nova Sonic requirements
+NOVA_SONIC_SAMPLE_RATE = 16000
+NOVA_SONIC_CHANNELS = 1
+NOVA_SONIC_FORMAT = "pcm"
+
+# Polly configuration
+POLLY_VOICE_ID = "Matthew"  # US English male voice
+POLLY_ENGINE = "neural"  # Higher quality neural engine
+
+# Cache directory for generated audio
+CACHE_DIR = Path(__file__).parent.parent / ".audio_cache"
+
+
+class AudioGenerator:
+    """Generate test audio using Amazon Polly with caching."""
+
+    def __init__(self, region: str = "us-east-1"):
+        """Initialize audio generator with Polly client.
+
+        Args:
+            region: AWS region for Polly service.
+        """
+        self.polly_client = boto3.client("polly", region_name=region)
+        self._ensure_cache_dir()
+
+    def _ensure_cache_dir(self) -> None:
+        """Create cache directory if it doesn't exist."""
+        CACHE_DIR.mkdir(parents=True, exist_ok=True)
+
+    def _get_cache_key(self, text: str, voice_id: str) -> str:
+        """Generate cache key from text and voice."""
+        content = f"{text}:{voice_id}".encode("utf-8")
+        return hashlib.md5(content).hexdigest()
+
+    def _get_cache_path(self, cache_key: str) -> Path:
+        """Get cache file path for given key."""
+        return CACHE_DIR / f"{cache_key}.pcm"
+
+    async def generate_audio(
+        self,
+        text: str,
+        voice_id: str = POLLY_VOICE_ID,
+        use_cache: bool = True,
+    ) -> bytes:
+        """Generate audio from text using Polly with caching.
+
+        Args:
+            text: Text to convert to speech.
+            voice_id: Polly voice ID to use.
+            use_cache: Whether to use cached audio if available.
+
+        Returns:
+            Raw PCM audio bytes at 16kHz mono (Nova Sonic format).
+        """
+        # Check cache first
+        if use_cache:
+            cache_key = self._get_cache_key(text, voice_id)
+            cache_path = self._get_cache_path(cache_key)
+
+            if cache_path.exists():
+                logger.debug(f"Using cached audio for: {text[:50]}...")
+                return cache_path.read_bytes()
+
+        # Generate audio with Polly
+        logger.debug(f"Generating audio with Polly: {text[:50]}...")
+
+        try:
+            response = self.polly_client.synthesize_speech(
+                Text=text,
+                OutputFormat="pcm",  # Raw PCM format
+                VoiceId=voice_id,
+                Engine=POLLY_ENGINE,
+                SampleRate=str(NOVA_SONIC_SAMPLE_RATE),
+            )
+
+            # Read audio data
+            audio_data = response["AudioStream"].read()
+
+            # Cache for future use
+            if use_cache:
+                cache_path.write_bytes(audio_data)
+                logger.debug(f"Cached audio: {cache_path}")
+
+            return audio_data
+
+        except Exception as e:
+            logger.error(f"Polly audio generation failed: {e}")
+            raise
+
+    def create_audio_input_event(
+        self,
+        audio_data: bytes,
+        format: Literal["pcm", "wav", "opus", "mp3"] = NOVA_SONIC_FORMAT,
+        sample_rate: int = NOVA_SONIC_SAMPLE_RATE,
+        channels: int = NOVA_SONIC_CHANNELS,
+    ) -> dict:
+        """Create AudioInputEvent from raw audio data.
+
+        Args:
+            audio_data: Raw audio bytes.
+            format: Audio format.
+            sample_rate: Sample rate in Hz.
+            channels: Number of audio channels.
+
+        Returns:
+            AudioInputEvent dict ready for agent.send().
+        """
+        return {
+            "audioData": audio_data,
+            "format": format,
+            "sampleRate": sample_rate,
+            "channels": channels,
+        }
+
+    def clear_cache(self) -> None:
+        """Clear all cached audio files."""
+        if CACHE_DIR.exists():
+            for cache_file in CACHE_DIR.glob("*.pcm"):
+                cache_file.unlink()
+            logger.info("Audio cache cleared")
+
+
+# Convenience function for quick audio generation
+async def generate_test_audio(text: str, use_cache: bool = True) -> dict:
+    """Generate test audio input event from text.
+
+    Convenience function that creates an AudioGenerator and returns
+    a ready-to-use AudioInputEvent.
+
+    Args:
+        text: Text to convert to speech.
+        use_cache: Whether to use cached audio.
+
+    Returns:
+        AudioInputEvent dict ready for agent.send().
+    """
+    generator = AudioGenerator()
+    audio_data = await generator.generate_audio(text, use_cache=use_cache)
+    return generator.create_audio_input_event(audio_data)