114 lines
3.2 KiB
Python
114 lines
3.2 KiB
Python
"""
|
|
Pytest configuration and fixtures for voice-assistant tests.
|
|
"""
|
|
import asyncio
|
|
import base64
|
|
import os
|
|
from typing import AsyncGenerator
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
# Set test environment variables before importing
|
|
os.environ.setdefault("NATS_URL", "nats://localhost:4222")
|
|
os.environ.setdefault("REDIS_URL", "redis://localhost:6379")
|
|
os.environ.setdefault("MILVUS_HOST", "localhost")
|
|
os.environ.setdefault("OTEL_ENABLED", "false")
|
|
os.environ.setdefault("MLFLOW_ENABLED", "false")
|
|
|
|
|
|
@pytest.fixture(scope="session")
|
|
def event_loop():
|
|
"""Create event loop for async tests."""
|
|
loop = asyncio.new_event_loop()
|
|
yield loop
|
|
loop.close()
|
|
|
|
|
|
@pytest.fixture
|
|
def sample_audio_b64():
|
|
"""Sample base64 encoded audio for testing."""
|
|
# 16-bit PCM silence (44 bytes header + 1000 samples)
|
|
wav_header = bytes([
|
|
0x52, 0x49, 0x46, 0x46, # "RIFF"
|
|
0x24, 0x08, 0x00, 0x00, # File size
|
|
0x57, 0x41, 0x56, 0x45, # "WAVE"
|
|
0x66, 0x6D, 0x74, 0x20, # "fmt "
|
|
0x10, 0x00, 0x00, 0x00, # Chunk size
|
|
0x01, 0x00, # PCM format
|
|
0x01, 0x00, # Mono
|
|
0x80, 0x3E, 0x00, 0x00, # Sample rate (16000)
|
|
0x00, 0x7D, 0x00, 0x00, # Byte rate
|
|
0x02, 0x00, # Block align
|
|
0x10, 0x00, # Bits per sample
|
|
0x64, 0x61, 0x74, 0x61, # "data"
|
|
0x00, 0x08, 0x00, 0x00, # Data size
|
|
])
|
|
silence = bytes([0x00] * 2048)
|
|
return base64.b64encode(wav_header + silence).decode()
|
|
|
|
|
|
@pytest.fixture
|
|
def sample_embedding():
|
|
"""Sample embedding vector."""
|
|
return [0.1] * 1024
|
|
|
|
|
|
@pytest.fixture
|
|
def sample_documents():
|
|
"""Sample search results."""
|
|
return [
|
|
{"text": "Machine learning is a subset of AI.", "score": 0.95},
|
|
{"text": "Deep learning uses neural networks.", "score": 0.90},
|
|
{"text": "AI enables intelligent automation.", "score": 0.85},
|
|
]
|
|
|
|
|
|
@pytest.fixture
|
|
def sample_reranked():
|
|
"""Sample reranked results."""
|
|
return [
|
|
{"document": "Machine learning is a subset of AI.", "score": 0.98},
|
|
{"document": "Deep learning uses neural networks.", "score": 0.85},
|
|
]
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_nats_message():
|
|
"""Create a mock NATS message."""
|
|
msg = MagicMock()
|
|
msg.subject = "voice.request"
|
|
msg.reply = "voice.response.test-123"
|
|
return msg
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_voice_request(sample_audio_b64):
|
|
"""Sample voice request payload."""
|
|
return {
|
|
"request_id": "test-request-123",
|
|
"audio": sample_audio_b64,
|
|
"language": "en",
|
|
"collection": "test_collection",
|
|
}
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_clients():
|
|
"""Mock all service clients."""
|
|
with patch("voice_assistant.STTClient") as stt, \
|
|
patch("voice_assistant.EmbeddingsClient") as embeddings, \
|
|
patch("voice_assistant.RerankerClient") as reranker, \
|
|
patch("voice_assistant.LLMClient") as llm, \
|
|
patch("voice_assistant.TTSClient") as tts, \
|
|
patch("voice_assistant.MilvusClient") as milvus:
|
|
|
|
yield {
|
|
"stt": stt,
|
|
"embeddings": embeddings,
|
|
"reranker": reranker,
|
|
"llm": llm,
|
|
"tts": tts,
|
|
"milvus": milvus,
|
|
}
|