Files
voculinux/tests/test_main.py
T
b71164a214 feat(tray): add recent dictation snippets history menu 🤖🤖🤖 (#487)
* feat(tray): add recent dictation snippets history menu

Add a "Recent Snippets" submenu to the tray indicator that lists recent
dictation sessions. Clicking a snippet copies its full text to the
clipboard for easy re-pasting; a "Clear History" entry empties the list.

Snippets are recorded on recognition, independent of whether text
injection succeeds, so text can be recovered even when injection silently
no-ops (e.g. on some Wayland compositors). One snippet corresponds to one
dictation session (everything said between start and stop); voice commands
are not recorded since they take a separate callback path.

History is kept in memory only (newest-first, capped, never written to
disk) and clears on quit — a deliberate privacy choice for a tool that
sees everything dictated.

Adds a "history" config section (enabled, max_items) and matching
Settings UI under General. Includes unit tests for the new
TranscriptionHistory store.

* fix(history): sanitize saved max_items to keep startup robust

A hand-edited config.json can store a non-numeric or out-of-range
history.max_items. Coerce the value on load (and in TranscriptionHistory
itself) with a positive-int fallback so a malformed preference warns and
falls back to the default instead of aborting startup or crashing the
Settings dialog spin button.

* fix(history): fold late-arriving segments into their originating session

stop_recognition() only waits on the recognition worker for a bounded
time before reporting IDLE, so a slow final segment can still fire its
text callback afterwards — landing in the next session's pending
snippet, or starting a phantom entry of its own.

Track the open session (state transitions, its worker thread, and its
pending segments) under a lock and file each segment by origin: while a
session is open only its own worker's segments accumulate into it; a
segment arriving after finalization extends that session's committed
snippet via the new TranscriptionHistory.extend_latest(), and a segment
from a session that produced nothing forms its own entry instead of
polluting the next one. ERROR now also finalizes the session, so a
partial result is still recorded and the tracker never wedges open.

Covered by new tests driving the callback and state wiring directly:
mid-session LISTENING/PROCESSING transitions, back-to-back sessions,
late callbacks from a lingering worker thread, disabled history, and
invalid saved limits.

* style(ui): type the new history handlers in tray and settings

* fix(history): keep cleared snippets from reappearing via late segments

The recognition worker can still emit a final segment after the manager's
bounded stop wait reported IDLE. Once the user cleared the history, that
late text could not extend the now-empty history and was appended as a
new entry — resurrecting snippets the user explicitly removed.

Give TranscriptionHistory a clear epoch: clear() (and disabling, which
clears) bumps it even when the store is already empty. add() and
extend_latest() accept an expected_epoch and refuse stale epochs
atomically under the store lock, so the check cannot race the clear.

main.py tracks the epoch each session opened under and the epoch of the
most recently ended session. Session commits and every late-segment
write are guarded by them, so text produced before a clear — whether a
session's whole snippet, its straggler segments, or leftovers committed
on the next session's start — cannot re-enter history afterwards.

* test(history): cover clear-epoch rejection and type new signatures

Add regression tests for the cleared-history race: a late segment after
clear() must not reappear, a session ending after a mid-session clear
leaves no snippet, and add/extend_latest refuse a stale expected_epoch
while unguarded writes keep working.

Annotate the new signatures per the repo guide: _boot's extra_settings
and return tuple, every test in the new TestSessionHistoryRecording and
history-menu tray classes, and all methods of the history store tests.

* fix(main): keep post-clear dictation by rebinding the session epoch

* fix(history): drop the boundary segment when a clear lands mid-session

* fix: scope dictation history to post-clear audio via capture-time stamps

The engine now tags each queued audio segment with the monotonic time
capture began and reports it to a new segment_callbacks hook. History
recording drops segments whose capture started before the last clear,
so pre-clear audio that decodes late can no longer resurrect cleared
entries or leak into a later session, while text dictated after a
mid-session clear is kept instead of being discarded by a stale epoch
check.

* fix(settings): keep Test Dictation speech out of transcription history

The mic test swaps text callbacks for its preview, but history recording
runs on segment callbacks, so test speech still landed in Recent
Snippets. The test now saves and clears the segment callback list
alongside the text list — restoring both on success and on the
start-failure path. register/unregister_segment_callback also gain the
missing return annotations and the engine gets symmetric
get/set_segment_callbacks accessors.

* fix(settings): floor test-dictation speech instead of swapping callbacks

Instead of clearing the segment callback list during the mic test —
which would have dropped leftovers from a still-finishing dictation —
the engine exposes test_capture_floor: segments captured at or after
the stamp are test speech and stay out of Recent Snippets, while
earlier-session leftovers continue to file normally.

* refactor(history): narrow change-callback exception surface

---------

Co-authored-by: Jatin K Malik <jatinkrmalik@gmail.com>
2026-09-30 00:06:51 -07:00

1618 lines
62 KiB
Python

"""
Tests for the main module functionality.
"""
import argparse
import sys
import threading
import time
import unittest
from contextlib import ExitStack
from typing import Any, Callable, Dict, Optional, Tuple
from unittest.mock import ANY, MagicMock, patch
# Mock GTK modules before importing vocalinux.main
sys.modules["gi"] = MagicMock()
sys.modules["gi.repository"] = MagicMock()
# Update import to use the new package structure
from vocalinux.common_types import RecognitionState
from vocalinux.main import check_dependencies, main, parse_arguments
from vocalinux.ui.transcription_history import TranscriptionHistory
class TestMainModule(unittest.TestCase):
"""Test cases for the main module."""
def test_parse_arguments_defaults(self):
"""Test argument parsing with defaults."""
# Test with no arguments (model/engine/language will be None without defaults)
with patch("sys.argv", ["vocalinux"]):
args = parse_arguments()
self.assertFalse(args.debug)
self.assertIsNone(args.model) # No default set, loaded from config instead
self.assertIsNone(args.engine)
self.assertIsNone(args.language)
self.assertFalse(args.wayland)
self.assertFalse(args.start_minimized)
def test_parse_arguments_custom(self):
"""Test argument parsing with custom values."""
# Test with custom arguments
with patch(
"sys.argv",
[
"vocalinux",
"--debug",
"--model",
"large",
"--engine",
"whisper",
"--language",
"fr",
"--wayland",
"--start-minimized",
],
):
args = parse_arguments()
self.assertTrue(args.debug)
self.assertEqual(args.model, "large")
self.assertEqual(args.engine, "whisper")
self.assertEqual(args.language, "fr")
self.assertTrue(args.wayland)
self.assertTrue(args.start_minimized)
def test_parse_arguments_model_values(self):
"""Test model parsing for base and exact whisper.cpp model IDs."""
with patch("sys.argv", ["vocalinux", "--model", "small"]):
args = parse_arguments()
self.assertEqual(args.model, "small")
with patch("sys.argv", ["vocalinux", "--model", "medium"]):
args = parse_arguments()
self.assertEqual(args.model, "medium")
with patch("sys.argv", ["vocalinux", "--model", "large"]):
args = parse_arguments()
self.assertEqual(args.model, "large")
with patch("sys.argv", ["vocalinux", "--model", "medium.en-q5_0"]):
args = parse_arguments()
self.assertEqual(args.model, "medium.en-q5_0")
with patch("sys.argv", ["vocalinux", "--model", "large-v3-turbo"]):
args = parse_arguments()
self.assertEqual(args.model, "large-v3-turbo")
def test_parse_arguments_engine_choices(self):
"""Test that engine only accepts valid choices."""
with patch("sys.argv", ["vocalinux", "--engine", "vosk"]):
args = parse_arguments()
self.assertEqual(args.engine, "vosk")
with patch("sys.argv", ["vocalinux", "--engine", "whisper"]):
args = parse_arguments()
self.assertEqual(args.engine, "whisper")
def test_parse_arguments_language_choices(self):
"""Test that language only accepts valid choices from the catalog."""
from vocalinux.main import LANGUAGE_CHOICES
from vocalinux.utils.vosk_model_info import SUPPORTED_LANGUAGES
self.assertEqual(LANGUAGE_CHOICES, tuple(SUPPORTED_LANGUAGES.keys()))
# Spot-check newly exposed languages (issue #565 and catalog expansion).
for lang in ["auto", "en-us", "hu", "ja", "ko", "ar", "nl", "pl", "uk"]:
self.assertIn(lang, LANGUAGE_CHOICES)
with patch("sys.argv", ["vocalinux", "--language", lang]):
args = parse_arguments()
self.assertEqual(args.language, lang)
@patch("vocalinux.main.sys.exit")
@patch("vocalinux.main.check_dependencies")
@patch("vocalinux.main.parse_arguments")
def test_main_exits_on_missing_deps(self, mock_parse, mock_check_deps, mock_exit):
"""Test that main exits when dependencies are missing."""
mock_check_deps.return_value = False
mock_args = MagicMock()
mock_args.debug = False
mock_parse.return_value = mock_args
# Make sys.exit raise SystemExit to stop execution
mock_exit.side_effect = SystemExit(1)
with patch("vocalinux.main.logger"):
try:
main()
except SystemExit:
pass
mock_exit.assert_called_with(1)
@patch("vocalinux.main.check_dependencies")
@patch("vocalinux.main.parse_arguments")
@patch("vocalinux.main.sys.exit")
@patch("vocalinux.ui.config_manager.get_shared_config_manager")
@patch("vocalinux.ui.logging_manager.initialize_logging")
def test_main_exits_on_init_error(
self, mock_init_logging, mock_config, mock_exit, mock_parse, mock_check_deps
):
"""Test that main exits when initialization fails."""
mock_check_deps.return_value = True
mock_args = MagicMock()
mock_args.debug = False
mock_args.model = "small"
mock_args.engine = "vosk"
mock_args.language = "en-us"
mock_args.wayland = False
mock_parse.return_value = mock_args
# Mock config
mock_config_instance = MagicMock()
mock_config_instance.get_settings.return_value = {
"general": {"first_run": False},
}
mock_config.return_value = mock_config_instance
# Make SpeechRecognitionManager raise an exception
with patch(
"vocalinux.speech_recognition.recognition_manager.SpeechRecognitionManager",
side_effect=Exception("Init error"),
):
with patch("vocalinux.main.logger"):
main()
mock_exit.assert_called_once_with(1)
@patch("vocalinux.main.check_dependencies")
@patch("vocalinux.ui.action_handler.ActionHandler")
@patch("vocalinux.speech_recognition.recognition_manager.SpeechRecognitionManager")
@patch("vocalinux.text_injection.text_injector.TextInjector")
@patch("vocalinux.ui.tray_indicator.TrayIndicator")
@patch("vocalinux.main.logging")
@patch("vocalinux.ui.config_manager.get_shared_config_manager")
@patch("vocalinux.ui.logging_manager.initialize_logging")
def test_main_initializes_components(
self,
mock_init_logging,
mock_config_manager,
mock_logging,
mock_tray,
mock_text,
mock_speech,
mock_action_handler,
mock_check_deps,
):
"""Test that main initializes all the required components."""
# Mock dependency check to return True
mock_check_deps.return_value = True
# Mock ConfigManager to return empty settings (use command-line defaults)
mock_config_instance = MagicMock()
mock_config_instance.get_bool.return_value = True
mock_config_instance.get_settings.return_value = {
"speech_recognition": {},
"general": {"first_run": False},
}
mock_config_instance.get_model_size_for_engine.return_value = "medium"
mock_config_manager.return_value = mock_config_instance
# Mock objects
mock_speech_instance = MagicMock()
mock_text_instance = MagicMock()
mock_tray_instance = MagicMock()
mock_action_instance = MagicMock()
# Setup return values
mock_speech.return_value = mock_speech_instance
mock_text.return_value = mock_text_instance
mock_tray.return_value = mock_tray_instance
mock_action_handler.return_value = mock_action_instance
# Mock the arguments
with patch("vocalinux.main.parse_arguments") as mock_parse:
mock_args = MagicMock()
mock_args.debug = False
mock_args.model = "medium"
mock_args.engine = "vosk"
mock_args.language = "en-us"
mock_args.wayland = True
mock_parse.return_value = mock_args
# Call main function
main()
# Verify components were initialized correctly
mock_speech.assert_called_once_with(
engine="vosk",
model_size="medium",
language="en-us",
vad_sensitivity=3,
silence_timeout=2.0,
stop_sound_guard_ms=200,
buffer_during_reload=True,
voice_commands_enabled=None,
audio_device_index=None,
audio_device_name=None,
whispercpp_no_timestamps=True,
whispercpp_no_context=True,
whispercpp_initial_prompt="",
whispercpp_temperature=0.0,
whispercpp_temperature_inc=-1.0,
whispercpp_entropy_thold=2.4,
whispercpp_logprob_thold=-1.0,
whispercpp_no_speech_thold=0.6,
whispercpp_n_threads=0,
whispercpp_gpu_device=None,
remote_api_url="",
remote_api_key="",
remote_api_endpoint="/inference",
remote_api_model="whisper-1",
)
mock_text.assert_called_once_with(wayland_mode=True)
mock_action_handler.assert_called_once_with(mock_text_instance)
mock_tray.assert_called_once_with(
speech_engine=mock_speech_instance,
text_injector=mock_text_instance,
transcription_history=ANY,
)
# Verify callbacks were registered
mock_speech_instance.register_text_callback.assert_called_once()
mock_speech_instance.register_segment_callback.assert_called_once()
mock_speech_instance.register_action_callback.assert_called_once_with(
mock_action_instance.handle_action
)
mock_speech_instance.register_state_callback.assert_called_once()
# Verify the tray indicator was started
mock_tray_instance.run.assert_called_once()
@patch("vocalinux.main.check_dependencies")
@patch("vocalinux.speech_recognition.recognition_manager.SpeechRecognitionManager")
@patch("vocalinux.text_injection.text_injector.TextInjector")
@patch("vocalinux.ui.tray_indicator.TrayIndicator")
@patch("vocalinux.ui.config_manager.get_shared_config_manager")
@patch("vocalinux.ui.logging_manager.initialize_logging")
def test_registered_callbacks_keep_spacing_across_processing_to_listening(
self,
mock_init_logging,
mock_config_manager,
mock_tray,
mock_text,
mock_speech,
mock_check_deps,
):
"""Test main callback wiring preserves spacing between in-session segments."""
mock_check_deps.return_value = True
mock_config_instance = MagicMock()
mock_config_instance.get_settings.return_value = {
"speech_recognition": {},
"general": {"first_run": False},
}
mock_config_manager.return_value = mock_config_instance
mock_speech_instance = MagicMock()
mock_text_instance = MagicMock()
mock_text_instance.inject_text.return_value = True
mock_tray_instance = MagicMock()
mock_speech.return_value = mock_speech_instance
mock_text.return_value = mock_text_instance
mock_tray.return_value = mock_tray_instance
with patch("vocalinux.main.parse_arguments") as mock_parse:
mock_args = MagicMock()
mock_args.debug = False
mock_args.model = "medium"
mock_args.engine = "vosk"
mock_args.language = "en-us"
mock_args.wayland = False
mock_args.start_minimized = False
mock_parse.return_value = mock_args
main()
text_callback = mock_speech_instance.register_text_callback.call_args.args[0]
state_callback = mock_speech_instance.register_state_callback.call_args.args[0]
text_callback("Hello.")
state_callback(RecognitionState.PROCESSING)
state_callback(RecognitionState.LISTENING)
text_callback("World")
calls = [call.args[0] for call in mock_text_instance.inject_text.call_args_list]
self.assertEqual(calls, ["Hello. ", "World "])
state_callback(RecognitionState.IDLE)
mock_text_instance.inject_text.reset_mock()
text_callback("Next session")
# Trailing space persists in the previous field; next session starts clean
# (no leading space) but still gets its own trailing space.
mock_text_instance.inject_text.assert_called_once_with("Next session ")
def _run_main_and_get_text_callback(
self,
*,
engine: str,
auto_capitalize: bool,
mock_config_manager,
mock_tray,
mock_text,
mock_speech,
mock_check_deps,
):
"""Boot main() with mocked deps and return the registered text callback."""
mock_check_deps.return_value = True
mock_config_instance = MagicMock()
mock_config_instance.get_settings.return_value = {
"speech_recognition": {},
"general": {"first_run": False},
}
mock_config_instance.get.side_effect = lambda section, key, default=None: (
auto_capitalize if section == "text_injection" and key == "auto_capitalize" else default
)
mock_config_manager.return_value = mock_config_instance
mock_speech_instance = MagicMock()
mock_speech_instance.engine = engine
mock_text_instance = MagicMock()
mock_text_instance.inject_text.return_value = True
mock_tray_instance = MagicMock()
mock_speech.return_value = mock_speech_instance
mock_text.return_value = mock_text_instance
mock_tray.return_value = mock_tray_instance
with patch("vocalinux.main.parse_arguments") as mock_parse:
mock_args = MagicMock()
mock_args.debug = False
mock_args.model = "medium"
mock_args.engine = engine
mock_args.language = "en-us"
mock_args.wayland = False
mock_args.start_minimized = False
mock_parse.return_value = mock_args
with patch("sys.argv", ["vocalinux", "--engine", engine]):
main()
text_callback = mock_speech_instance.register_text_callback.call_args.args[0]
return text_callback, mock_text_instance
@patch("vocalinux.main.check_dependencies")
@patch("vocalinux.speech_recognition.recognition_manager.SpeechRecognitionManager")
@patch("vocalinux.text_injection.text_injector.TextInjector")
@patch("vocalinux.ui.tray_indicator.TrayIndicator")
@patch("vocalinux.ui.config_manager.get_shared_config_manager")
@patch("vocalinux.ui.logging_manager.initialize_logging")
def test_vosk_auto_capitalize_applies_to_injected_text(
self,
mock_init_logging,
mock_config_manager,
mock_tray,
mock_text,
mock_speech,
mock_check_deps,
):
"""Vosk engine capitalizes sentences when auto_capitalize is enabled."""
text_callback, mock_text_instance = self._run_main_and_get_text_callback(
engine="vosk",
auto_capitalize=True,
mock_config_manager=mock_config_manager,
mock_tray=mock_tray,
mock_text=mock_text,
mock_speech=mock_speech,
mock_check_deps=mock_check_deps,
)
text_callback("hello world. goodbye")
mock_text_instance.inject_text.assert_called_once_with("Hello world. Goodbye ")
@patch("vocalinux.main.check_dependencies")
@patch("vocalinux.speech_recognition.recognition_manager.SpeechRecognitionManager")
@patch("vocalinux.text_injection.text_injector.TextInjector")
@patch("vocalinux.ui.tray_indicator.TrayIndicator")
@patch("vocalinux.ui.config_manager.get_shared_config_manager")
@patch("vocalinux.ui.logging_manager.initialize_logging")
def test_vosk_auto_capitalize_can_be_disabled(
self,
mock_init_logging,
mock_config_manager,
mock_tray,
mock_text,
mock_speech,
mock_check_deps,
):
"""Vosk leaves casing unchanged when auto_capitalize is disabled."""
text_callback, mock_text_instance = self._run_main_and_get_text_callback(
engine="vosk",
auto_capitalize=False,
mock_config_manager=mock_config_manager,
mock_tray=mock_tray,
mock_text=mock_text,
mock_speech=mock_speech,
mock_check_deps=mock_check_deps,
)
text_callback("hello world. goodbye")
mock_text_instance.inject_text.assert_called_once_with("hello world. goodbye ")
@patch("vocalinux.main.check_dependencies")
@patch("vocalinux.speech_recognition.recognition_manager.SpeechRecognitionManager")
@patch("vocalinux.text_injection.text_injector.TextInjector")
@patch("vocalinux.ui.tray_indicator.TrayIndicator")
@patch("vocalinux.ui.config_manager.get_shared_config_manager")
@patch("vocalinux.ui.logging_manager.initialize_logging")
def test_whisper_skips_auto_capitalize(
self,
mock_init_logging,
mock_config_manager,
mock_tray,
mock_text,
mock_speech,
mock_check_deps,
):
"""Whisper engines keep model casing even when auto_capitalize is on."""
text_callback, mock_text_instance = self._run_main_and_get_text_callback(
engine="whisper_cpp",
auto_capitalize=True,
mock_config_manager=mock_config_manager,
mock_tray=mock_tray,
mock_text=mock_text,
mock_speech=mock_speech,
mock_check_deps=mock_check_deps,
)
text_callback("hello world. goodbye")
mock_text_instance.inject_text.assert_called_once_with("hello world. goodbye ")
@patch("vocalinux.main.check_dependencies")
@patch("vocalinux.ui.action_handler.ActionHandler")
@patch("vocalinux.speech_recognition.recognition_manager.SpeechRecognitionManager")
@patch("vocalinux.text_injection.text_injector.TextInjector")
@patch("vocalinux.ui.tray_indicator.TrayIndicator")
@patch("vocalinux.main.logging")
@patch("vocalinux.ui.config_manager.get_shared_config_manager")
@patch("vocalinux.ui.logging_manager.initialize_logging")
def test_main_with_debug_enabled(
self,
mock_init_logging,
mock_config_manager,
mock_logging,
mock_tray,
mock_text,
mock_speech,
mock_action_handler,
mock_check_deps,
):
"""Test that debug mode enables debug logging."""
import logging # Import for DEBUG constant
mock_check_deps.return_value = True
# Mock ConfigManager
mock_config_instance = MagicMock()
mock_config_instance.get_settings.return_value = {
"speech_recognition": {},
"general": {"first_run": False},
}
mock_config_manager.return_value = mock_config_instance
# Mock objects
mock_speech_instance = MagicMock()
mock_text_instance = MagicMock()
mock_tray_instance = MagicMock()
mock_action_instance = MagicMock()
mock_speech.return_value = mock_speech_instance
mock_text.return_value = mock_text_instance
mock_tray.return_value = mock_tray_instance
mock_action_handler.return_value = mock_action_instance
with patch("vocalinux.main.parse_arguments") as mock_parse:
# Create mock args with debug enabled
mock_args = MagicMock()
mock_args.debug = True
mock_args.model = "small"
mock_args.engine = "vosk"
mock_args.language = "en-us"
mock_args.wayland = False
mock_parse.return_value = mock_args
# Create mock loggers
root_logger = MagicMock()
mock_logging.getLogger.return_value = root_logger
# Call main
main()
# Verify root logger had setLevel called with DEBUG
root_logger.setLevel.assert_called()
@patch("vocalinux.main.check_dependencies")
@patch("vocalinux.ui.action_handler.ActionHandler")
@patch("vocalinux.speech_recognition.recognition_manager.SpeechRecognitionManager")
@patch("vocalinux.text_injection.text_injector.TextInjector")
@patch("vocalinux.ui.tray_indicator.TrayIndicator")
@patch("vocalinux.ui.first_run_dialog.show_first_run_dialog")
@patch("vocalinux.ui.config_manager.get_shared_config_manager")
@patch("vocalinux.ui.logging_manager.initialize_logging")
def test_main_first_run_later_keeps_prompt_enabled(
self,
mock_init_logging,
mock_config_manager,
mock_first_run_dialog,
mock_tray,
mock_text,
mock_speech,
mock_action_handler,
mock_check_deps,
):
"""Selecting later on first-run does not disable future prompt."""
mock_check_deps.return_value = True
mock_first_run_dialog.return_value = "later"
mock_config_instance = MagicMock()
mock_config_instance.get_settings.return_value = {
"speech_recognition": {},
"general": {"first_run": True},
}
mock_config_manager.return_value = mock_config_instance
mock_speech.return_value = MagicMock()
mock_text.return_value = MagicMock()
mock_tray.return_value = MagicMock()
mock_action_instance = MagicMock()
mock_action_handler.return_value = mock_action_instance
with patch("vocalinux.main.parse_arguments") as mock_parse:
mock_args = MagicMock()
mock_args.debug = False
mock_args.model = "small"
mock_args.engine = "vosk"
mock_args.language = "en-us"
mock_args.wayland = False
mock_parse.return_value = mock_args
with patch("vocalinux.main.logger"):
main()
self.assertFalse(
any(
call.args == ("general", "first_run", False)
for call in mock_config_instance.set.call_args_list
)
)
mock_config_instance.save_settings.assert_not_called()
@patch("vocalinux.main.check_dependencies")
@patch("vocalinux.ui.action_handler.ActionHandler")
@patch("vocalinux.speech_recognition.recognition_manager.SpeechRecognitionManager")
@patch("vocalinux.text_injection.text_injector.TextInjector")
@patch("vocalinux.ui.tray_indicator.TrayIndicator")
@patch("vocalinux.ui.first_run_dialog.show_first_run_dialog")
@patch("vocalinux.ui.config_manager.get_shared_config_manager")
@patch("vocalinux.ui.logging_manager.initialize_logging")
def test_main_start_minimized_skips_first_run_prompt(
self,
mock_init_logging,
mock_config_manager,
mock_first_run_dialog,
mock_tray,
mock_text,
mock_speech,
mock_action_handler,
mock_check_deps,
):
mock_check_deps.return_value = True
mock_config_instance = MagicMock()
mock_config_instance.get_settings.return_value = {
"speech_recognition": {},
"general": {"first_run": True},
}
mock_config_manager.return_value = mock_config_instance
mock_speech.return_value = MagicMock()
mock_text.return_value = MagicMock()
mock_tray.return_value = MagicMock()
mock_action_handler.return_value = MagicMock()
with patch("vocalinux.main.parse_arguments") as mock_parse:
mock_args = MagicMock()
mock_args.debug = False
mock_args.model = "small"
mock_args.engine = "vosk"
mock_args.language = "en-us"
mock_args.wayland = False
mock_args.start_minimized = True
mock_parse.return_value = mock_args
with patch("vocalinux.main.logger"):
main()
mock_first_run_dialog.assert_not_called()
self.assertFalse(
any(
call.args[0:2] == ("general", "first_run")
for call in mock_config_instance.set.call_args_list
)
)
class TestCheckDependencies(unittest.TestCase):
"""Test cases for check_dependencies function."""
def test_check_dependencies_all_available(self):
"""Test when all dependencies are available."""
# Mock all the imports that check_dependencies does
mock_gi = MagicMock()
mock_gi.require_version = MagicMock()
mock_gtk = MagicMock()
mock_appindicator = MagicMock()
mock_pynput = MagicMock()
mock_requests = MagicMock()
with patch.dict(
"sys.modules",
{
"gi": mock_gi,
"gi.repository": MagicMock(Gtk=mock_gtk, AppIndicator3=mock_appindicator),
"pynput": mock_pynput,
"requests": mock_requests,
},
):
result = check_dependencies()
self.assertTrue(result)
def test_check_dependencies_does_not_require_pynput(self):
"""Test startup is allowed when the optional pynput backend is unavailable."""
mock_gi = MagicMock()
mock_gi.require_version = MagicMock()
mock_gtk = MagicMock()
mock_appindicator = MagicMock()
mock_requests = MagicMock()
with patch.dict(
"sys.modules",
{
"gi": mock_gi,
"gi.repository": MagicMock(Gtk=mock_gtk, AppIndicator3=mock_appindicator),
"requests": mock_requests,
},
):
result = check_dependencies()
self.assertTrue(result)
def test_check_dependencies_missing_gtk(self):
"""Test when GTK is missing."""
# Make gi.require_version raise ValueError for Gtk
def require_version_side_effect(name, version):
if name == "Gtk":
raise ValueError("Gtk not found")
mock_gi = MagicMock()
mock_gi.require_version = MagicMock(side_effect=require_version_side_effect)
mock_pynput = MagicMock()
mock_requests = MagicMock()
with patch.dict(
"sys.modules",
{
"gi": mock_gi,
"pynput": mock_pynput,
"requests": mock_requests,
},
):
with patch("vocalinux.main.logger"):
result = check_dependencies()
self.assertFalse(result)
def test_check_dependencies_missing_appindicator_with_ayatana_fallback(self):
"""Test when legacy AppIndicator3 is missing but Ayatana is available."""
# Prefer Ayatana; legacy AppIndicator3 is only a fallback.
def require_version_side_effect(name, version):
if name == "AppIndicator3":
raise ValueError("AppIndicator3 not found")
# AyatanaAppIndicator3 works fine
mock_gi = MagicMock()
mock_gi.require_version = MagicMock(side_effect=require_version_side_effect)
mock_gtk = MagicMock()
mock_ayatana = MagicMock()
mock_pynput = MagicMock()
mock_requests = MagicMock()
with patch.dict(
"sys.modules",
{
"gi": mock_gi,
"gi.repository": MagicMock(Gtk=mock_gtk, AyatanaAppIndicator3=mock_ayatana),
"pynput": mock_pynput,
"requests": mock_requests,
},
):
with patch("vocalinux.main.logger"):
result = check_dependencies()
# Should return True because AyatanaAppIndicator3 works
self.assertTrue(result)
def test_check_dependencies_falls_back_to_legacy_appindicator(self):
"""Test when Ayatana is missing but legacy AppIndicator3 is available."""
def require_version_side_effect(name, version):
if name in ("AyatanaAppIndicator3", "AyatanaAppindicator3"):
raise ValueError(f"{name} not found")
mock_gi = MagicMock()
mock_gi.require_version = MagicMock(side_effect=require_version_side_effect)
mock_gtk = MagicMock()
mock_appindicator = MagicMock()
mock_pynput = MagicMock()
mock_requests = MagicMock()
with patch.dict(
"sys.modules",
{
"gi": mock_gi,
"gi.repository": MagicMock(Gtk=mock_gtk, AppIndicator3=mock_appindicator),
"pynput": mock_pynput,
"requests": mock_requests,
},
):
with patch("vocalinux.main.logger"):
result = check_dependencies()
self.assertTrue(result)
def test_check_dependencies_falls_back_to_lowercase_ayatana(self):
"""Test rare lowercase AyatanaAppindicator3 typelib is accepted."""
def require_version_side_effect(name, version):
if name == "AyatanaAppIndicator3":
raise ValueError("AyatanaAppIndicator3 not found")
mock_gi = MagicMock()
mock_gi.require_version = MagicMock(side_effect=require_version_side_effect)
mock_gtk = MagicMock()
mock_ayatana_lower = MagicMock()
mock_pynput = MagicMock()
mock_requests = MagicMock()
with patch.dict(
"sys.modules",
{
"gi": mock_gi,
"gi.repository": MagicMock(Gtk=mock_gtk, AyatanaAppindicator3=mock_ayatana_lower),
"pynput": mock_pynput,
"requests": mock_requests,
},
):
with patch("vocalinux.main.logger"):
result = check_dependencies()
self.assertTrue(result)
def test_check_dependencies_missing_both_appindicators(self):
"""Test when both AppIndicator3 and AyatanaAppIndicator3 are missing."""
# Make gi.require_version raise ValueError for all AppIndicator variants
def require_version_side_effect(name, version):
if name in (
"AppIndicator3",
"AyatanaAppIndicator3",
"AyatanaAppindicator3",
):
raise ValueError(f"{name} not found")
mock_gi = MagicMock()
mock_gi.require_version = MagicMock(side_effect=require_version_side_effect)
mock_gtk = MagicMock()
mock_pynput = MagicMock()
mock_requests = MagicMock()
with patch.dict(
"sys.modules",
{
"gi": mock_gi,
"gi.repository": MagicMock(Gtk=mock_gtk),
"pynput": mock_pynput,
"requests": mock_requests,
},
):
with patch("vocalinux.main.logger"):
result = check_dependencies()
# Should return False because both indicators are missing
self.assertFalse(result)
class TestMainConfigPrecedence(unittest.TestCase):
"""Test cases for configuration precedence in main."""
@patch("vocalinux.main.check_dependencies")
@patch("vocalinux.ui.action_handler.ActionHandler")
@patch("vocalinux.speech_recognition.recognition_manager.SpeechRecognitionManager")
@patch("vocalinux.text_injection.text_injector.TextInjector")
@patch("vocalinux.ui.tray_indicator.TrayIndicator")
@patch("vocalinux.ui.config_manager.get_shared_config_manager")
@patch("vocalinux.ui.logging_manager.initialize_logging")
def test_cli_args_override_config(
self,
mock_init_logging,
mock_config_manager,
mock_tray,
mock_text,
mock_speech,
mock_action_handler,
mock_check_deps,
):
"""Test that CLI arguments take precedence over saved config."""
mock_check_deps.return_value = True
# Mock ConfigManager to return saved settings
mock_config_instance = MagicMock()
mock_config_instance.get_settings.return_value = {
"speech_recognition": {
"engine": "vosk",
"model_size": "small",
"language": "en-us",
},
"general": {"first_run": False},
}
mock_config_manager.return_value = mock_config_instance
mock_speech_instance = MagicMock()
mock_text_instance = MagicMock()
mock_tray_instance = MagicMock()
mock_action_instance = MagicMock()
mock_speech.return_value = mock_speech_instance
mock_text.return_value = mock_text_instance
mock_tray.return_value = mock_tray_instance
mock_action_handler.return_value = mock_action_instance
# Simulate CLI args being set
with patch(
"sys.argv",
[
"vocalinux",
"--engine",
"whisper",
"--model",
"large",
"--language",
"fr",
],
):
with patch("vocalinux.main.logger"):
main()
# CLI args should override config
mock_speech.assert_called_once()
call_kwargs = mock_speech.call_args[1]
self.assertEqual(call_kwargs["engine"], "whisper")
self.assertEqual(call_kwargs["model_size"], "large")
self.assertEqual(call_kwargs["language"], "fr")
@patch("vocalinux.main.check_dependencies")
@patch("vocalinux.ui.action_handler.ActionHandler")
@patch("vocalinux.speech_recognition.recognition_manager.SpeechRecognitionManager")
@patch("vocalinux.text_injection.text_injector.TextInjector")
@patch("vocalinux.ui.tray_indicator.TrayIndicator")
@patch("vocalinux.ui.config_manager.get_shared_config_manager")
@patch("vocalinux.ui.logging_manager.initialize_logging")
def test_config_used_when_no_cli_args(
self,
mock_init_logging,
mock_config_manager,
mock_tray,
mock_text,
mock_speech,
mock_action_handler,
mock_check_deps,
):
"""Test that saved config is used when CLI args not provided."""
mock_check_deps.return_value = True
# Mock ConfigManager to return saved settings
mock_config_instance = MagicMock()
mock_config_instance.get_settings.return_value = {
"speech_recognition": {
"engine": "whisper",
"model_size": "medium",
"language": "de",
},
"audio": {
"device_index": 2,
},
"general": {"first_run": False},
}
mock_config_instance.get_model_size_for_engine.return_value = "medium"
mock_config_manager.return_value = mock_config_instance
mock_speech_instance = MagicMock()
mock_text_instance = MagicMock()
mock_tray_instance = MagicMock()
mock_action_instance = MagicMock()
mock_speech.return_value = mock_speech_instance
mock_text.return_value = mock_text_instance
mock_tray.return_value = mock_tray_instance
mock_action_handler.return_value = mock_action_instance
# No CLI args for engine/model/language
with patch("sys.argv", ["vocalinux"]):
with patch("vocalinux.main.logger"):
main()
# Config values should be used
mock_speech.assert_called_once()
call_kwargs = mock_speech.call_args[1]
self.assertEqual(call_kwargs["engine"], "whisper")
self.assertEqual(call_kwargs["model_size"], "medium")
self.assertEqual(call_kwargs["language"], "de")
self.assertEqual(call_kwargs["audio_device_index"], 2)
@patch("vocalinux.main.check_dependencies")
@patch("vocalinux.ui.action_handler.ActionHandler")
@patch("vocalinux.speech_recognition.recognition_manager.SpeechRecognitionManager")
@patch("vocalinux.text_injection.text_injector.TextInjector")
@patch("vocalinux.ui.tray_indicator.TrayIndicator")
@patch("vocalinux.ui.config_manager.get_shared_config_manager")
@patch("vocalinux.ui.logging_manager.initialize_logging")
def test_cli_language_normalized_for_parakeet(
self,
mock_init_logging,
mock_config_manager,
mock_tray,
mock_text,
mock_speech,
mock_action_handler,
mock_check_deps,
):
"""CLI --language must not be stored when --engine is parakeet."""
mock_check_deps.return_value = True
mock_config_instance = MagicMock()
mock_config_instance.get_settings.return_value = {
"speech_recognition": {
"engine": "whisper_cpp",
"model_size": "small",
"language": "en-us",
},
"general": {"first_run": False},
}
mock_config_manager.return_value = mock_config_instance
mock_speech.return_value = MagicMock()
mock_text.return_value = MagicMock()
mock_tray.return_value = MagicMock()
mock_action_handler.return_value = MagicMock()
with patch(
"sys.argv",
[
"vocalinux",
"--engine",
"parakeet",
"--model",
"v3-european",
"--language",
"fr",
],
):
with patch("vocalinux.main.logger"):
main()
mock_speech.assert_called_once()
call_kwargs = mock_speech.call_args[1]
self.assertEqual(call_kwargs["engine"], "parakeet")
self.assertEqual(call_kwargs["language"], "auto")
@patch("vocalinux.main.check_dependencies")
@patch("vocalinux.ui.action_handler.ActionHandler")
@patch("vocalinux.speech_recognition.recognition_manager.SpeechRecognitionManager")
@patch("vocalinux.text_injection.text_injector.TextInjector")
@patch("vocalinux.ui.tray_indicator.TrayIndicator")
@patch("vocalinux.ui.config_manager.get_shared_config_manager")
@patch("vocalinux.ui.logging_manager.initialize_logging")
def test_saved_language_normalized_for_parakeet(
self,
mock_init_logging,
mock_config_manager,
mock_tray,
mock_text,
mock_speech,
mock_action_handler,
mock_check_deps,
):
"""A leftover saved language must not be stored when the engine is parakeet."""
mock_check_deps.return_value = True
mock_config_instance = MagicMock()
mock_config_instance.get_settings.return_value = {
"speech_recognition": {
"engine": "parakeet",
"model_size": "v3-european",
"language": "de",
},
"general": {"first_run": False},
}
mock_config_instance.get_model_size_for_engine.return_value = "v3-european"
mock_config_manager.return_value = mock_config_instance
mock_speech.return_value = MagicMock()
mock_text.return_value = MagicMock()
mock_tray.return_value = MagicMock()
mock_action_handler.return_value = MagicMock()
with patch("sys.argv", ["vocalinux"]):
with patch("vocalinux.main.logger"):
main()
mock_speech.assert_called_once()
call_kwargs = mock_speech.call_args[1]
self.assertEqual(call_kwargs["engine"], "parakeet")
self.assertEqual(call_kwargs["language"], "auto")
@patch("vocalinux.main.check_dependencies")
@patch("vocalinux.ui.action_handler.ActionHandler")
@patch("vocalinux.speech_recognition.recognition_manager.SpeechRecognitionManager")
@patch("vocalinux.text_injection.text_injector.TextInjector")
@patch("vocalinux.ui.tray_indicator.TrayIndicator")
@patch("vocalinux.ui.config_manager.ConfigManager")
@patch("vocalinux.ui.logging_manager.initialize_logging")
def test_model_resolved_per_engine_not_from_generic_key(
self,
mock_init_logging,
mock_config_manager,
mock_tray,
mock_text,
mock_speech,
mock_action_handler,
mock_check_deps,
):
"""Startup must use the engine's own model, not the stale generic key.
The generic "model_size" key holds whichever engine was saved last, so
a config carrying another engine's model must not decide what the
configured engine loads.
"""
mock_check_deps.return_value = True
mock_config_instance = MagicMock()
mock_config_instance.get_settings.return_value = {
"speech_recognition": {
"engine": "whisper_cpp",
# Left behind by the last VOSK save.
"model_size": "medium",
"vosk_model_size": "medium",
"whisper_cpp_model_size": "small",
"language": "en-us",
},
"general": {"first_run": False},
}
mock_config_instance.get_model_size_for_engine.side_effect = lambda engine: {
"vosk": "medium",
"whisper_cpp": "small",
}[engine]
mock_config_manager.return_value = mock_config_instance
mock_speech.return_value = MagicMock()
mock_text.return_value = MagicMock()
mock_tray.return_value = MagicMock()
mock_action_handler.return_value = MagicMock()
with patch("sys.argv", ["vocalinux"]):
with patch("vocalinux.main.logger"):
main()
mock_config_instance.get_model_size_for_engine.assert_called_with("whisper_cpp")
call_kwargs = mock_speech.call_args[1]
self.assertEqual(call_kwargs["engine"], "whisper_cpp")
self.assertEqual(call_kwargs["model_size"], "small")
class TestTextCallbackSpacing(unittest.TestCase):
"""Test spacing logic in text_callback_wrapper."""
def _make_callback(self, append_trailing_space: bool = True):
"""Build text_callback_wrapper with mocked dependencies."""
from vocalinux.ui.action_handler import ActionHandler
text_system = MagicMock()
text_system.inject_text.return_value = True
action_handler = ActionHandler(text_system)
def text_callback_wrapper(text: str):
text_to_inject = text.lstrip().rstrip(" \t")
if not text_to_inject:
return
if append_trailing_space:
if not text_to_inject.endswith((" ", "\t", "\n")):
text_to_inject += " "
elif action_handler.last_injected_text and action_handler.last_injected_text.strip():
text_to_inject = " " + text_to_inject
success = text_system.inject_text(text_to_inject)
if success:
action_handler.set_last_injected_text(text_to_inject)
def on_state_change(state: RecognitionState):
if state == RecognitionState.IDLE:
action_handler.set_last_injected_text("")
return text_callback_wrapper, on_state_change, text_system, action_handler
def test_first_segment_has_trailing_space(self):
cb, _, text_system, _ = self._make_callback()
cb("Hello world")
text_system.inject_text.assert_called_once_with("Hello world ")
def test_subsequent_segment_has_trailing_space_not_leading(self):
cb, _, text_system, _ = self._make_callback()
cb("Hello")
cb("world")
calls = [c.args[0] for c in text_system.inject_text.call_args_list]
self.assertEqual(calls, ["Hello ", "world "])
def test_cross_session_keeps_trailing_space_without_leading_space(self):
cb, on_state_change, text_system, _ = self._make_callback()
cb("first session")
on_state_change(RecognitionState.IDLE)
text_system.inject_text.reset_mock()
cb("second session")
# No leading space (empty-field safe); trailing space still appended.
text_system.inject_text.assert_called_once_with("second session ")
def test_whitespace_only_input_is_skipped(self):
cb, _, text_system, _ = self._make_callback()
cb(" ")
text_system.inject_text.assert_not_called()
def test_input_with_leading_space_is_stripped(self):
cb, _, text_system, _ = self._make_callback()
cb(" Hello world")
text_system.inject_text.assert_called_once_with("Hello world ")
def test_multiple_segments_all_get_trailing_spaces(self):
cb, _, text_system, _ = self._make_callback()
cb("one")
cb("two")
cb("three")
calls = [c.args[0] for c in text_system.inject_text.call_args_list]
self.assertEqual(calls, ["one ", "two ", "three "])
def test_space_after_punctuation_segment(self):
cb, _, text_system, _ = self._make_callback()
cb("Hello.")
cb("World")
calls = [c.args[0] for c in text_system.inject_text.call_args_list]
self.assertEqual(calls, ["Hello. ", "World "])
def test_newline_segment_does_not_get_trailing_space(self):
cb, _, text_system, _ = self._make_callback()
cb("Hello.\n")
text_system.inject_text.assert_called_once_with("Hello.\n")
def test_processing_to_listening_keeps_segment_spacing(self):
cb, on_state_change, text_system, _ = self._make_callback()
cb("Hello.")
on_state_change(RecognitionState.PROCESSING)
on_state_change(RecognitionState.LISTENING)
cb("World")
calls = [c.args[0] for c in text_system.inject_text.call_args_list]
self.assertEqual(calls, ["Hello. ", "World "])
def test_legacy_mode_uses_leading_space_in_session(self):
cb, _, text_system, _ = self._make_callback(append_trailing_space=False)
cb("Hello.")
cb("World")
calls = [c.args[0] for c in text_system.inject_text.call_args_list]
self.assertEqual(calls, ["Hello.", " World"])
def test_legacy_mode_clears_leading_space_across_sessions(self):
cb, on_state_change, text_system, _ = self._make_callback(append_trailing_space=False)
cb("first session")
on_state_change(RecognitionState.IDLE)
text_system.inject_text.reset_mock()
cb("second session")
text_system.inject_text.assert_called_once_with("second session")
class TestShouldAppendTrailingSpace(unittest.TestCase):
"""Test disk-backed trailing-space setting reader."""
def test_reads_setting_from_disk_with_true_default(self):
import json
import os
import tempfile
from vocalinux.main import _should_append_trailing_space
with tempfile.TemporaryDirectory() as d:
with patch("vocalinux.utils.paths.config_dir", return_value=d):
# No config file -> default True
self.assertTrue(_should_append_trailing_space())
cfg = os.path.join(d, "config.json")
with open(cfg, "w") as f:
json.dump({"text_injection": {}}, f)
self.assertTrue(_should_append_trailing_space())
with open(cfg, "w") as f:
json.dump({"text_injection": {"append_trailing_space": False}}, f)
self.assertFalse(_should_append_trailing_space())
with open(cfg, "w") as f:
json.dump({"text_injection": {"append_trailing_space": True}}, f)
self.assertTrue(_should_append_trailing_space())
def test_returns_true_when_config_read_fails(self):
from vocalinux.main import _should_append_trailing_space
with patch("vocalinux.utils.paths.config_dir", side_effect=OSError("boom")):
self.assertTrue(_should_append_trailing_space())
class TestMainCallbackTrailingSpaceEdges(unittest.TestCase):
"""Exercise trailing-space edge paths through the real main() callback."""
def _boot_under_patches(self, *, append_trailing_space: bool = True, inject_ok: bool = True):
"""Return (exit_stack, text_cb, mock_text) with patches still active."""
from contextlib import ExitStack
stack = ExitStack()
stack.enter_context(patch("vocalinux.main.check_dependencies", return_value=True))
mock_config_cls = stack.enter_context(
patch("vocalinux.ui.config_manager.get_shared_config_manager")
)
mock_config = MagicMock()
mock_config.get_settings.return_value = {
"speech_recognition": {},
"general": {"first_run": False},
}
mock_config.get.return_value = False # auto_capitalize off
mock_config_cls.return_value = mock_config
mock_speech_cls = stack.enter_context(
patch("vocalinux.speech_recognition.recognition_manager.SpeechRecognitionManager")
)
mock_speech = MagicMock()
mock_speech.engine = "whisper_cpp"
mock_speech_cls.return_value = mock_speech
mock_text_cls = stack.enter_context(
patch("vocalinux.text_injection.text_injector.TextInjector")
)
mock_text = MagicMock()
mock_text.inject_text.return_value = inject_ok
mock_text_cls.return_value = mock_text
mock_tray_cls = stack.enter_context(patch("vocalinux.ui.tray_indicator.TrayIndicator"))
mock_tray_cls.return_value = MagicMock()
stack.enter_context(patch("vocalinux.ui.logging_manager.initialize_logging"))
stack.enter_context(
patch(
"vocalinux.main._should_append_trailing_space",
return_value=append_trailing_space,
)
)
mock_parse = stack.enter_context(patch("vocalinux.main.parse_arguments"))
stack.enter_context(patch("sys.argv", ["vocalinux"]))
mock_args = MagicMock()
mock_args.debug = False
mock_args.model = "tiny"
mock_args.engine = "whisper_cpp"
mock_args.language = "en-us"
mock_args.wayland = False
mock_args.start_minimized = False
mock_parse.return_value = mock_args
try:
main()
text_cb = mock_speech.register_text_callback.call_args.args[0]
except BaseException:
# The caller only closes the stack once it gets one back, so a
# failure here would leak these patches into every later test.
stack.close()
raise
return stack, text_cb, mock_text
def test_whitespace_only_is_skipped_through_main(self):
stack, text_cb, mock_text = self._boot_under_patches()
try:
text_cb(" \t ")
mock_text.inject_text.assert_not_called()
finally:
stack.close()
def test_newline_segment_skips_trailing_space_through_main(self):
stack, text_cb, mock_text = self._boot_under_patches()
try:
text_cb("Hello.\n")
mock_text.inject_text.assert_called_once_with("Hello.\n")
finally:
stack.close()
def test_legacy_mode_adds_leading_space_in_session(self):
stack, text_cb, mock_text = self._boot_under_patches(append_trailing_space=False)
try:
text_cb("Hello.")
text_cb("World")
calls = [c.args[0] for c in mock_text.inject_text.call_args_list]
self.assertEqual(calls, ["Hello.", " World"])
finally:
stack.close()
def test_failed_inject_does_not_remember_text(self):
stack, text_cb, mock_text = self._boot_under_patches(
append_trailing_space=False, inject_ok=False
)
try:
text_cb("Hello.")
mock_text.inject_text.reset_mock()
mock_text.inject_text.return_value = True
text_cb("World")
# Failure means last_injected stays empty; next segment has no leading space
mock_text.inject_text.assert_called_once_with("World")
finally:
stack.close()
class TestSessionHistoryRecording(unittest.TestCase):
"""Segments commit to transcription history as one snippet per session."""
def _boot(self, *, extra_settings: Optional[Dict[str, Any]] = None) -> Tuple[
ExitStack,
Callable[[str], None],
Callable[[str, float], None],
Callable[[RecognitionState], None],
TranscriptionHistory,
]:
"""Boot main() under mocks; return (stack, text_cb, segment_cb, state_cb, history).
The history is the real TranscriptionHistory instance handed to the
(mocked) TrayIndicator, so assertions observe actual recording.
segment_cb(text, started_at) delivers a recognized segment with the
monotonic time its audio capture began — history recording consumes
capture time so tests control pre-/post-clear placement precisely.
"""
stack = ExitStack()
stack.enter_context(patch("vocalinux.main.check_dependencies", return_value=True))
mock_config_cls = stack.enter_context(
patch("vocalinux.ui.config_manager.get_shared_config_manager")
)
mock_config = MagicMock()
settings = {
"speech_recognition": {},
"general": {"first_run": False},
}
if extra_settings:
settings.update(extra_settings)
mock_config.get_settings.return_value = settings
mock_config.get.return_value = False # auto_capitalize off
mock_config_cls.return_value = mock_config
mock_speech_cls = stack.enter_context(
patch("vocalinux.speech_recognition.recognition_manager.SpeechRecognitionManager")
)
mock_speech = MagicMock()
mock_speech.engine = "whisper_cpp"
mock_speech_cls.return_value = mock_speech
mock_text_cls = stack.enter_context(
patch("vocalinux.text_injection.text_injector.TextInjector")
)
mock_text = MagicMock()
mock_text.inject_text.return_value = True
mock_text_cls.return_value = mock_text
mock_tray_cls = stack.enter_context(patch("vocalinux.ui.tray_indicator.TrayIndicator"))
mock_tray_cls.return_value = MagicMock()
stack.enter_context(patch("vocalinux.ui.logging_manager.initialize_logging"))
mock_parse = stack.enter_context(patch("vocalinux.main.parse_arguments"))
stack.enter_context(patch("sys.argv", ["vocalinux"]))
mock_args = MagicMock()
mock_args.debug = False
mock_args.model = "tiny"
mock_args.engine = "whisper_cpp"
mock_args.language = "en-us"
mock_args.wayland = False
mock_args.start_minimized = False
mock_parse.return_value = mock_args
try:
main()
text_cb = mock_speech.register_text_callback.call_args.args[0]
segment_cb = mock_speech.register_segment_callback.call_args.args[0]
state_cb = mock_speech.register_state_callback.call_args.args[0]
history = mock_tray_cls.call_args.kwargs["transcription_history"]
except BaseException:
stack.close()
raise
return stack, text_cb, segment_cb, state_cb, history
def test_session_segments_commit_as_single_snippet(self) -> None:
stack, _, segment_cb, state_cb, history = self._boot()
try:
state_cb(RecognitionState.LISTENING)
segment_cb("Hello.", time.monotonic())
state_cb(RecognitionState.PROCESSING)
state_cb(RecognitionState.LISTENING)
segment_cb("World", time.monotonic())
state_cb(RecognitionState.IDLE)
self.assertEqual(history.get_all(), ["Hello. World"])
finally:
stack.close()
def test_successive_sessions_record_separate_snippets_newest_first(self) -> None:
stack, _, segment_cb, state_cb, history = self._boot()
try:
state_cb(RecognitionState.LISTENING)
segment_cb("first session", time.monotonic())
state_cb(RecognitionState.IDLE)
state_cb(RecognitionState.LISTENING)
segment_cb("second session", time.monotonic())
state_cb(RecognitionState.IDLE)
self.assertEqual(history.get_all(), ["second session", "first session"])
finally:
stack.close()
def test_late_segment_merges_into_its_own_sessions_snippet(self) -> None:
"""A worker that outlives the bounded stop wait lands in its snippet."""
stack, _, segment_cb, state_cb, history = self._boot()
try:
state_cb(RecognitionState.LISTENING)
segment_cb("hello", time.monotonic())
segment_cb("world", time.monotonic())
# IDLE fires while the worker still holds a final segment.
state_cb(RecognitionState.IDLE)
segment_cb("late tail", time.monotonic())
self.assertEqual(history.get_all(), ["hello world late tail"])
# The next session's snippet is not contaminated.
state_cb(RecognitionState.LISTENING)
segment_cb("next session", time.monotonic())
state_cb(RecognitionState.IDLE)
self.assertEqual(history.get_all(), ["next session", "hello world late tail"])
finally:
stack.close()
def test_late_segment_from_old_worker_during_next_session(self) -> None:
"""A leftover worker delivering on its own thread stays out of the open session."""
stack, _, segment_cb, state_cb, history = self._boot()
try:
state_cb(RecognitionState.LISTENING)
segment_cb("one", time.monotonic())
state_cb(RecognitionState.IDLE)
state_cb(RecognitionState.LISTENING)
segment_cb("two", time.monotonic())
# The previous session's worker finally delivers on its own thread.
t = threading.Thread(target=segment_cb, args=("trailing", time.monotonic()))
t.start()
t.join()
self.assertEqual(history.get_all(), ["one trailing"])
state_cb(RecognitionState.IDLE)
self.assertEqual(history.get_all(), ["two", "one trailing"])
finally:
stack.close()
def test_session_without_segments_creates_no_snippet(self) -> None:
stack, _, segment_cb, state_cb, history = self._boot()
try:
state_cb(RecognitionState.LISTENING)
state_cb(RecognitionState.IDLE)
self.assertEqual(history.get_all(), [])
# A segment arriving now belongs to that empty session: it becomes
# its own snippet rather than merging into an older entry.
segment_cb("orphan", time.monotonic())
self.assertEqual(history.get_all(), ["orphan"])
finally:
stack.close()
def test_error_state_also_commits_partial_snippet(self) -> None:
stack, _, segment_cb, state_cb, history = self._boot()
try:
state_cb(RecognitionState.LISTENING)
segment_cb("partial", time.monotonic())
state_cb(RecognitionState.ERROR)
self.assertEqual(history.get_all(), ["partial"])
finally:
stack.close()
def test_history_disabled_records_nothing(self) -> None:
stack, _, segment_cb, state_cb, history = self._boot(
extra_settings={"history": {"enabled": False}}
)
try:
state_cb(RecognitionState.LISTENING)
segment_cb("hello", time.monotonic())
state_cb(RecognitionState.IDLE)
self.assertEqual(history.get_all(), [])
finally:
stack.close()
def test_invalid_history_max_items_uses_default(self) -> None:
"""A corrupted saved limit must not prevent startup."""
stack, _, _, _, history = self._boot(extra_settings={"history": {"max_items": "abc"}})
try:
self.assertEqual(history.max_items, 10)
finally:
stack.close()
def test_late_segment_after_clear_does_not_reappear(self) -> None:
"""A straggler from an ended session must not resurrect cleared history."""
stack, _, segment_cb, state_cb, history = self._boot()
try:
state_cb(RecognitionState.LISTENING)
segment_cb("before clear", time.monotonic())
state_cb(RecognitionState.IDLE)
self.assertEqual(history.get_all(), ["before clear"])
history.clear()
# Its audio was captured before the clear; the decode only
# finished afterwards, so it must still be refused.
segment_cb("late tail", history.cleared_at - 1.0)
self.assertEqual(history.get_all(), [])
finally:
stack.close()
def test_late_orphan_segment_after_clear_does_not_reappear(self) -> None:
"""Late-only output of an empty session is refused after a clear too."""
stack, _, segment_cb, state_cb, history = self._boot()
try:
state_cb(RecognitionState.LISTENING)
state_cb(RecognitionState.IDLE) # empty session: no snippet
history.clear()
segment_cb("orphan", history.cleared_at - 1.0)
self.assertEqual(history.get_all(), [])
finally:
stack.close()
def test_clear_during_session_drops_its_snippet(self) -> None:
"""A clear issued while a session runs keeps its whole snippet out."""
stack, _, segment_cb, state_cb, history = self._boot()
try:
state_cb(RecognitionState.LISTENING)
segment_cb("dictated", time.monotonic())
history.clear()
state_cb(RecognitionState.IDLE)
self.assertEqual(history.get_all(), [])
finally:
stack.close()
def test_dictation_after_mid_session_clear_is_kept(self) -> None:
"""A mid-session clear splits segments by capture time, not arrival.
Speech captured before the clear is refused no matter when its
decode lands; speech captured after it joins the session's snippet.
"""
stack, _, segment_cb, state_cb, history = self._boot()
try:
state_cb(RecognitionState.LISTENING)
segment_cb("before", time.monotonic())
history.clear()
boundary = history.cleared_at
# A decode of pre-clear audio finishing after the clear is still
# refused — this is the segment the epoch check used to keep or
# drop wholesale.
segment_cb("pre-clear tail", boundary - 1.0)
segment_cb("after clear", time.monotonic())
state_cb(RecognitionState.IDLE)
self.assertEqual(history.get_all(), ["after clear"])
finally:
stack.close()
if __name__ == "__main__":
unittest.main()