forked from Rockachopa/Timmy-time-dashboard
Kimi Agent
9f244ffc70
Co-authored-by: Kimi Agent <kimi@timmy.local> Co-committed-by: Kimi Agent <kimi@timmy.local>
364 lines
12 KiB
Python
364 lines
12 KiB
Python
"""Tests for the sovereign voice loop.
|
|
|
|
These tests verify the VoiceLoop components without requiring a microphone,
|
|
Whisper model, or Piper installation — all I/O is mocked.
|
|
"""
|
|
|
|
from pathlib import Path
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
try:
|
|
import numpy as np
|
|
except ImportError:
|
|
np = None
|
|
|
|
try:
|
|
from timmy.voice_loop import VoiceConfig, VoiceLoop, _rms, _strip_markdown
|
|
except ImportError:
|
|
pass # pytestmark will skip all tests anyway
|
|
|
|
pytestmark = pytest.mark.skipif(np is None, reason="numpy not installed")
|
|
|
|
|
|
# ── VoiceConfig tests ──────────────────────────────────────────────────────
|
|
|
|
|
|
class TestVoiceConfig:
|
|
def test_defaults(self):
|
|
cfg = VoiceConfig()
|
|
assert cfg.whisper_model == "base.en"
|
|
assert cfg.sample_rate == 16000
|
|
assert cfg.silence_threshold == 0.015
|
|
assert cfg.silence_duration == 1.5
|
|
assert cfg.min_utterance == 0.5
|
|
assert cfg.max_utterance == 30.0
|
|
assert cfg.session_id == "voice"
|
|
assert cfg.use_say_fallback is False
|
|
|
|
def test_custom_values(self):
|
|
cfg = VoiceConfig(
|
|
whisper_model="tiny.en",
|
|
silence_threshold=0.02,
|
|
session_id="custom",
|
|
use_say_fallback=True,
|
|
)
|
|
assert cfg.whisper_model == "tiny.en"
|
|
assert cfg.silence_threshold == 0.02
|
|
assert cfg.session_id == "custom"
|
|
assert cfg.use_say_fallback is True
|
|
|
|
|
|
# ── VoiceLoop unit tests ──────────────────────────────────────────────────
|
|
|
|
|
|
class TestVoiceLoopInit:
|
|
def test_default_config(self):
|
|
loop = VoiceLoop()
|
|
assert loop.config.whisper_model == "base.en"
|
|
assert loop._running is False
|
|
assert loop._speaking is False
|
|
|
|
def test_custom_config(self):
|
|
cfg = VoiceConfig(whisper_model="tiny.en")
|
|
loop = VoiceLoop(config=cfg)
|
|
assert loop.config.whisper_model == "tiny.en"
|
|
|
|
|
|
class TestPiperFallback:
|
|
def test_falls_back_to_say_when_no_voice_file(self):
|
|
cfg = VoiceConfig(piper_voice=Path("/nonexistent/voice.onnx"))
|
|
loop = VoiceLoop(config=cfg)
|
|
loop._ensure_piper()
|
|
assert loop.config.use_say_fallback is True
|
|
|
|
def test_keeps_piper_when_voice_exists(self, tmp_path):
|
|
voice_file = tmp_path / "test.onnx"
|
|
voice_file.write_bytes(b"fake model")
|
|
cfg = VoiceConfig(piper_voice=voice_file)
|
|
loop = VoiceLoop(config=cfg)
|
|
loop._ensure_piper()
|
|
assert loop.config.use_say_fallback is False
|
|
|
|
|
|
class TestTranscribe:
|
|
def test_transcribes_audio(self):
|
|
"""Whisper transcription returns cleaned text."""
|
|
loop = VoiceLoop()
|
|
|
|
mock_model = MagicMock()
|
|
mock_model.transcribe.return_value = {"text": " Hello Timmy "}
|
|
loop._whisper_model = mock_model
|
|
|
|
audio = np.random.randn(16000).astype(np.float32)
|
|
result = loop._transcribe(audio)
|
|
|
|
assert result == "Hello Timmy"
|
|
mock_model.transcribe.assert_called_once()
|
|
|
|
def test_transcribes_empty_returns_empty(self):
|
|
loop = VoiceLoop()
|
|
mock_model = MagicMock()
|
|
mock_model.transcribe.return_value = {"text": " "}
|
|
loop._whisper_model = mock_model
|
|
|
|
audio = np.random.randn(16000).astype(np.float32)
|
|
result = loop._transcribe(audio)
|
|
assert result == ""
|
|
|
|
|
|
class TestStripMarkdown:
|
|
def test_strips_bold(self):
|
|
assert _strip_markdown("**hello**") == "hello"
|
|
|
|
def test_strips_italic(self):
|
|
assert _strip_markdown("*hello*") == "hello"
|
|
|
|
def test_strips_headers(self):
|
|
assert _strip_markdown("## Header\ntext") == "Header\ntext"
|
|
|
|
def test_strips_bullets(self):
|
|
assert _strip_markdown("- item one\n- item two") == "item one\nitem two"
|
|
|
|
def test_strips_numbered_lists(self):
|
|
assert _strip_markdown("1. first\n2. second") == "first\nsecond"
|
|
|
|
def test_strips_inline_code(self):
|
|
assert _strip_markdown("use `pip install`") == "use pip install"
|
|
|
|
def test_strips_links(self):
|
|
assert _strip_markdown("[click here](https://x.com)") == "click here"
|
|
|
|
def test_preserves_plain_text(self):
|
|
assert _strip_markdown("Hello, how are you?") == "Hello, how are you?"
|
|
|
|
def test_empty_string(self):
|
|
assert _strip_markdown("") == ""
|
|
|
|
def test_none_passthrough(self):
|
|
assert _strip_markdown(None) is None
|
|
|
|
def test_complex_markdown(self):
|
|
md = "**1. First** thing\n- use `code`\n*emphasis*"
|
|
result = _strip_markdown(md)
|
|
assert "**" not in result
|
|
assert "`" not in result
|
|
assert "*" not in result
|
|
|
|
|
|
class TestRms:
|
|
def test_silent_block(self):
|
|
block = np.zeros(1600, dtype=np.float32)
|
|
assert _rms(block) == pytest.approx(0.0, abs=1e-7)
|
|
|
|
def test_loud_block(self):
|
|
block = np.ones(1600, dtype=np.float32)
|
|
assert _rms(block) == pytest.approx(1.0, abs=1e-5)
|
|
|
|
|
|
class TestFinalizeUtterance:
|
|
def test_returns_none_for_empty(self):
|
|
assert VoiceLoop._finalize_utterance([], min_blocks=5, sample_rate=16000) is None
|
|
|
|
def test_returns_none_for_too_short(self):
|
|
chunks = [np.zeros(1600, dtype=np.float32) for _ in range(3)]
|
|
assert VoiceLoop._finalize_utterance(chunks, min_blocks=5, sample_rate=16000) is None
|
|
|
|
def test_returns_audio_for_sufficient_chunks(self):
|
|
chunks = [np.ones(1600, dtype=np.float32) for _ in range(6)]
|
|
result = VoiceLoop._finalize_utterance(chunks, min_blocks=5, sample_rate=16000)
|
|
assert result is not None
|
|
assert len(result) == 6 * 1600
|
|
|
|
|
|
class TestThink:
|
|
def test_think_returns_response(self):
|
|
loop = VoiceLoop()
|
|
loop._loop = MagicMock()
|
|
loop._loop.is_closed.return_value = False
|
|
loop._loop.run_until_complete.return_value = "I am Timmy."
|
|
result = loop._think("Who are you?")
|
|
assert result == "I am Timmy."
|
|
|
|
def test_think_handles_error(self):
|
|
loop = VoiceLoop()
|
|
loop._loop = MagicMock()
|
|
loop._loop.is_closed.return_value = False
|
|
loop._loop.run_until_complete.side_effect = RuntimeError("Ollama down")
|
|
result = loop._think("test")
|
|
assert "trouble" in result.lower()
|
|
|
|
def test_think_strips_markdown(self):
|
|
loop = VoiceLoop()
|
|
loop._loop = MagicMock()
|
|
loop._loop.is_closed.return_value = False
|
|
loop._loop.run_until_complete.return_value = "**Hello** from *Timmy*"
|
|
result = loop._think("test")
|
|
assert "**" not in result
|
|
assert "*" not in result
|
|
assert "Hello" in result
|
|
|
|
|
|
class TestSpeakSay:
|
|
@patch("subprocess.Popen")
|
|
def test_speak_say_calls_subprocess(self, mock_popen):
|
|
mock_proc = MagicMock()
|
|
mock_proc.wait.return_value = 0
|
|
mock_popen.return_value = mock_proc
|
|
|
|
cfg = VoiceConfig(use_say_fallback=True)
|
|
loop = VoiceLoop(config=cfg)
|
|
loop._speak_say("Hello")
|
|
|
|
mock_popen.assert_called_once()
|
|
args = mock_popen.call_args[0][0]
|
|
assert args[0] == "say"
|
|
assert "Hello" in args
|
|
|
|
@patch("subprocess.Popen", side_effect=FileNotFoundError)
|
|
def test_speak_say_handles_missing(self, mock_popen):
|
|
cfg = VoiceConfig(use_say_fallback=True)
|
|
loop = VoiceLoop(config=cfg)
|
|
# Should not raise
|
|
loop._speak_say("Hello")
|
|
|
|
|
|
class TestSpeakPiper:
|
|
@patch("timmy.voice_loop.VoiceLoop._play_audio")
|
|
@patch("subprocess.run")
|
|
def test_speak_piper_generates_and_plays(self, mock_run, mock_play):
|
|
mock_run.return_value = MagicMock(returncode=0, stderr="")
|
|
|
|
voice_path = Path("/tmp/test_voice.onnx")
|
|
cfg = VoiceConfig(piper_voice=voice_path)
|
|
loop = VoiceLoop(config=cfg)
|
|
loop._speak_piper("Hello from Piper")
|
|
|
|
# Piper was called
|
|
mock_run.assert_called_once()
|
|
cmd = mock_run.call_args[0][0]
|
|
assert cmd[0] == "piper"
|
|
assert "--model" in cmd
|
|
|
|
# Audio was played
|
|
mock_play.assert_called_once()
|
|
|
|
@patch("timmy.voice_loop.VoiceLoop._speak_say")
|
|
@patch("subprocess.run")
|
|
def test_speak_piper_falls_back_on_error(self, mock_run, mock_say):
|
|
mock_run.return_value = MagicMock(returncode=1, stderr="model error")
|
|
|
|
cfg = VoiceConfig(piper_voice=Path("/tmp/test.onnx"))
|
|
loop = VoiceLoop(config=cfg)
|
|
loop._speak_piper("test")
|
|
|
|
# Should fall back to say
|
|
mock_say.assert_called_once_with("test")
|
|
|
|
|
|
class TestHallucinationFilter:
|
|
"""Whisper tends to hallucinate on silence/noise. The loop should filter these."""
|
|
|
|
def test_known_hallucinations_filtered(self):
|
|
loop = VoiceLoop()
|
|
hallucinations = [
|
|
"you",
|
|
"thanks.",
|
|
"Thank you.",
|
|
"Bye.",
|
|
"Thanks for watching!",
|
|
"Thank you for watching!",
|
|
"",
|
|
]
|
|
for text in hallucinations:
|
|
assert loop._is_hallucination(text), f"'{text}' should be filtered"
|
|
|
|
def test_real_speech_not_filtered(self):
|
|
loop = VoiceLoop()
|
|
assert not loop._is_hallucination("Hello Timmy")
|
|
assert not loop._is_hallucination("What time is it?")
|
|
|
|
|
|
class TestExitCommands:
|
|
"""Voice loop should recognize exit commands."""
|
|
|
|
def test_exit_commands(self):
|
|
loop = VoiceLoop()
|
|
exits = ["goodbye", "exit", "quit", "stop", "goodbye timmy", "stop listening"]
|
|
for cmd in exits:
|
|
assert loop._is_exit_command(cmd), f"'{cmd}' should be an exit command"
|
|
|
|
def test_exit_with_punctuation(self):
|
|
loop = VoiceLoop()
|
|
assert loop._is_exit_command("goodbye!")
|
|
assert loop._is_exit_command("stop.")
|
|
|
|
def test_non_exit_commands(self):
|
|
loop = VoiceLoop()
|
|
assert not loop._is_exit_command("hello")
|
|
assert not loop._is_exit_command("what time is it")
|
|
|
|
|
|
class TestPlayAudio:
|
|
@patch("subprocess.Popen")
|
|
def test_play_audio_calls_afplay(self, mock_popen):
|
|
mock_proc = MagicMock()
|
|
mock_proc.poll.side_effect = [None, 0] # Running, then done
|
|
mock_popen.return_value = mock_proc
|
|
|
|
loop = VoiceLoop()
|
|
loop._play_audio("/tmp/test.wav")
|
|
|
|
mock_popen.assert_called_once()
|
|
args = mock_popen.call_args[0][0]
|
|
assert args[0] == "afplay"
|
|
|
|
@patch("subprocess.Popen")
|
|
def test_play_audio_interruptible(self, mock_popen):
|
|
mock_proc = MagicMock()
|
|
# Simulate running, then we interrupt
|
|
call_count = 0
|
|
|
|
def poll_side_effect():
|
|
nonlocal call_count
|
|
call_count += 1
|
|
return None # Always running
|
|
|
|
mock_proc.poll.side_effect = poll_side_effect
|
|
mock_popen.return_value = mock_proc
|
|
|
|
loop = VoiceLoop()
|
|
loop._interrupted = True # Pre-set interrupt
|
|
loop._play_audio("/tmp/test.wav")
|
|
|
|
mock_proc.terminate.assert_called_once()
|
|
|
|
|
|
class TestStopMethod:
|
|
def test_stop_sets_running_false(self):
|
|
loop = VoiceLoop()
|
|
loop._running = True
|
|
loop.stop()
|
|
assert loop._running is False
|
|
|
|
|
|
class TestSpeakSetsFlag:
|
|
@patch("timmy.voice_loop.VoiceLoop._speak_say")
|
|
def test_speaking_flag_set_during_speech(self, mock_say):
|
|
cfg = VoiceConfig(use_say_fallback=True)
|
|
loop = VoiceLoop(config=cfg)
|
|
|
|
# Before speak
|
|
assert loop._speaking is False
|
|
|
|
# Mock say to check flag during execution
|
|
def check_flag(text):
|
|
assert loop._speaking is True
|
|
|
|
mock_say.side_effect = check_flag
|
|
loop._speak("Hello")
|
|
|
|
# After speak
|
|
assert loop._speaking is False
|