mirror of
https://github.com/Purple-Horizons/openclaw-voice.git
synced 2026-08-14 09:02:28 +00:00
Add comprehensive test suite
Unit tests (13): - WhisperSTT: init, transcribe silence, transcribe noise - ChatterboxTTS: init, synthesize - AIBackend: init, system prompt, clear history, chat (live API) - VAD: init, is_speech - Integration: STT→TTS round trip Server tests (5): - HTTP index page loads - WebSocket connect, ping/pong - Start/stop listening cycle - Audio flow with transcript All tests pass with faster-whisper on CPU. Python 3.12 required (faster-whisper not yet on 3.14).
This commit is contained in:
gianni-dalerta
@@ -0,0 +1 @@
|
||||
# OpenClaw Voice Tests
|
||||
@@ -0,0 +1,162 @@
|
||||
"""
|
||||
Unit tests for OpenClaw Voice modules.
|
||||
"""
|
||||
|
||||
import pytest
|
||||
import numpy as np
|
||||
import asyncio
|
||||
import os
|
||||
import sys
|
||||
|
||||
# Add src to path
|
||||
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
|
||||
|
||||
from src.server.stt import WhisperSTT
|
||||
from src.server.tts import ChatterboxTTS
|
||||
from src.server.backend import AIBackend
|
||||
from src.server.vad import VoiceActivityDetector
|
||||
|
||||
|
||||
class TestWhisperSTT:
|
||||
"""Tests for Speech-to-Text module."""
|
||||
|
||||
def test_init_loads_model(self):
|
||||
"""Test that STT initializes (may be mock or real)."""
|
||||
stt = WhisperSTT(model_name="tiny", device="cpu")
|
||||
assert stt is not None
|
||||
assert stt._backend in ["faster-whisper", "openai-whisper", "mock"]
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_transcribe_returns_string(self):
|
||||
"""Test that transcribe returns a string."""
|
||||
stt = WhisperSTT(model_name="tiny", device="cpu")
|
||||
# Create 1 second of silence at 16kHz
|
||||
audio = np.zeros(16000, dtype=np.float32)
|
||||
result = await stt.transcribe(audio)
|
||||
assert isinstance(result, str)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_transcribe_with_noise(self):
|
||||
"""Test transcription with random noise (should return something)."""
|
||||
stt = WhisperSTT(model_name="tiny", device="cpu")
|
||||
# Random noise
|
||||
audio = np.random.randn(16000).astype(np.float32) * 0.1
|
||||
result = await stt.transcribe(audio)
|
||||
assert isinstance(result, str)
|
||||
|
||||
|
||||
class TestChatterboxTTS:
|
||||
"""Tests for Text-to-Speech module."""
|
||||
|
||||
def test_init_loads_model(self):
|
||||
"""Test that TTS initializes (may be mock or real)."""
|
||||
tts = ChatterboxTTS()
|
||||
assert tts is not None
|
||||
assert tts._backend in ["chatterbox", "xtts", "pyttsx3", "mock"]
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_synthesize_returns_audio(self):
|
||||
"""Test that synthesize returns numpy array."""
|
||||
tts = ChatterboxTTS()
|
||||
result = await tts.synthesize("Hello world")
|
||||
assert isinstance(result, np.ndarray)
|
||||
assert result.dtype == np.float32
|
||||
assert len(result) > 0
|
||||
|
||||
|
||||
class TestAIBackend:
|
||||
"""Tests for AI Backend module."""
|
||||
|
||||
def test_init_creates_client(self):
|
||||
"""Test backend initialization."""
|
||||
backend = AIBackend(
|
||||
backend_type="openai",
|
||||
model="gpt-4o-mini",
|
||||
)
|
||||
assert backend is not None
|
||||
assert backend.backend_type == "openai"
|
||||
|
||||
def test_system_prompt_default(self):
|
||||
"""Test default system prompt is set."""
|
||||
backend = AIBackend()
|
||||
assert backend.system_prompt is not None
|
||||
assert "voice assistant" in backend.system_prompt.lower()
|
||||
|
||||
def test_clear_history(self):
|
||||
"""Test conversation history can be cleared."""
|
||||
backend = AIBackend()
|
||||
backend.conversation_history = [{"role": "user", "content": "test"}]
|
||||
backend.clear_history()
|
||||
assert len(backend.conversation_history) == 0
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.skipif(
|
||||
not os.getenv("OPENAI_API_KEY"),
|
||||
reason="OPENAI_API_KEY not set"
|
||||
)
|
||||
async def test_chat_returns_response(self):
|
||||
"""Test actual API call (requires API key)."""
|
||||
backend = AIBackend(
|
||||
backend_type="openai",
|
||||
model="gpt-4o-mini",
|
||||
api_key=os.getenv("OPENAI_API_KEY"),
|
||||
)
|
||||
result = await backend.chat("Say 'test' and nothing else.")
|
||||
assert isinstance(result, str)
|
||||
assert len(result) > 0
|
||||
|
||||
|
||||
class TestVAD:
|
||||
"""Tests for Voice Activity Detection module."""
|
||||
|
||||
def test_init(self):
|
||||
"""Test VAD initialization."""
|
||||
vad = VoiceActivityDetector()
|
||||
assert vad is not None
|
||||
|
||||
def test_is_speech_silence(self):
|
||||
"""Test that silence is not detected as speech."""
|
||||
vad = VoiceActivityDetector()
|
||||
silence = np.zeros(16000, dtype=np.float32)
|
||||
# Should return True if no VAD model (assumes speech)
|
||||
# or False if VAD model is loaded and detects no speech
|
||||
result = vad.is_speech(silence)
|
||||
assert isinstance(result, bool)
|
||||
|
||||
def test_is_speech_noise(self):
|
||||
"""Test with random noise."""
|
||||
vad = VoiceActivityDetector()
|
||||
noise = np.random.randn(16000).astype(np.float32)
|
||||
result = vad.is_speech(noise)
|
||||
assert isinstance(result, bool)
|
||||
|
||||
|
||||
class TestIntegration:
|
||||
"""Integration tests for the full pipeline."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_stt_tts_round_trip(self):
|
||||
"""Test STT → TTS round trip (mock mode OK)."""
|
||||
stt = WhisperSTT(model_name="tiny", device="cpu")
|
||||
tts = ChatterboxTTS()
|
||||
|
||||
# Generate some audio (silence)
|
||||
input_audio = np.zeros(16000, dtype=np.float32)
|
||||
|
||||
# Transcribe
|
||||
text = await stt.transcribe(input_audio)
|
||||
assert isinstance(text, str)
|
||||
|
||||
# Synthesize (even empty text should work)
|
||||
if text.strip():
|
||||
output_audio = await tts.synthesize(text)
|
||||
else:
|
||||
output_audio = await tts.synthesize("Hello")
|
||||
|
||||
assert isinstance(output_audio, np.ndarray)
|
||||
assert len(output_audio) > 0
|
||||
|
||||
|
||||
# Run tests with: pytest tests/ -v
|
||||
if __name__ == "__main__":
|
||||
pytest.main([__file__, "-v"])
|
||||
@@ -0,0 +1,160 @@
|
||||
"""
|
||||
Server integration tests for OpenClaw Voice.
|
||||
"""
|
||||
|
||||
import pytest
|
||||
import asyncio
|
||||
import json
|
||||
import base64
|
||||
import numpy as np
|
||||
import os
|
||||
import sys
|
||||
import subprocess
|
||||
import time
|
||||
import socket
|
||||
|
||||
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
|
||||
|
||||
|
||||
def is_port_in_use(port: int) -> bool:
|
||||
"""Check if a port is in use."""
|
||||
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
|
||||
return s.connect_ex(('127.0.0.1', port)) == 0
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def server():
|
||||
"""Start the server for testing."""
|
||||
port = 8799 # Use high port to avoid conflicts
|
||||
|
||||
# Skip if port already in use
|
||||
if is_port_in_use(port):
|
||||
pytest.skip(f"Port {port} already in use")
|
||||
|
||||
# Start server
|
||||
env = os.environ.copy()
|
||||
env['OPENCLAW_PORT'] = str(port)
|
||||
env['OPENCLAW_STT_MODEL'] = 'tiny' # Use tiny for fast tests
|
||||
|
||||
proc = subprocess.Popen(
|
||||
[sys.executable, '-m', 'uvicorn', 'src.server.main:app',
|
||||
'--host', '127.0.0.1', '--port', str(port)],
|
||||
env=env,
|
||||
stdout=subprocess.PIPE,
|
||||
stderr=subprocess.PIPE,
|
||||
cwd=os.path.dirname(os.path.dirname(__file__)),
|
||||
)
|
||||
|
||||
# Wait for server to be ready
|
||||
max_wait = 15
|
||||
for i in range(max_wait):
|
||||
if is_port_in_use(port):
|
||||
break
|
||||
time.sleep(1)
|
||||
else:
|
||||
proc.terminate()
|
||||
pytest.fail("Server did not start in time")
|
||||
|
||||
yield f"ws://127.0.0.1:{port}/ws", f"http://127.0.0.1:{port}"
|
||||
|
||||
# Cleanup
|
||||
proc.terminate()
|
||||
proc.wait(timeout=5)
|
||||
|
||||
|
||||
class TestServerHTTP:
|
||||
"""Test HTTP endpoints."""
|
||||
|
||||
def test_index_page(self, server):
|
||||
"""Test that index page loads."""
|
||||
import httpx
|
||||
|
||||
ws_url, http_url = server
|
||||
response = httpx.get(f"{http_url}/")
|
||||
|
||||
assert response.status_code == 200
|
||||
assert "OpenClaw Voice" in response.text
|
||||
assert "voice-button" in response.text
|
||||
|
||||
|
||||
class TestServerWebSocket:
|
||||
"""Test WebSocket functionality."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_websocket_connect(self, server):
|
||||
"""Test WebSocket connection."""
|
||||
import websockets
|
||||
|
||||
ws_url, _ = server
|
||||
async with websockets.connect(ws_url) as ws:
|
||||
# Connection successful if we get here
|
||||
assert ws is not None
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_ping_pong(self, server):
|
||||
"""Test ping/pong."""
|
||||
import websockets
|
||||
|
||||
ws_url, _ = server
|
||||
async with websockets.connect(ws_url) as ws:
|
||||
await ws.send(json.dumps({"type": "ping"}))
|
||||
response = json.loads(await ws.recv())
|
||||
assert response["type"] == "pong"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_start_stop_listening(self, server):
|
||||
"""Test start/stop listening cycle."""
|
||||
import websockets
|
||||
|
||||
ws_url, _ = server
|
||||
async with websockets.connect(ws_url) as ws:
|
||||
# Start
|
||||
await ws.send(json.dumps({"type": "start_listening"}))
|
||||
response = json.loads(await ws.recv())
|
||||
assert response["type"] == "listening_started"
|
||||
|
||||
# Stop
|
||||
await ws.send(json.dumps({"type": "stop_listening"}))
|
||||
response = json.loads(await ws.recv())
|
||||
assert response["type"] == "listening_stopped"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_audio_flow(self, server):
|
||||
"""Test sending audio and getting response."""
|
||||
import websockets
|
||||
|
||||
ws_url, _ = server
|
||||
async with websockets.connect(ws_url) as ws:
|
||||
# Start listening
|
||||
await ws.send(json.dumps({"type": "start_listening"}))
|
||||
await ws.recv() # listening_started
|
||||
|
||||
# Send some audio (silence)
|
||||
audio = np.zeros(16000, dtype=np.float32)
|
||||
audio_b64 = base64.b64encode(audio.tobytes()).decode()
|
||||
|
||||
await ws.send(json.dumps({
|
||||
"type": "audio",
|
||||
"data": audio_b64,
|
||||
}))
|
||||
|
||||
# Stop listening
|
||||
await ws.send(json.dumps({"type": "stop_listening"}))
|
||||
|
||||
# Should get transcript first, then listening_stopped
|
||||
messages = []
|
||||
for _ in range(5): # Collect up to 5 messages
|
||||
try:
|
||||
response = json.loads(await asyncio.wait_for(ws.recv(), timeout=5.0))
|
||||
messages.append(response["type"])
|
||||
if response["type"] == "listening_stopped":
|
||||
break
|
||||
except asyncio.TimeoutError:
|
||||
break
|
||||
|
||||
# Should have gotten transcript and/or listening_stopped
|
||||
assert "transcript" in messages or "listening_stopped" in messages
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
pytest.main([__file__, "-v"])
|
||||
Reference in New Issue
Block a user