xavierk/voculinux
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 3 Wiki Activity
Files
fa4da40e7b397820a3baffe9fbc64dc148ed92fa
voculinux/tests/test_recognition_manager_core.py
T
Jatin K MalikandGitHub fa4da40e7b test(coverage): improve test coverage from 57% to 80%+ (#327) 
* test(coverage): improve test coverage from 57% to 80%+

Add 30 new test files with 861 additional test cases to improve
overall test coverage from 57% to 80%+ (statement coverage).

Key modules covered:
- recognition_manager.py: 43% → 69% (audio device detection, transcription,
  model download, engine initialization, buffer management)
- text_injector.py: improved coverage for X11/Wayland text injection,
  environment detection, clipboard fallback, shortcut injection
- ibus_engine.py: 22% → 73% (engine lifecycle, text injection, socket server)
- evdev_backend.py: 11% → 78% (device detection, key event handling)
- pynput_backend.py: 43% → 76% (start/stop, key handling, callbacks)
- keyboard_backends/__init__.py: 54% → improved (backend creation, env detection)
- whispercpp_model_info.py: 32% → improved (GPU detection, model recommendations)
- about_dialog.py, first_run_dialog.py, main.py, single_instance.py,
  common_types.py, config_manager.py, autostart_manager.py: various improvements

All 1433 tests pass with proper test isolation to prevent mock pollution
between test modules.

* test(quality): audit and fix test quality issues

Code review audit identified and fixed the following issues:

- Remove bare try/except blocks that swallow all errors (7 tests fixed
  in recognition_manager_deep, replaced with proper assertions or deleted)
- Delete source code inspection tests that read .py files as strings
  instead of testing behavior (deleted test_about_dialog_coverage.py and
  test_first_run_dialog_coverage.py entirely - ~35 useless tests removed)
- Add meaningful assertions replacing weak assertIsNotNone/assertIsInstance
  checks (common_types_coverage, ibus_engine_coverage, keyboard_backends_extra)
- Fix tests that mock the system under test instead of its dependencies
  (text_injector_extra now mocks subprocess.run instead of injection methods)
- Add missing assertions to tests with no verification
  (ibus_engine focus_out, log window info, copy to clipboard)
- Verify download tests check requests.get was called with correct URLs
  (coverage_boost download tests)
- Fix pytest warning by renaming test_audio_input import to avoid
  pytest collecting it as a test function

All 1356 tests pass with 0 failures and 0 warnings.
Statement coverage: 80.3% (3257/4054 lines covered).

* style(tests): fix black formatting and isort import ordering

Auto-format all new test files with black and isort to pass CI lint checks.

* fix(tests): resolve CI hangs and rename test files for clarity

- Fix CI hangs by properly mocking gi/GTK/IBus in all test files
- Fix sys.modules pollution in recognition manager tests by using tmp_path
- Add pytest-timeout to prevent future hangs
- Rename 28 test files to use descriptive, consistent names mapping to source modules

* fix(tests): resolve Python 3.13 CI failures

- Fix _load_stdlib_module() in test_recognition_manager_downloads.py to
  handle stdlib modules that are packages (directories with __init__.py)
  rather than single .py files. In Python 3.13, zipfile became a package,
  causing FileNotFoundError during test collection (exit code 2).

- Fix invalid escape sequence '$' in test_cross_distro_compatibility.py
  by using a raw string. Python 3.13 reports this as SyntaxWarning and
  Python 3.14 will make it a SyntaxError.

* fix(tests): use direct import for zipfile instead of _load_stdlib_module

The _load_stdlib_module approach fails on Python 3.13 where zipfile is a
package (zipfile/__init__.py). Loading it via spec_from_file_location with
a custom name breaks relative imports inside the package (e.g.
'from ._path import ...' resolves to '_stdlib_zipfile._path' which doesn't
exist).

Simply import zipfile at module level before any test mocks can pollute
sys.modules — this is safe because the module-level code runs during test
collection, before any test fixtures or patches are applied.

* fix(tests): mock gi/GTK in conftest to prevent CI hangs

On CI, real PyGObject is installed and module-level imports in vocalinux
source files (ibus_engine, tray_indicator, settings_dialog, etc.) trigger
GTK initialization or IBus daemon connections that hang indefinitely in
headless environments. pytest-timeout's signal-based timeout cannot
interrupt these C-level blocking calls.

Fix by mocking gi and gi.repository in conftest.py (which runs before any
test module is collected), ensuring all test files are protected regardless
of whether they remember to mock individually.

Also add timeout-minutes: 10 to the CI test job as a safety net so that
any remaining hangs fail fast instead of running indefinitely.

* fix(tests): unconditionally mock gi in conftest for CI

The previous 'if gi not in sys.modules' guard was ineffective on CI
because PyGObject is installed via pip, so the real gi module is already
loaded. This caused GTK initialization and IBus daemon connections that
hang indefinitely in headless CI environments.

Now unconditionally replace gi and gi.repository with mocks during test
collection, regardless of whether the real package is installed.

* fix(tests): reduce timeout to 10s and use thread-based timeout

Use timeout_method=thread instead of signal (default) so that
pytest-timeout can interrupt C-level blocking calls (GTK operations,
socket accepts, subprocess waits) that the signal-based method cannot
handle. Also reduce per-test timeout from 30s to 10s since no legitimate
unit test should take longer, and hanging tests were consuming the entire
CI job timeout budget.

* fix(tests): set global socket timeout to prevent blocking on CI

Tests in test_ibus_engine.py create real Unix sockets with daemon threads
that block on socket.accept(). When tests fail or timeout, these threads
remain blocked indefinitely, preventing pytest from completing. The
thread-based timeout catches the hang but cannot terminate the blocking
accept() call.

Set socket.setdefaulttimeout(5) in conftest.py so all socket operations
(accept, connect, recv) will raise socket.timeout after 5 seconds instead
of blocking forever.

* fix(tests): add cleanup fixture for IBus socket server threads

Add autouse fixture that stops any leftover VocalinuxEngine socket server
threads after each test. Tests that trigger _start_socket_server() spawn
daemon threads blocking on socket.accept() that persist after the test
completes, eventually causing pytest-timeout to kill the process.

The cleanup fixture sets _server_running=False and closes the server
socket, which unblocks the accept() call and allows the daemon thread
to terminate.

* fix(tests): reduce socket timeout to 0.1s to avoid slowing tests

The 5-second socket timeout caused socket operations in daemon threads
to block for 5 seconds per iteration before timing out, significantly
slowing the test suite on CI. Reduce to 0.1s so blocking socket calls
fail fast without noticeably impacting test execution time.

* fix(tests): use 2s socket timeout and increase CI timeout to 15min

The 0.1s socket timeout was too aggressive and caused legitimate test
socket operations to fail. Use 2s which is long enough for test sockets
to work but short enough to unblock daemon threads.

Increase CI job timeout to 15 minutes to accommodate the socket timeout
overhead from daemon threads that retry every 2 seconds.

* fix(tests): block known-blocking daemon threads to prevent CI hangs

Instead of using socket timeouts (which either break tests or are too
slow), prevent specific known-blocking daemon threads from starting:
- ibus_engine.server_thread: blocks on socket.accept()
- evdev_backend._monitor_devices: blocks on select.select()

These threads are identified by their target function name and skipped
during Thread.start(). The thread is marked as started/stopped so
join() calls in cleanup code don't raise RuntimeError.

Test-created daemon threads (e.g. mock socket servers) are unaffected
since they use different target function names.

* fix(tests): patch Thread.join for Python 3.13 compatibility

Python 3.13 changed threading internals — join() now uses _handle.join()
which raises 'thread not started' for threads that were never actually
started. Instead of trying to fake internal state, patch join() to be a
no-op for threads that were skipped by our _safe_thread_start guard.

This fixes the 2 remaining test_evdev_backend_ext failures on Python 3.13.
2026-03-17 14:05:16 -07:00

932 lines
36 KiB
Python
Raw Blame History

"""
Additional coverage tests for speech recognition manager.
This file supplements test_recognition_manager.py with tests for uncovered areas.
Focus on edge cases, error paths, and helper functions.
"""
import os
import sys
import threading
import unittest
from unittest.mock import MagicMock, Mock, call, patch
# Mock modules before importing any modules that might use them
sys.modules["vosk"] = MagicMock()
sys.modules["whisper"] = MagicMock()
mock_pywhispercpp = MagicMock()
mock_pywhispercpp.model = MagicMock()
mock_pywhispercpp.model.Model = MagicMock()
sys.modules["pywhispercpp"] = mock_pywhispercpp
sys.modules["pywhispercpp.model"] = mock_pywhispercpp.model
sys.modules["requests"] = MagicMock()
sys.modules["pyaudio"] = MagicMock()
sys.modules["wave"] = MagicMock()
sys.modules["tqdm"] = MagicMock()
sys.modules["numpy"] = MagicMock()
sys.modules["torch"] = MagicMock()
sys.modules["psutil"] = MagicMock()
# Import the shared mock from conftest
from conftest import mock_audio_feedback # noqa: E402
# Update import paths to use the new package structure
from vocalinux.common_types import RecognitionState # noqa: E402
from vocalinux.speech_recognition.command_processor import CommandProcessor # noqa: E402
from vocalinux.speech_recognition.recognition_manager import ( # noqa: E402
SpeechRecognitionManager,
_get_supported_channels,
_get_supported_sample_rate,
get_audio_input_devices,
)
class TestGetAudioInputDevices(unittest.TestCase):
"""Test cases for get_audio_input_devices function."""
def test_get_audio_input_devices_success(self):
"""Test successfully retrieving audio input devices."""
# Create mock for PyAudio
mock_device_info = {
"index": 0,
"name": "Built-in Microphone",
"maxInputChannels": 2,
}
mock_pyaudio_instance = MagicMock()
mock_pyaudio_instance.get_device_count.return_value = 2
mock_pyaudio_instance.get_device_info_by_index.side_effect = [
mock_device_info,
{"index": 1, "name": "USB Microphone", "maxInputChannels": 1},
]
mock_pyaudio_instance.get_default_input_device_info.return_value = {"index": 0}
mock_pyaudio = MagicMock()
mock_pyaudio.PyAudio.return_value = mock_pyaudio_instance
with patch.dict(sys.modules, {"pyaudio": mock_pyaudio}):
devices = get_audio_input_devices()
self.assertEqual(len(devices), 2)
self.assertEqual(devices[0][0], 0)
self.assertEqual(devices[0][1], "Built-in Microphone")
self.assertTrue(devices[0][2]) # Is default
self.assertFalse(devices[1][2]) # Not default
def test_get_audio_input_devices_no_default(self):
"""Test when no default input device is available."""
mock_pyaudio_instance = MagicMock()
mock_pyaudio_instance.get_device_count.return_value = 1
mock_pyaudio_instance.get_device_info_by_index.return_value = {
"index": 0,
"name": "Microphone",
"maxInputChannels": 1,
}
mock_pyaudio_instance.get_default_input_device_info.side_effect = IOError("No default")
mock_pyaudio = MagicMock()
mock_pyaudio.PyAudio.return_value = mock_pyaudio_instance
with patch.dict(sys.modules, {"pyaudio": mock_pyaudio}):
devices = get_audio_input_devices()
self.assertEqual(len(devices), 1)
self.assertFalse(devices[0][2]) # Not default
def test_get_audio_input_devices_skip_no_input_channels(self):
"""Test that devices with no input channels are skipped."""
mock_pyaudio_instance = MagicMock()
mock_pyaudio_instance.get_device_count.return_value = 2
mock_pyaudio_instance.get_device_info_by_index.side_effect = [
{"index": 0, "name": "Speaker", "maxInputChannels": 0},
{"index": 1, "name": "Microphone", "maxInputChannels": 2},
]
mock_pyaudio_instance.get_default_input_device_info.return_value = {"index": 1}
mock_pyaudio = MagicMock()
mock_pyaudio.PyAudio.return_value = mock_pyaudio_instance
with patch.dict(sys.modules, {"pyaudio": mock_pyaudio}):
devices = get_audio_input_devices()
# Should only have the microphone, not the speaker
self.assertEqual(len(devices), 1)
self.assertEqual(devices[0][1], "Microphone")
def test_get_audio_input_devices_oserror(self):
"""Test handling of OSError when getting device info."""
mock_pyaudio_instance = MagicMock()
mock_pyaudio_instance.get_device_count.return_value = 2
mock_pyaudio_instance.get_device_info_by_index.side_effect = [
OSError("Device error"),
{"index": 1, "name": "Microphone", "maxInputChannels": 1},
]
mock_pyaudio_instance.get_default_input_device_info.return_value = {"index": 1}
mock_pyaudio = MagicMock()
mock_pyaudio.PyAudio.return_value = mock_pyaudio_instance
with patch.dict(sys.modules, {"pyaudio": mock_pyaudio}):
devices = get_audio_input_devices()
# Should skip the device with error and include the good one
self.assertEqual(len(devices), 1)
self.assertEqual(devices[0][1], "Microphone")
def test_get_audio_input_devices_import_error(self):
"""Test handling when PyAudio is not installed."""
with patch.dict(sys.modules, {"pyaudio": None}):
# Need to reload the module to test import error handling
# For now, just verify the function handles it gracefully
devices = get_audio_input_devices()
self.assertEqual(devices, [])
def test_get_audio_input_devices_exception(self):
"""Test handling of general exceptions."""
mock_pyaudio = MagicMock()
mock_pyaudio.PyAudio.side_effect = RuntimeError("Unexpected error")
with patch.dict(sys.modules, {"pyaudio": mock_pyaudio}):
devices = get_audio_input_devices()
self.assertEqual(devices, [])
class TestGetSupportedChannels(unittest.TestCase):
"""Test cases for _get_supported_channels helper function."""
def test_get_supported_channels_mono(self):
"""Test detection of mono support."""
mock_stream = MagicMock()
mock_audio = MagicMock()
mock_audio.open.return_value = mock_stream
mock_audio.paInt16 = 8
channels = _get_supported_channels(mock_audio, 0)
self.assertEqual(channels, 1)
mock_stream.close.assert_called_once()
def test_get_supported_channels_stereo(self):
"""Test detection of stereo support."""
mock_stream = MagicMock()
mock_audio = MagicMock()
mock_audio.paInt16 = 8
# First call (mono) fails, second (stereo) succeeds
call_count = [0]
def open_side_effect(**kwargs):
call_count[0] += 1
if kwargs.get("channels") == 1:
raise IOError("invalid number of channels")
return mock_stream
mock_audio.open.side_effect = open_side_effect
channels = _get_supported_channels(mock_audio, 0)
self.assertEqual(channels, 2)
def test_get_supported_channels_default_fallback(self):
"""Test fallback to mono when no channels are supported."""
mock_audio = MagicMock()
mock_audio.paInt16 = 8
mock_audio.open.side_effect = IOError("Invalid channels")
channels = _get_supported_channels(mock_audio, 0)
self.assertEqual(channels, 1)
def test_get_supported_channels_with_device_index(self):
"""Test channel detection with specific device index."""
mock_stream = MagicMock()
mock_audio = MagicMock()
mock_audio.open.return_value = mock_stream
mock_audio.paInt16 = 8
channels = _get_supported_channels(mock_audio, device_index=2)
self.assertEqual(channels, 1)
# Verify device index was passed to open
mock_audio.open.assert_called_once()
call_kwargs = mock_audio.open.call_args[1]
self.assertEqual(call_kwargs["input_device_index"], 2)
class TestGetSupportedSampleRate(unittest.TestCase):
"""Test cases for _get_supported_sample_rate helper function."""
def test_get_supported_sample_rate_default(self):
"""Test getting default sample rate from device."""
mock_stream = MagicMock()
mock_audio = MagicMock()
mock_audio.get_device_info_by_index.return_value = {"defaultSampleRate": 16000}
mock_audio.open.return_value = mock_stream
mock_audio.paInt16 = 8
rate = _get_supported_sample_rate(mock_audio, 0, channels=1)
self.assertEqual(rate, 16000)
def test_get_supported_sample_rate_fallback(self):
"""Test fallback to common rates when default fails."""
mock_audio = MagicMock()
mock_audio.get_device_info_by_index.return_value = {"defaultSampleRate": 44100}
mock_audio.paInt16 = 8
# Make default rate fail, but succeed on first common rate (48000)
call_count = [0]
def open_side_effect(**kwargs):
call_count[0] += 1
if kwargs.get("rate") == 44100:
raise IOError("Rate not supported")
return MagicMock()
mock_audio.open.side_effect = open_side_effect
rate = _get_supported_sample_rate(mock_audio, 0, channels=1)
# Should return first working rate (48000 is first in COMMON_RATES)
self.assertIn(rate, [48000, 44100, 16000, 8000])
def test_get_supported_sample_rate_default_fallback(self):
"""Test fallback to 16000Hz when nothing works."""
mock_audio = MagicMock()
mock_audio.get_device_info_by_index.return_value = {"defaultSampleRate": 48000}
mock_audio.paInt16 = 8
mock_audio.open.side_effect = IOError("All rates failed")
rate = _get_supported_sample_rate(mock_audio, 0, channels=1)
self.assertEqual(rate, 16000)
class TestBufferManagement(unittest.TestCase):
"""Test cases for buffer management methods."""
def setUp(self):
"""Set up test fixtures."""
# Create patches for manager initialization
self.mockKaldi = patch.object(sys.modules["vosk"], "KaldiRecognizer")
self.mockModel = patch.object(sys.modules["vosk"], "Model")
self.mockMakeDirs = patch("os.makedirs")
self.mockThread = patch("threading.Thread")
self.mockPath = patch.object(SpeechRecognitionManager, "_get_vosk_model_path")
self.mockDownload = patch.object(SpeechRecognitionManager, "_download_vosk_model")
self.kaldiMock = self.mockKaldi.start()
self.modelMock = self.mockModel.start()
self.makeDirsMock = self.mockMakeDirs.start()
self.threadMock = self.mockThread.start()
self.pathMock = self.mockPath.start()
self.downloadMock = self.mockDownload.start()
self.pathMock.return_value = "/mock/path/vosk-model"
self.recognizerMock = MagicMock()
self.kaldiMock.return_value = self.recognizerMock
self.recognizerMock.FinalResult.return_value = '{"text": ""}'
self.threadInstance = MagicMock()
self.threadMock.return_value = self.threadInstance
# Reset audio feedback mocks
mock_audio_feedback.play_start_sound.reset_mock()
mock_audio_feedback.play_stop_sound.reset_mock()
mock_audio_feedback.play_error_sound.reset_mock()
# Patch os.path functions
self.patcher_exists = patch("os.path.exists", return_value=True)
self.mock_exists = self.patcher_exists.start()
def tearDown(self):
"""Clean up patches."""
self.mockKaldi.stop()
self.mockModel.stop()
self.mockMakeDirs.stop()
self.mockThread.stop()
self.mockPath.stop()
self.mockDownload.stop()
self.patcher_exists.stop()
def test_set_buffer_limit(self):
"""Test setting buffer limit."""
manager = SpeechRecognitionManager(engine="vosk")
# Default should be 5000 (max buffer size in implementation)
self.assertEqual(manager._max_buffer_size, 5000)
# Just verify the manager has the buffer tracking
self.assertIsNotNone(manager.audio_buffer)
def test_get_buffer_stats(self):
"""Test getting buffer statistics."""
manager = SpeechRecognitionManager(engine="vosk")
# Empty buffer
stats = manager.get_buffer_stats()
self.assertIsNotNone(stats)
self.assertIn("buffer_size", stats)
self.assertIn("buffer_limit", stats)
self.assertIn("memory_usage_bytes", stats)
self.assertEqual(stats["buffer_size"], 0)
# Add data to buffer
manager.audio_buffer = [b"\x00" * 1024, b"\x00" * 1024]
stats = manager.get_buffer_stats()
self.assertIsNotNone(stats)
self.assertGreater(stats["buffer_size"], 0)
def test_get_buffer_stats_at_limit(self):
"""Test buffer stats when at limit."""
manager = SpeechRecognitionManager(engine="vosk")
# Fill buffer
manager.audio_buffer = [b"\x00" * 1024 for _ in range(100)]
stats = manager.get_buffer_stats()
self.assertIsNotNone(stats)
self.assertGreater(stats["buffer_size"], 0)
class TestProcessPartialResult(unittest.TestCase):
"""Test cases for _process_partial_result method."""
def setUp(self):
"""Set up test fixtures."""
self.mockKaldi = patch.object(sys.modules["vosk"], "KaldiRecognizer")
self.mockModel = patch.object(sys.modules["vosk"], "Model")
self.mockMakeDirs = patch("os.makedirs")
self.mockThread = patch("threading.Thread")
self.mockPath = patch.object(SpeechRecognitionManager, "_get_vosk_model_path")
self.mockDownload = patch.object(SpeechRecognitionManager, "_download_vosk_model")
self.kaldiMock = self.mockKaldi.start()
self.modelMock = self.mockModel.start()
self.makeDirsMock = self.mockMakeDirs.start()
self.threadMock = self.mockThread.start()
self.pathMock = self.mockPath.start()
self.downloadMock = self.mockDownload.start()
self.pathMock.return_value = "/mock/path/vosk-model"
self.recognizerMock = MagicMock()
self.kaldiMock.return_value = self.recognizerMock
self.recognizerMock.FinalResult.return_value = '{"text": ""}'
self.threadInstance = MagicMock()
self.threadMock.return_value = self.threadInstance
mock_audio_feedback.play_start_sound.reset_mock()
mock_audio_feedback.play_stop_sound.reset_mock()
mock_audio_feedback.play_error_sound.reset_mock()
self.patcher_exists = patch("os.path.exists", return_value=True)
self.mock_exists = self.patcher_exists.start()
def tearDown(self):
"""Clean up patches."""
self.mockKaldi.stop()
self.mockModel.stop()
self.mockMakeDirs.stop()
self.mockThread.stop()
self.mockPath.stop()
self.mockDownload.stop()
self.patcher_exists.stop()
def test_process_final_buffer_vosk_with_text(self):
"""Test processing final buffer with VOSK with text result."""
manager = SpeechRecognitionManager(engine="vosk")
manager.audio_buffer = [b"\x00" * 512]
# Mock final result with text
self.recognizerMock.FinalResult.return_value = '{"text": "hello world"}'
callback_mock = MagicMock()
manager.register_text_callback(callback_mock)
manager._process_final_buffer()
# Buffer should be cleared
self.assertEqual(manager.audio_buffer, [])
def test_process_final_buffer_empty_no_callback(self):
"""Test processing empty final buffer does not call callbacks."""
manager = SpeechRecognitionManager(engine="vosk")
manager.audio_buffer = []
callback_mock = MagicMock()
manager.register_text_callback(callback_mock)
manager._process_final_buffer()
# Callback should not be called for empty buffer
callback_mock.assert_not_called()
def test_recognizer_none_during_processing(self):
"""Test handling when recognizer becomes None during processing."""
manager = SpeechRecognitionManager(engine="vosk")
manager.audio_buffer = [b"\x00" * 512]
# Set recognizer to None
manager.recognizer = None
callback_mock = MagicMock()
manager.register_text_callback(callback_mock)
# Should handle gracefully without crashing
manager._process_final_buffer()
# Callback should not be called
callback_mock.assert_not_called()
class TestStartStopRecognition(unittest.TestCase):
"""Test cases for start_recognition and stop_recognition flows."""
def setUp(self):
"""Set up test fixtures."""
self.mockKaldi = patch.object(sys.modules["vosk"], "KaldiRecognizer")
self.mockModel = patch.object(sys.modules["vosk"], "Model")
self.mockMakeDirs = patch("os.makedirs")
self.mockThread = patch("threading.Thread")
self.mockPath = patch.object(SpeechRecognitionManager, "_get_vosk_model_path")
self.mockDownload = patch.object(SpeechRecognitionManager, "_download_vosk_model")
self.kaldiMock = self.mockKaldi.start()
self.modelMock = self.mockModel.start()
self.makeDirsMock = self.mockMakeDirs.start()
self.threadMock = self.mockThread.start()
self.pathMock = self.mockPath.start()
self.downloadMock = self.mockDownload.start()
self.pathMock.return_value = "/mock/path/vosk-model"
self.recognizerMock = MagicMock()
self.kaldiMock.return_value = self.recognizerMock
self.recognizerMock.FinalResult.return_value = '{"text": ""}'
self.threadInstance = MagicMock()
self.threadMock.return_value = self.threadInstance
mock_audio_feedback.play_start_sound.reset_mock()
mock_audio_feedback.play_stop_sound.reset_mock()
mock_audio_feedback.play_error_sound.reset_mock()
self.patcher_exists = patch("os.path.exists", return_value=True)
self.mock_exists = self.patcher_exists.start()
def tearDown(self):
"""Clean up patches."""
self.mockKaldi.stop()
self.mockModel.stop()
self.mockMakeDirs.stop()
self.mockThread.stop()
self.mockPath.stop()
self.mockDownload.stop()
self.patcher_exists.stop()
def test_start_recognition_success(self):
"""Test successful recognition start."""
manager = SpeechRecognitionManager(engine="vosk")
manager.start_recognition()
self.assertEqual(manager.state, RecognitionState.LISTENING)
self.assertTrue(manager.should_record)
mock_audio_feedback.play_start_sound.assert_called_once()
def test_start_recognition_already_listening(self):
"""Test starting when already listening."""
manager = SpeechRecognitionManager(engine="vosk")
# First start
manager.start_recognition()
self.threadMock.reset_mock()
self.threadInstance.reset_mock()
# Second start should not create new threads
manager.audio_thread = self.threadInstance
manager.recognition_thread = self.threadInstance
manager.start_recognition()
# No new threads created
self.threadMock.assert_not_called()
def test_stop_recognition_success(self):
"""Test successful recognition stop."""
manager = SpeechRecognitionManager(engine="vosk")
manager.start_recognition()
manager.audio_thread = self.threadInstance
manager.recognition_thread = self.threadInstance
manager.stop_recognition()
self.assertEqual(manager.state, RecognitionState.IDLE)
self.assertFalse(manager.should_record)
mock_audio_feedback.play_stop_sound.assert_called_once()
def test_stop_recognition_when_idle(self):
"""Test stopping when already idle."""
manager = SpeechRecognitionManager(engine="vosk")
# Verify initial state is idle
self.assertEqual(manager.state, RecognitionState.IDLE)
# Stop should do nothing
manager.stop_recognition()
self.assertEqual(manager.state, RecognitionState.IDLE)
mock_audio_feedback.play_stop_sound.assert_not_called()
class TestWhisperInitialization(unittest.TestCase):
"""Test cases for Whisper engine initialization."""
def setUp(self):
"""Set up test fixtures."""
self.mockKaldi = patch.object(sys.modules["vosk"], "KaldiRecognizer")
self.mockModel = patch.object(sys.modules["vosk"], "Model")
self.mockMakeDirs = patch("os.makedirs")
self.mockThread = patch("threading.Thread")
self.mockPath = patch.object(SpeechRecognitionManager, "_get_vosk_model_path")
self.mockDownload = patch.object(SpeechRecognitionManager, "_download_vosk_model")
self.kaldiMock = self.mockKaldi.start()
self.modelMock = self.mockModel.start()
self.makeDirsMock = self.mockMakeDirs.start()
self.threadMock = self.mockThread.start()
self.pathMock = self.mockPath.start()
self.downloadMock = self.mockDownload.start()
self.threadInstance = MagicMock()
self.threadMock.return_value = self.threadInstance
mock_audio_feedback.play_start_sound.reset_mock()
mock_audio_feedback.play_stop_sound.reset_mock()
mock_audio_feedback.play_error_sound.reset_mock()
self.patcher_exists = patch("os.path.exists", return_value=False)
self.mock_exists = self.patcher_exists.start()
def tearDown(self):
"""Clean up patches."""
self.mockKaldi.stop()
self.mockModel.stop()
self.mockMakeDirs.stop()
self.mockThread.stop()
self.mockPath.stop()
self.mockDownload.stop()
self.patcher_exists.stop()
def test_whisper_invalid_model_size(self):
"""Test Whisper with invalid model size."""
whisper_mock = MagicMock()
torch_mock = MagicMock()
whisper_mock.load_model = MagicMock()
torch_mock.cuda.is_available.return_value = False
with patch.dict(sys.modules, {"whisper": whisper_mock, "torch": torch_mock}):
# Should convert invalid model to "base"
manager = SpeechRecognitionManager(
engine="whisper", model_size="invalid_size", defer_download=True
)
# The implementation converts invalid sizes to "base"
self.assertEqual(manager.model_size, "base")
def test_whisper_defer_download(self):
"""Test Whisper with defer_download enabled."""
whisper_mock = MagicMock()
torch_mock = MagicMock()
torch_mock.cuda.is_available.return_value = False
with patch.dict(sys.modules, {"whisper": whisper_mock, "torch": torch_mock}):
with patch("os.path.exists", return_value=False):
manager = SpeechRecognitionManager(
engine="whisper", model_size="tiny", defer_download=True
)
self.assertFalse(manager._model_initialized)
def test_whisper_cuda_available(self):
"""Test Whisper with CUDA available."""
whisper_mock = MagicMock()
torch_mock = MagicMock()
torch_mock.cuda.is_available.return_value = True
whisper_mock.load_model = MagicMock(return_value=MagicMock())
with patch.dict(sys.modules, {"whisper": whisper_mock, "torch": torch_mock}):
with patch("os.path.exists", return_value=True):
manager = SpeechRecognitionManager(engine="whisper", model_size="tiny")
self.assertEqual(manager.engine, "whisper")
class TestWhispercppInitialization(unittest.TestCase):
"""Test cases for whisper.cpp engine initialization."""
def setUp(self):
"""Set up test fixtures."""
self.mockKaldi = patch.object(sys.modules["vosk"], "KaldiRecognizer")
self.mockModel = patch.object(sys.modules["vosk"], "Model")
self.mockMakeDirs = patch("os.makedirs")
self.mockThread = patch("threading.Thread")
self.mockPath = patch.object(SpeechRecognitionManager, "_get_vosk_model_path")
self.mockDownload = patch.object(SpeechRecognitionManager, "_download_vosk_model")
self.kaldiMock = self.mockKaldi.start()
self.modelMock = self.mockModel.start()
self.makeDirsMock = self.mockMakeDirs.start()
self.threadMock = self.mockThread.start()
self.pathMock = self.mockPath.start()
self.downloadMock = self.mockDownload.start()
self.threadInstance = MagicMock()
self.threadMock.return_value = self.threadInstance
mock_audio_feedback.play_start_sound.reset_mock()
mock_audio_feedback.play_stop_sound.reset_mock()
mock_audio_feedback.play_error_sound.reset_mock()
self.patcher_exists = patch("os.path.exists", return_value=False)
self.mock_exists = self.patcher_exists.start()
def tearDown(self):
"""Clean up patches."""
self.mockKaldi.stop()
self.mockModel.stop()
self.mockMakeDirs.stop()
self.mockThread.stop()
self.mockPath.stop()
self.mockDownload.stop()
self.patcher_exists.stop()
def test_whispercpp_invalid_model_size(self):
"""Test whisper.cpp with invalid model size."""
mock_pywhispercpp = MagicMock()
mock_pywhispercpp.model.Model = MagicMock()
mock_backend = MagicMock()
mock_backend.name = "CPU"
with patch.dict(
sys.modules,
{"pywhispercpp": mock_pywhispercpp, "pywhispercpp.model": mock_pywhispercpp.model},
):
with patch(
"vocalinux.speech_recognition.recognition_manager.get_model_path",
return_value="/path/model.bin",
):
with patch(
"vocalinux.utils.whispercpp_model_info.detect_compute_backend",
return_value=(mock_backend, {}),
):
with patch("os.path.exists", return_value=True):
with patch("os.stat") as mock_stat:
mock_stat.return_value.st_size = 1000000
# Should convert invalid model to "tiny"
manager = SpeechRecognitionManager(
engine="whisper_cpp", model_size="invalid", defer_download=True
)
# Invalid size should be converted to "tiny"
self.assertEqual(manager.model_size, "tiny")
def test_whispercpp_defer_download(self):
"""Test whisper.cpp with defer_download enabled."""
mock_pywhispercpp = MagicMock()
mock_pywhispercpp.model.Model = MagicMock()
with patch.dict(
sys.modules,
{"pywhispercpp": mock_pywhispercpp, "pywhispercpp.model": mock_pywhispercpp.model},
):
# Mock get_model_path to return a non-existent path
with patch(
"vocalinux.speech_recognition.recognition_manager.get_model_path",
return_value="/nonexistent/model.bin",
):
with patch("os.path.exists", return_value=False):
manager = SpeechRecognitionManager(
engine="whisper_cpp", model_size="tiny", defer_download=True
)
self.assertFalse(manager._model_initialized)
class TestReconfiguration(unittest.TestCase):
"""Test cases for reconfigure method."""
def setUp(self):
"""Set up test fixtures."""
self.mockKaldi = patch.object(sys.modules["vosk"], "KaldiRecognizer")
self.mockModel = patch.object(sys.modules["vosk"], "Model")
self.mockMakeDirs = patch("os.makedirs")
self.mockThread = patch("threading.Thread")
self.mockPath = patch.object(SpeechRecognitionManager, "_get_vosk_model_path")
self.mockDownload = patch.object(SpeechRecognitionManager, "_download_vosk_model")
self.kaldiMock = self.mockKaldi.start()
self.modelMock = self.mockModel.start()
self.makeDirsMock = self.mockMakeDirs.start()
self.threadMock = self.mockThread.start()
self.pathMock = self.mockPath.start()
self.downloadMock = self.mockDownload.start()
self.pathMock.return_value = "/mock/path/vosk-model"
self.recognizerMock = MagicMock()
self.kaldiMock.return_value = self.recognizerMock
self.recognizerMock.FinalResult.return_value = '{"text": ""}'
self.threadInstance = MagicMock()
self.threadMock.return_value = self.threadInstance
mock_audio_feedback.play_start_sound.reset_mock()
mock_audio_feedback.play_stop_sound.reset_mock()
mock_audio_feedback.play_error_sound.reset_mock()
self.patcher_exists = patch("os.path.exists", return_value=True)
self.mock_exists = self.patcher_exists.start()
def tearDown(self):
"""Clean up patches."""
self.mockKaldi.stop()
self.mockModel.stop()
self.mockMakeDirs.stop()
self.mockThread.stop()
self.mockPath.stop()
self.mockDownload.stop()
self.patcher_exists.stop()
def test_reconfigure_while_listening(self):
"""Test reconfiguring while listening."""
manager = SpeechRecognitionManager(engine="vosk")
manager.audio_thread = self.threadInstance
manager.recognition_thread = self.threadInstance
# Start listening
manager.start_recognition()
self.assertEqual(manager.state, RecognitionState.LISTENING)
# Reconfigure engine change should work
manager.reconfigure(engine="vosk", model_size="medium", force_download=False)
# After reconfigure, should still be initialized
self.assertEqual(manager.engine, "vosk")
self.assertEqual(manager.model_size, "medium")
def test_reconfigure_language_only(self):
"""Test reconfiguring only language."""
manager = SpeechRecognitionManager(engine="vosk", language="en-us")
self.assertEqual(manager.language, "en-us")
manager.reconfigure(language="de", force_download=False)
self.assertEqual(manager.language, "de")
def test_reconfigure_model_size_only(self):
"""Test reconfiguring only model size."""
manager = SpeechRecognitionManager(engine="vosk", model_size="small")
self.assertEqual(manager.model_size, "small")
manager.reconfigure(model_size="medium", force_download=False)
self.assertEqual(manager.model_size, "medium")
def test_reconfigure_no_changes(self):
"""Test reconfiguring with no actual changes."""
manager = SpeechRecognitionManager(engine="vosk", model_size="small")
# Reconfigure with same values
manager.reconfigure(model_size="small", force_download=False)
# Should have no issues
self.assertEqual(manager.model_size, "small")
class TestProcessFinalBuffer(unittest.TestCase):
"""Test cases for _process_final_buffer method."""
def setUp(self):
"""Set up test fixtures."""
self.mockKaldi = patch.object(sys.modules["vosk"], "KaldiRecognizer")
self.mockModel = patch.object(sys.modules["vosk"], "Model")
self.mockMakeDirs = patch("os.makedirs")
self.mockThread = patch("threading.Thread")
self.mockPath = patch.object(SpeechRecognitionManager, "_get_vosk_model_path")
self.mockDownload = patch.object(SpeechRecognitionManager, "_download_vosk_model")
self.kaldiMock = self.mockKaldi.start()
self.modelMock = self.mockModel.start()
self.makeDirsMock = self.mockMakeDirs.start()
self.threadMock = self.mockThread.start()
self.pathMock = self.mockPath.start()
self.downloadMock = self.mockDownload.start()
self.pathMock.return_value = "/mock/path/vosk-model"
self.recognizerMock = MagicMock()
self.kaldiMock.return_value = self.recognizerMock
self.recognizerMock.FinalResult.return_value = '{"text": "hello"}'
self.threadInstance = MagicMock()
self.threadMock.return_value = self.threadInstance
mock_audio_feedback.play_start_sound.reset_mock()
mock_audio_feedback.play_stop_sound.reset_mock()
mock_audio_feedback.play_error_sound.reset_mock()
self.patcher_exists = patch("os.path.exists", return_value=True)
self.mock_exists = self.patcher_exists.start()
def tearDown(self):
"""Clean up patches."""
self.mockKaldi.stop()
self.mockModel.stop()
self.mockMakeDirs.stop()
self.mockThread.stop()
self.mockPath.stop()
self.mockDownload.stop()
self.patcher_exists.stop()
def test_process_final_buffer_vosk_empty(self):
"""Test processing final buffer with VOSK when empty."""
manager = SpeechRecognitionManager(engine="vosk")
manager.audio_buffer = []
# Should return without calling recognizer
manager._process_final_buffer()
self.recognizerMock.AcceptWaveform.assert_not_called()
def test_process_final_buffer_vosk_with_data(self):
"""Test processing final buffer with VOSK with data."""
manager = SpeechRecognitionManager(engine="vosk")
manager.audio_buffer = [b"\x00" * 512, b"\x00" * 512]
# Register callback to verify it's called
callback_mock = MagicMock()
manager.register_text_callback(callback_mock)
manager._process_final_buffer()
# Recognizer should be called with data
self.assertEqual(self.recognizerMock.AcceptWaveform.call_count, 2)
def test_process_final_buffer_vosk_empty_result(self):
"""Test processing final buffer with VOSK returning empty result."""
manager = SpeechRecognitionManager(engine="vosk")
manager.audio_buffer = [b"\x00" * 512]
# Mock empty result
self.recognizerMock.FinalResult.return_value = '{"text": ""}'
callback_mock = MagicMock()
manager.register_text_callback(callback_mock)
manager._process_final_buffer()
# Callback should not be called for empty result
callback_mock.assert_not_called()
class TestDownloadModels(unittest.TestCase):
"""Test cases for model download methods."""
def setUp(self):
"""Set up test fixtures."""
self.mockKaldi = patch.object(sys.modules["vosk"], "KaldiRecognizer")
self.mockModel = patch.object(sys.modules["vosk"], "Model")
self.mockMakeDirs = patch("os.makedirs")
self.mockThread = patch("threading.Thread")
self.mockPath = patch.object(SpeechRecognitionManager, "_get_vosk_model_path")
self.mockDownload = patch.object(SpeechRecognitionManager, "_download_vosk_model")
self.kaldiMock = self.mockKaldi.start()
self.modelMock = self.mockModel.start()
self.makeDirsMock = self.mockMakeDirs.start()
self.threadMock = self.mockThread.start()
self.pathMock = self.mockPath.start()
self.mockDownload.stop() # Stop this one so we test the real method
self.pathMock.return_value = "/mock/path/vosk-model"
self.recognizerMock = MagicMock()
self.kaldiMock.return_value = self.recognizerMock
self.recognizerMock.FinalResult.return_value = '{"text": ""}'
self.threadInstance = MagicMock()
self.threadMock.return_value = self.threadInstance
mock_audio_feedback.play_start_sound.reset_mock()
mock_audio_feedback.play_stop_sound.reset_mock()
mock_audio_feedback.play_error_sound.reset_mock()
self.patcher_exists = patch("os.path.exists", return_value=False)
self.mock_exists = self.patcher_exists.start()
def tearDown(self):
"""Clean up patches."""
self.mockKaldi.stop()
self.mockModel.stop()
self.mockMakeDirs.stop()
self.mockThread.stop()
self.mockPath.stop()
self.patcher_exists.stop()
def test_cancel_download(self):
"""Test cancelling a download."""
manager = SpeechRecognitionManager(engine="vosk")
self.assertFalse(manager._download_cancelled)
manager.cancel_download()
self.assertTrue(manager._download_cancelled)
def test_set_download_progress_callback(self):
"""Test setting download progress callback."""
manager = SpeechRecognitionManager(engine="vosk")
callback = MagicMock()
manager.set_download_progress_callback(callback)
self.assertEqual(manager._download_progress_callback, callback)
# Clear callback
manager.set_download_progress_callback(None)
self.assertIsNone(manager._download_progress_callback)
if __name__ == "__main__":
unittest.main()
Reference in New Issue View Git Blame Copy Permalink