nifty-coder/stemsplit-backend

1"""2Pytest configuration and fixtures for voice control tests.3 4This module provides common fixtures and configuration for all tests5in the voice control optimization system.6"""7 8import pytest9import asyncio10import tempfile11import os12from typing import Dict, Any13from unittest.mock import Mock, AsyncMock14from datetime import datetime, timedelta15 16from hypothesis import settings, Verbosity17from hypothesis.strategies import composite, integers, floats, text, booleans, lists, dictionaries18 19from voice_control.models import (20    ProviderConfig, ProviderType, TranscriptionResult, UsageStats,21    ProviderStatus, ProcessedAudio, QuotaStatus, QuotaType22)23from voice_control.interfaces import STTProvider24 25 26# Configure Hypothesis for property-based testing27settings.register_profile("default", max_examples=100, verbosity=Verbosity.normal)28settings.register_profile("ci", max_examples=1000, verbosity=Verbosity.verbose)29settings.register_profile("dev", max_examples=10, verbosity=Verbosity.verbose)30 31# Use CI profile in CI environment, dev profile for local development32profile = os.environ.get("HYPOTHESIS_PROFILE", "default")33settings.load_profile(profile)34 35 36@pytest.fixture(scope="session")37def event_loop():38    """Create an instance of the default event loop for the test session."""39    loop = asyncio.get_event_loop_policy().new_event_loop()40    yield loop41    loop.close()42 43 44@pytest.fixture45def temp_dir():46    """Create a temporary directory for test files."""47    with tempfile.TemporaryDirectory() as tmpdir:48        yield tmpdir49 50 51@pytest.fixture52def sample_audio_data():53    """Generate sample audio data for testing."""54    # Generate 1 second of silence at 44.1kHz, 16-bit, mono55    import numpy as np56    sample_rate = 4410057    duration = 1.058    samples = int(sample_rate * duration)59    audio_array = np.zeros(samples, dtype=np.int16)60    return audio_array.tobytes()61 62 63@pytest.fixture64def sample_provider_config():65    """Create a sample provider configuration for testing."""66    return ProviderConfig(67        name="test_provider",68        provider_type=ProviderType.WEB_SPEECH_API,69        enabled=True,70        priority=1,71        free_tier_limits={"requests_per_day": 1000, "audio_minutes_per_month": 60},72        rate_limits={"requests_per_minute": 10, "audio_minutes_per_hour": 5},73        supported_formats=["webm", "wav", "mp3"],74        supported_languages=["en-US", "en-GB"],75        cost_per_minute=0.0,76        api_credentials={"api_key": "test_key"},77        timeout_seconds=30,78        max_retries=379    )80 81 82@pytest.fixture83def sample_transcription_result():84    """Create a sample transcription result for testing."""85    return TranscriptionResult(86        text="Hello world",87        confidence=0.95,88        provider="test_provider",89        processing_time=1.5,90        audio_duration=2.0,91        language="en-US",92        alternatives=["Hello world", "Hello word"],93        is_final=True,94        session_id="test_session_123"95    )96 97 98@pytest.fixture99def sample_usage_stats():100    """Create sample usage statistics for testing."""101    now = datetime.utcnow()102    return UsageStats(103        provider="test_provider",104        requests_count=100,105        audio_minutes=50.5,106        estimated_cost=2.50,107        success_rate=0.98,108        average_latency=1.2,109        time_window="hour",110        window_start=now - timedelta(hours=1),111        window_end=now112    )113 114 115@pytest.fixture116def sample_provider_status():117    """Create sample provider status for testing."""118    return ProviderStatus(119        name="test_provider",120        available=True,121        current_load=0.3,122        quota_remaining={"requests": 900.0, "audio_minutes": 45.5},123        last_error=None,124        response_time_avg=1.1,125        circuit_breaker_state="CLOSED",126        consecutive_failures=0127    )128 129 130@pytest.fixture131def mock_stt_provider(sample_provider_config, sample_transcription_result):132    """Create a mock STT provider for testing."""133    134    class MockSTTProvider(STTProvider):135        def __init__(self, config: ProviderConfig):136            super().__init__(config)137            self.health_status = True138            self.quota_status = {139                "requests_per_minute": QuotaStatus(140                    provider=config.name,141                    quota_type=QuotaType.REQUESTS_PER_MINUTE,142                    current_usage=5,143                    limit=10,144                    remaining=5,145                    reset_time=datetime.utcnow() + timedelta(minutes=1),146                    percentage_used=0.5147                )148            }149        150        async def transcribe_audio(self, audio_data: bytes, format: str, language: str = "en-US", **kwargs):151            return sample_transcription_result152        153        async def transcribe_streaming(self, audio_stream, format: str, language: str = "en-US", **kwargs):154            async for chunk in audio_stream:155                yield sample_transcription_result156        157        async def check_health(self) -> bool:158            return self.health_status159        160        async def get_quota_status(self):161            return self.quota_status162        163        def supports_format(self, format: str) -> bool:164            return format in self.config.supported_formats165        166        def supports_language(self, language: str) -> bool:167            return language in self.config.supported_languages168        169        async def estimate_cost(self, audio_duration: float) -> float:170            return audio_duration * self.config.cost_per_minute / 60.0171    172    return MockSTTProvider(sample_provider_config)173 174 175# Hypothesis strategies for generating test data176@composite177def provider_config_strategy(draw):178    """Strategy for generating valid ProviderConfig instances."""179    return ProviderConfig(180        name=draw(text(min_size=1, max_size=50)),181        provider_type=draw(integers(min_value=0, max_value=4).map(lambda x: list(ProviderType)[x])),182        enabled=draw(booleans()),183        priority=draw(integers(min_value=0, max_value=100)),184        free_tier_limits=draw(dictionaries(text(), integers(min_value=0))),185        rate_limits=draw(dictionaries(text(), integers(min_value=1))),186        supported_formats=draw(lists(text(min_size=1), min_size=1)),187        supported_languages=draw(lists(text(min_size=2, max_size=5), min_size=1)),188        cost_per_minute=draw(floats(min_value=0.0, max_value=1.0)),189        api_credentials=draw(dictionaries(text(), text())),190        timeout_seconds=draw(integers(min_value=1, max_value=300)),191        max_retries=draw(integers(min_value=0, max_value=10))192    )193 194 195@composite196def transcription_result_strategy(draw):197    """Strategy for generating valid TranscriptionResult instances."""198    return TranscriptionResult(199        text=draw(text()),200        confidence=draw(floats(min_value=0.0, max_value=1.0)),201        provider=draw(text(min_size=1)),202        processing_time=draw(floats(min_value=0.0, max_value=60.0)),203        audio_duration=draw(floats(min_value=0.1, max_value=3600.0)),204        language=draw(text(min_size=2, max_size=10)),205        alternatives=draw(lists(text(), max_size=5)),206        is_final=draw(booleans())207    )208 209 210@composite211def audio_data_strategy(draw):212    """Strategy for generating audio data bytes."""213    size = draw(integers(min_value=1024, max_value=1024*1024))  # 1KB to 1MB214    return bytes(draw(integers(min_value=0, max_value=255)) for _ in range(size))215 216 217@composite218def quota_status_strategy(draw):219    """Strategy for generating valid QuotaStatus instances."""220    limit = draw(floats(min_value=1.0, max_value=10000.0))221    current_usage = draw(floats(min_value=0.0, max_value=limit * 1.1))  # Allow slight over-usage222    223    return QuotaStatus(224        provider=draw(text(min_size=1)),225        quota_type=draw(integers(min_value=0, max_value=4).map(lambda x: list(QuotaType)[x])),226        current_usage=current_usage,227        limit=limit,228        remaining=max(0, limit - current_usage),229        reset_time=datetime.utcnow() + timedelta(hours=draw(integers(min_value=1, max_value=24))),230        percentage_used=min(current_usage / limit, 1.0) if limit > 0 else 0.0231    )232 233 234# Test data constants235VALID_AUDIO_FORMATS = ["webm", "wav", "mp3", "ogg", "flac"]236VALID_LANGUAGES = ["en-US", "en-GB", "es-ES", "fr-FR", "de-DE", "ja-JP", "zh-CN"]237PROVIDER_NAMES = ["web_speech_api", "google_speech", "azure_speech", "assembly_ai", "deepgram"]