daviestechlabs/voice-assistant
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: ruff formatting, allow-direct-references, and noqa for Kubeflow pipeline vars
All checks were successful
CI / Lint (push) Successful in 51s
Details
CI / Test (push) Successful in 55s
Details
CI / Release (push) Successful in 7s
Details
CI / Notify (push) Successful in 2s
Details

Browse Source
This commit is contained in:
Billy Davies
2026-02-02 08:44:14 -05:00
parent 58465b77d8
commit 0462412353
5 changed files with 208 additions and 237 deletions
Download Patch File Download Diff File Expand all files Collapse all files

82
tests/conftest.py
View File

@@ -1,11 +1,11 @@
"""
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
from unittest.mock import MagicMock, patch
import pytest
@@ -29,21 +29,54 @@ def event_loop():
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
])
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()
@@ -96,13 +129,14 @@ def mock_voice_request(sample_audio_b64):
@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:
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,

97
tests/test_voice_assistant.py
View File

@@ -1,9 +1,9 @@
"""
Unit tests for VoiceAssistant handler.
"""
import base64
import pytest
from unittest.mock import AsyncMock, MagicMock, patch
from unittest.mock import AsyncMock, patch
# Import after environment is set up in conftest
from voice_assistant import VoiceAssistant, VoiceSettings
@@ -11,11 +11,11 @@ from voice_assistant import VoiceAssistant, VoiceSettings
class TestVoiceSettings:
"""Tests for VoiceSettings configuration."""
def test_default_settings(self):
"""Test default settings values."""
settings = VoiceSettings()
assert settings.service_name == "voice-assistant"
assert settings.rag_top_k == 10
assert settings.rag_rerank_top_k == 5
@@ -24,37 +24,35 @@ class TestVoiceSettings:
assert settings.tts_language == "en"
assert settings.include_transcription is True
assert settings.include_sources is False
def test_custom_settings(self, monkeypatch):
"""Test settings from environment."""
monkeypatch.setenv("RAG_TOP_K", "20")
monkeypatch.setenv("RAG_COLLECTION", "custom_collection")
# Note: Would need to re-instantiate settings to pick up env vars
settings = VoiceSettings(
rag_top_k=20,
rag_collection="custom_collection"
)
settings = VoiceSettings(rag_top_k=20, rag_collection="custom_collection")
assert settings.rag_top_k == 20
assert settings.rag_collection == "custom_collection"
class TestVoiceAssistant:
"""Tests for VoiceAssistant handler."""
@pytest.fixture
def handler(self):
"""Create handler with mocked clients."""
with patch("voice_assistant.STTClient"), \
patch("voice_assistant.EmbeddingsClient"), \
patch("voice_assistant.RerankerClient"), \
patch("voice_assistant.LLMClient"), \
patch("voice_assistant.TTSClient"), \
patch("voice_assistant.MilvusClient"):
with (
patch("voice_assistant.STTClient"),
patch("voice_assistant.EmbeddingsClient"),
patch("voice_assistant.RerankerClient"),
patch("voice_assistant.LLMClient"),
patch("voice_assistant.TTSClient"),
patch("voice_assistant.MilvusClient"),
):
handler = VoiceAssistant()
# Setup mock clients
handler.stt = AsyncMock()
handler.embeddings = AsyncMock()
@@ -63,15 +61,15 @@ class TestVoiceAssistant:
handler.tts = AsyncMock()
handler.milvus = AsyncMock()
handler.nats = AsyncMock()
yield handler
def test_init(self, handler):
"""Test handler initialization."""
assert handler.subject == "voice.request"
assert handler.queue_group == "voice-assistants"
assert handler.voice_settings.service_name == "voice-assistant"
@pytest.mark.asyncio
async def test_handle_message_success(
self,
@@ -90,16 +88,16 @@ class TestVoiceAssistant:
handler.reranker.rerank.return_value = sample_reranked
handler.llm.generate.return_value = "Machine learning is a type of AI."
handler.tts.synthesize.return_value = b"audio_bytes"
# Execute
result = await handler.handle_message(mock_nats_message, mock_voice_request)
# Verify
assert result["request_id"] == "test-request-123"
assert result["response"] == "Machine learning is a type of AI."
assert "audio" in result
assert result["transcription"] == "What is machine learning?"
# Verify pipeline was called
handler.stt.transcribe.assert_called_once()
handler.embeddings.embed_single.assert_called_once()
@@ -107,7 +105,7 @@ class TestVoiceAssistant:
handler.reranker.rerank.assert_called_once()
handler.llm.generate.assert_called_once()
handler.tts.synthesize.assert_called_once()
@pytest.mark.asyncio
async def test_handle_message_empty_transcription(
self,
@@ -117,15 +115,15 @@ class TestVoiceAssistant:
):
"""Test handling when transcription is empty."""
handler.stt.transcribe.return_value = {"text": ""}
result = await handler.handle_message(mock_nats_message, mock_voice_request)
assert "error" in result
assert result["error"] == "Could not transcribe audio"
# Verify pipeline stopped after transcription
handler.embeddings.embed_single.assert_not_called()
@pytest.mark.asyncio
async def test_handle_message_with_sources(
self,
@@ -138,7 +136,7 @@ class TestVoiceAssistant:
):
"""Test response includes sources when enabled."""
handler.voice_settings.include_sources = True
# Setup mocks
handler.stt.transcribe.return_value = {"text": "Hello"}
handler.embeddings.embed_single.return_value = sample_embedding
@@ -146,51 +144,52 @@ class TestVoiceAssistant:
handler.reranker.rerank.return_value = sample_reranked
handler.llm.generate.return_value = "Hi there!"
handler.tts.synthesize.return_value = b"audio"
result = await handler.handle_message(mock_nats_message, mock_voice_request)
assert "sources" in result
assert len(result["sources"]) <= 3
def test_build_context(self, handler):
"""Test context building from documents."""
documents = [
{"document": "First doc content"},
{"document": "Second doc content"},
]
context = handler._build_context(documents)
assert "First doc content" in context
assert "Second doc content" in context
@pytest.mark.asyncio
async def test_setup_initializes_clients(self):
"""Test that setup initializes all clients."""
with patch("voice_assistant.STTClient") as stt_cls, \
patch("voice_assistant.EmbeddingsClient") as emb_cls, \
patch("voice_assistant.RerankerClient") as rer_cls, \
patch("voice_assistant.LLMClient") as llm_cls, \
patch("voice_assistant.TTSClient") as tts_cls, \
patch("voice_assistant.MilvusClient") as mil_cls:
with (
patch("voice_assistant.STTClient") as stt_cls,
patch("voice_assistant.EmbeddingsClient") as emb_cls,
patch("voice_assistant.RerankerClient") as rer_cls,
patch("voice_assistant.LLMClient") as llm_cls,
patch("voice_assistant.TTSClient") as tts_cls,
patch("voice_assistant.MilvusClient") as mil_cls,
):
mil_cls.return_value.connect = AsyncMock()
handler = VoiceAssistant()
await handler.setup()
stt_cls.assert_called_once()
emb_cls.assert_called_once()
rer_cls.assert_called_once()
llm_cls.assert_called_once()
tts_cls.assert_called_once()
mil_cls.assert_called_once()
@pytest.mark.asyncio
async def test_teardown_closes_clients(self, handler):
"""Test that teardown closes all clients."""
await handler.teardown()
handler.stt.close.assert_called_once()
handler.embeddings.close.assert_called_once()
handler.reranker.close.assert_called_once()