0ddb127437
* fix(commands): keep line breaks from standalone voice commands
Two stacked defects ate the newline "new line" / "new paragraph"
produce when spoken alone:
* _normalize_segment_text stripped all leading whitespace, so the
command's leading newline never reached the injector (a clean
"new line" segment was dropped entirely).
* whisper.cpp transcripts arrive punctuated ("New line."), which
missed the exact-match ladder and left a stray "." after the
substituted break, injected as ". ".
Normalize now strips only spaces/tabs, and the generic path
substitutes on a punctuation-stripped core when the whole segment
is command phrases, so engine edge punctuation cannot leak while
real text keeps it.
Fixes #955
* fix(injection): deliver newline-only segments to the focused app
inject_text treated whitespace-only text as empty and returned early,
so the "\n" a standalone "new line" now correctly produces was
discarded before any backend ran ("Empty text provided, skipping
injection"). Only a truly empty segment is skipped; whitespace-only
input is typed/pasted verbatim (xdotool types "\n" as Return).
Space/tab-only segments never reach this point: segment normalization
strips them upstream.
Fixes #955
* fix(injection): keep skipping space/tab-only segments
The previous commit let every whitespace-only string through, which
also delivered bare spaces and tabs a caller never wants typed.
Narrow the skip to text with no content outside space/tab so "\n"
still reaches the backend while " " stays a no-op (the pre-existing
test_empty_text_injection contract).
Fixes #955
* fix(injection): deliver newline segments over IBus too
IBusTextInjector.inject_text had the same whitespace guard and dropped
"\n" before it reached the engine socket; narrow it the same way so
a newline-only segment is sent while space/tab-only stays a no-op.
Also let capitalize_sentences uppercase the first letter after leading
whitespace, so Vosk + auto_capitalize keeps working on "\n hello".
Covers Greptile's findings: real-socket test for the IBus path, the
capitalize regression via the main() callback seam, and type hints on
the new injector tests.
Fixes #955
* test(injection): add type hints to new IBus socket test
AGENTS.md requires hints on every signature; the Greptile review
flagged the new test for missing them.
---------
Co-authored-by: Jatin K Malik <jatinkrmalik@gmail.com>
2452 lines
99 KiB
Python
2452 lines
99 KiB
Python
"""
|
|
Tests for the main module functionality.
|
|
"""
|
|
|
|
import argparse
|
|
import queue
|
|
import sys
|
|
import threading
|
|
import time
|
|
import unittest
|
|
from contextlib import ExitStack
|
|
from types import SimpleNamespace
|
|
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.assertIsNone(args.dictionary_file)
|
|
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.dictation_pad.DictationPad")
|
|
@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_dictation_pad,
|
|
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_instance.get_str.return_value = "" # no post-processing script
|
|
mock_config_instance.is_dictate_to_pad_enabled.return_value = 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_pad_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_dictation_pad.return_value = mock_pad_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_language_candidates="",
|
|
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,
|
|
dictionary_manager=ANY,
|
|
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,
|
|
on_quit=ANY,
|
|
dictation_pad=mock_pad_instance,
|
|
)
|
|
# The pad starts from the shared manager's live capture flag.
|
|
mock_dictation_pad.assert_called_once_with(
|
|
enabled=False, config_manager=mock_config_instance
|
|
)
|
|
|
|
# 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()
|
|
# The registered action callback queues the action onto the
|
|
# post-processing worker, which dispatches to the action handler.
|
|
action_cb = mock_speech_instance.register_action_callback.call_args.args[0]
|
|
action_cb("select_all").result(timeout=10)
|
|
mock_action_instance.handle_action.assert_called_once_with("select_all")
|
|
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_instance.get_str.return_value = "" # no post-processing script
|
|
mock_config_instance.is_dictate_to_pad_enabled.return_value = 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]
|
|
|
|
# Injection now happens on the post-processing worker; the returned
|
|
# future drains it synchronously.
|
|
text_callback("Hello.").result(timeout=10)
|
|
state_callback(RecognitionState.PROCESSING)
|
|
state_callback(RecognitionState.LISTENING)
|
|
text_callback("World").result(timeout=10)
|
|
|
|
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").result(timeout=10)
|
|
# 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_instance.get_str.return_value = "" # no post-processing script
|
|
mock_config_instance.is_dictate_to_pad_enabled.return_value = False
|
|
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").result(timeout=10)
|
|
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").result(timeout=10)
|
|
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").result(timeout=10)
|
|
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())
|
|
|
|
|
|
def _boot_main_callbacks(
|
|
*,
|
|
append_trailing_space: bool = True,
|
|
inject_ok: bool = True,
|
|
dictate_to_pad: bool = False,
|
|
post_script: str = "",
|
|
) -> SimpleNamespace:
|
|
"""Boot main() with mocked deps; return the registered callbacks and mocks.
|
|
|
|
The patch stack stays open — the caller owns ``stack`` and must close it
|
|
in a finally block so no patch leaks into later tests.
|
|
"""
|
|
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.get_str.return_value = post_script
|
|
mock_config.is_dictate_to_pad_enabled.return_value = dictate_to_pad
|
|
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_pad_cls = stack.enter_context(patch("vocalinux.ui.dictation_pad.DictationPad"))
|
|
mock_pad = mock_pad_cls.return_value
|
|
|
|
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,
|
|
)
|
|
)
|
|
# Probing the focused window shells out to compositor tools; keep the
|
|
# callback tests deterministic by making the probe report "unknown",
|
|
# which keeps the permissive injection path.
|
|
stack.enter_context(
|
|
patch(
|
|
"vocalinux.text_injection.focused_window.get_focused_window",
|
|
return_value=None,
|
|
)
|
|
)
|
|
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]
|
|
action_cb = mock_speech.register_action_callback.call_args.args[0]
|
|
state_cb = mock_speech.register_state_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 SimpleNamespace(
|
|
stack=stack,
|
|
text_cb=text_cb,
|
|
action_cb=action_cb,
|
|
state_cb=state_cb,
|
|
text_system=mock_text,
|
|
pad=mock_pad,
|
|
pad_cls=mock_pad_cls,
|
|
config=mock_config,
|
|
mock_text=mock_text,
|
|
mock_config=mock_config,
|
|
mock_speech=mock_speech,
|
|
mock_tray_cls=mock_tray_cls,
|
|
)
|
|
|
|
|
|
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,
|
|
post_script: str = "",
|
|
) -> SimpleNamespace:
|
|
"""Boot main() under mocks; the shared helper does the wiring."""
|
|
return _boot_main_callbacks(
|
|
append_trailing_space=append_trailing_space,
|
|
inject_ok=inject_ok,
|
|
post_script=post_script,
|
|
)
|
|
|
|
def test_whitespace_only_is_skipped_through_main(self) -> None:
|
|
"""Whitespace-only segments return without touching the injector."""
|
|
boot = self._boot_under_patches()
|
|
try:
|
|
self.assertIsNone(boot.text_cb(" \t "))
|
|
boot.mock_text.inject_text.assert_not_called()
|
|
finally:
|
|
boot.stack.close()
|
|
|
|
def test_newline_segment_skips_trailing_space_through_main(self) -> None:
|
|
boot = self._boot_under_patches()
|
|
try:
|
|
boot.text_cb("Hello.\n").result(timeout=10)
|
|
boot.text_system.inject_text.assert_called_once_with("Hello.\n")
|
|
finally:
|
|
boot.stack.close()
|
|
|
|
def test_legacy_mode_adds_leading_space_in_session(self) -> None:
|
|
boot = self._boot_under_patches(append_trailing_space=False)
|
|
try:
|
|
boot.text_cb("Hello.").result(timeout=10)
|
|
boot.text_cb("World").result(timeout=10)
|
|
calls = [c.args[0] for c in boot.text_system.inject_text.call_args_list]
|
|
self.assertEqual(calls, ["Hello.", " World"])
|
|
finally:
|
|
boot.stack.close()
|
|
|
|
def test_failed_inject_does_not_remember_text(self) -> None:
|
|
boot = self._boot_under_patches(append_trailing_space=False, inject_ok=False)
|
|
try:
|
|
boot.text_cb("Hello.").result(timeout=10)
|
|
boot.text_system.inject_text.reset_mock()
|
|
boot.text_system.inject_text.return_value = True
|
|
boot.text_cb("World").result(timeout=10)
|
|
# Failure means last_injected stays empty; next segment has no leading space
|
|
boot.text_system.inject_text.assert_called_once_with("World")
|
|
finally:
|
|
boot.stack.close()
|
|
|
|
def test_post_processing_transform_reaches_injector(self) -> None:
|
|
"""Configured script output is what gets injected, spacing rules included."""
|
|
boot = self._boot_under_patches(post_script="/fake/script.sh")
|
|
try:
|
|
run_result = MagicMock()
|
|
run_result.returncode = 0
|
|
run_result.stderr = ""
|
|
run_result.stdout = "TRANSFORMED"
|
|
with patch("vocalinux.post_processor.subprocess.run", return_value=run_result):
|
|
boot.text_cb("hello").result(timeout=10)
|
|
boot.text_system.inject_text.assert_called_once_with("TRANSFORMED ")
|
|
|
|
# A transformed paragraph break keeps its newlines (and so
|
|
# skips the appended trailing space like any "\n" ending).
|
|
boot.text_system.inject_text.reset_mock()
|
|
run_result.stdout = "PARA ONE.\n\n"
|
|
boot.text_cb("para one.\n\n").result(timeout=10)
|
|
boot.text_system.inject_text.assert_called_once_with("PARA ONE.\n\n")
|
|
finally:
|
|
boot.stack.close()
|
|
|
|
def test_post_processing_empty_output_skips_injection(self) -> None:
|
|
"""A script that emits nothing swallows the segment — nothing injected."""
|
|
boot = self._boot_under_patches(post_script="/fake/script.sh")
|
|
try:
|
|
run_result = MagicMock()
|
|
run_result.returncode = 0
|
|
run_result.stderr = ""
|
|
run_result.stdout = ""
|
|
with patch("vocalinux.post_processor.subprocess.run", return_value=run_result):
|
|
boot.text_cb("hello").result(timeout=10)
|
|
boot.text_system.inject_text.assert_not_called()
|
|
finally:
|
|
boot.stack.close()
|
|
|
|
def test_segment_queues_behind_running_script_when_script_cleared(self) -> None:
|
|
"""Clearing the script path mid-queue cannot let a later segment overtake."""
|
|
boot = self._boot_under_patches(post_script="/fake/script.sh")
|
|
started = threading.Event()
|
|
gate = threading.Event()
|
|
|
|
def blocked_run(cmd: list, **kwargs: object) -> MagicMock:
|
|
started.set()
|
|
gate.wait(timeout=10)
|
|
return MagicMock(returncode=0, stdout="PROCESSED FIRST", stderr="")
|
|
|
|
try:
|
|
with patch("vocalinux.post_processor.subprocess.run", side_effect=blocked_run):
|
|
first = boot.text_cb("first")
|
|
self.assertTrue(started.wait(timeout=5))
|
|
# User disables the script while the first segment still runs;
|
|
# the second must still queue behind it, not inject directly.
|
|
boot.mock_config.get_str.return_value = ""
|
|
second = boot.text_cb("second")
|
|
gate.set()
|
|
self.assertIsNotNone(first)
|
|
self.assertIsNotNone(second)
|
|
first.result(timeout=10)
|
|
second.result(timeout=10)
|
|
calls = [c.args[0] for c in boot.text_system.inject_text.call_args_list]
|
|
self.assertEqual(calls, ["PROCESSED FIRST ", "second "])
|
|
finally:
|
|
gate.set()
|
|
boot.stack.close()
|
|
|
|
def test_queued_segment_dropped_when_focus_moves(self) -> None:
|
|
"""A segment that outlived its target app is dropped, not injected."""
|
|
from vocalinux.text_injection.focused_window import FocusedWindow
|
|
|
|
boot = self._boot_under_patches(post_script="/fake/script.sh")
|
|
try:
|
|
window_a = FocusedWindow(app_id="editor", wm_class="Editor", process_name="editor")
|
|
window_b = FocusedWindow(app_id="browser", wm_class="Browser", process_name="browser")
|
|
run_result = MagicMock(returncode=0, stdout="OUT", stderr="")
|
|
# First probe (submit time) sees the editor; the worker's re-probe
|
|
# after the script finds the browser — different application.
|
|
with (
|
|
patch(
|
|
"vocalinux.text_injection.focused_window.get_focused_window",
|
|
side_effect=[window_a, window_b, window_b, window_b],
|
|
),
|
|
patch("vocalinux.post_processor.subprocess.run", return_value=run_result),
|
|
):
|
|
boot.text_cb("hello").result(timeout=10)
|
|
boot.text_system.inject_text.assert_not_called()
|
|
finally:
|
|
boot.stack.close()
|
|
|
|
def test_same_app_focus_change_still_injects(self) -> None:
|
|
"""Focus probe returning the same app keeps the segment deliverable."""
|
|
from vocalinux.text_injection.focused_window import FocusedWindow
|
|
|
|
boot = self._boot_under_patches(post_script="/fake/script.sh")
|
|
try:
|
|
window = FocusedWindow(app_id="editor", process_name="editor")
|
|
# Two editor windows differ only by title — same application.
|
|
other_doc = FocusedWindow(app_id="editor", title="other.py", process_name="editor")
|
|
run_result = MagicMock(returncode=0, stdout="hello", stderr="")
|
|
with (
|
|
patch(
|
|
"vocalinux.text_injection.focused_window.get_focused_window",
|
|
side_effect=[window, other_doc, other_doc],
|
|
),
|
|
patch("vocalinux.post_processor.subprocess.run", return_value=run_result),
|
|
):
|
|
boot.text_cb("hello").result(timeout=10)
|
|
boot.text_system.inject_text.assert_called_once_with("hello ")
|
|
finally:
|
|
boot.stack.close()
|
|
|
|
def test_voice_action_queues_behind_pending_text(self) -> None:
|
|
"""A voice action cannot overtake a segment still waiting in the worker."""
|
|
boot = self._boot_under_patches(post_script="/fake/script.sh")
|
|
started = threading.Event()
|
|
gate = threading.Event()
|
|
|
|
def blocked_run(cmd: list, **kwargs: object) -> MagicMock:
|
|
started.set()
|
|
gate.wait(timeout=10)
|
|
return MagicMock(returncode=0, stdout="PROCESSED", stderr="")
|
|
|
|
try:
|
|
with patch("vocalinux.post_processor.subprocess.run", side_effect=blocked_run):
|
|
text_future = boot.text_cb("hello")
|
|
self.assertTrue(started.wait(timeout=5))
|
|
# The action is issued while the script still holds the worker;
|
|
# it must run only after the queued text has been injected.
|
|
action_future = boot.action_cb("select_all")
|
|
gate.set()
|
|
text_future.result(timeout=10)
|
|
action_future.result(timeout=10)
|
|
call_names = [c[0] for c in boot.text_system.mock_calls]
|
|
self.assertEqual(call_names, ["inject_text", "_inject_keyboard_shortcut"])
|
|
finally:
|
|
gate.set()
|
|
boot.stack.close()
|
|
|
|
def test_voice_action_without_script_still_uses_worker(self) -> None:
|
|
"""With no script configured, actions still queue behind pending text."""
|
|
boot = self._boot_under_patches()
|
|
try:
|
|
action_future = boot.action_cb("select_all")
|
|
self.assertIsNotNone(action_future)
|
|
action_future.result(timeout=10)
|
|
boot.text_system._inject_keyboard_shortcut.assert_called_once_with("ctrl+a")
|
|
finally:
|
|
boot.stack.close()
|
|
|
|
def test_idle_reset_waits_for_queued_delete_action(self) -> None:
|
|
"""A 'delete that' queued before IDLE must still see the text it targets."""
|
|
boot = self._boot_under_patches(post_script="/fake/script.sh")
|
|
started = threading.Event()
|
|
gate = threading.Event()
|
|
|
|
def blocked_run(cmd: list, **kwargs: object) -> MagicMock:
|
|
started.set()
|
|
gate.wait(timeout=10)
|
|
return MagicMock(returncode=0, stdout="HELLO", stderr="")
|
|
|
|
try:
|
|
with patch("vocalinux.post_processor.subprocess.run", side_effect=blocked_run):
|
|
text_future = boot.text_cb("hello")
|
|
self.assertTrue(started.wait(timeout=5))
|
|
delete_future = boot.action_cb("delete_last")
|
|
# The session ends while both jobs are queued; the reset must
|
|
# run only after the delete action consumed last_injected_text.
|
|
boot.state_cb(RecognitionState.IDLE)
|
|
gate.set()
|
|
text_future.result(timeout=10)
|
|
delete_future.result(timeout=10)
|
|
boot.text_system.press_backspace.assert_called_once_with(len("HELLO "))
|
|
finally:
|
|
gate.set()
|
|
boot.stack.close()
|
|
|
|
def test_action_waits_for_slow_probe(self) -> None:
|
|
"""A stalled focus probe delays the action but never discards it."""
|
|
boot = self._boot_under_patches(post_script="/fake/script.sh")
|
|
stall = threading.Event()
|
|
|
|
def stalled_probe() -> None:
|
|
stall.wait(30)
|
|
return None
|
|
|
|
try:
|
|
with patch(
|
|
"vocalinux.text_injection.focused_window.get_focused_window",
|
|
side_effect=stalled_probe,
|
|
):
|
|
action_future = boot.action_cb("select_all")
|
|
self.assertIsNotNone(action_future)
|
|
# While the probe has not answered the job cannot run the
|
|
# shortcut; once it answers, the command must still fire.
|
|
self.assertFalse(action_future.done())
|
|
stall.set()
|
|
action_future.result(timeout=10)
|
|
boot.text_system._inject_keyboard_shortcut.assert_called_once_with("ctrl+a")
|
|
finally:
|
|
stall.set()
|
|
boot.stack.close()
|
|
|
|
def test_queued_action_dropped_when_focus_moves(self) -> None:
|
|
"""A queued shortcut is dropped, not fired into the newly focused app."""
|
|
from vocalinux.text_injection.focused_window import FocusedWindow
|
|
|
|
boot = self._boot_under_patches(post_script="/fake/script.sh")
|
|
started = threading.Event()
|
|
gate = threading.Event()
|
|
probes_done = threading.Event()
|
|
probe_calls: list = []
|
|
|
|
def blocked_run(cmd: list, **kwargs: object) -> MagicMock:
|
|
started.set()
|
|
gate.wait(timeout=10)
|
|
return MagicMock(returncode=0, stdout="OUT", stderr="")
|
|
|
|
try:
|
|
window_a = FocusedWindow(app_id="editor", wm_class="Editor", process_name="editor")
|
|
window_b = FocusedWindow(app_id="browser", wm_class="Browser", process_name="browser")
|
|
window_iter = iter([window_a, window_a, window_b, window_b, window_b, window_b])
|
|
|
|
def fake_probe() -> FocusedWindow:
|
|
# Both submit-time probes capture the editor; the jobs' later
|
|
# re-probes then see the browser.
|
|
probe_calls.append(1)
|
|
if len(probe_calls) == 2:
|
|
probes_done.set()
|
|
return next(window_iter)
|
|
|
|
with (
|
|
patch(
|
|
"vocalinux.text_injection.focused_window.get_focused_window",
|
|
side_effect=fake_probe,
|
|
),
|
|
patch("vocalinux.post_processor.subprocess.run", side_effect=blocked_run),
|
|
):
|
|
text_future = boot.text_cb("hello")
|
|
self.assertTrue(started.wait(timeout=5))
|
|
action_future = boot.action_cb("select_all")
|
|
# Wait for both submit-time probes so the re-probes in the
|
|
# queued jobs deterministically observe the new application.
|
|
self.assertTrue(probes_done.wait(timeout=5))
|
|
gate.set()
|
|
text_future.result(timeout=10)
|
|
action_future.result(timeout=10)
|
|
boot.text_system._inject_keyboard_shortcut.assert_not_called()
|
|
finally:
|
|
gate.set()
|
|
boot.stack.close()
|
|
|
|
def test_quit_hook_stops_worker_and_blocks_new_submissions(self) -> None:
|
|
"""The tray's on_quit hook drains the worker and rejects new segments."""
|
|
boot = self._boot_under_patches()
|
|
try:
|
|
on_quit = boot.mock_tray_cls.call_args.kwargs["on_quit"]
|
|
on_quit()
|
|
self.assertIsNone(boot.text_cb("hello"))
|
|
boot.text_system.inject_text.assert_not_called()
|
|
finally:
|
|
boot.stack.close()
|
|
|
|
def test_running_job_drops_result_during_quit(self) -> None:
|
|
"""A script mid-flight when quit begins cannot inject afterwards."""
|
|
boot = self._boot_under_patches(post_script="/fake/script.sh")
|
|
started = threading.Event()
|
|
gate = threading.Event()
|
|
|
|
def blocked_run(cmd: list, **kwargs: object) -> MagicMock:
|
|
started.set()
|
|
gate.wait(timeout=10)
|
|
return MagicMock(returncode=0, stdout="TOO LATE", stderr="")
|
|
|
|
try:
|
|
with patch("vocalinux.post_processor.subprocess.run", side_effect=blocked_run):
|
|
future = boot.text_cb("hello")
|
|
self.assertTrue(started.wait(timeout=5))
|
|
quit_thread = threading.Thread(
|
|
target=boot.mock_tray_cls.call_args.kwargs["on_quit"]
|
|
)
|
|
quit_thread.start()
|
|
gate.set()
|
|
quit_thread.join(timeout=10)
|
|
self.assertFalse(quit_thread.is_alive())
|
|
future.result(timeout=10)
|
|
boot.text_system.inject_text.assert_not_called()
|
|
finally:
|
|
gate.set()
|
|
boot.stack.close()
|
|
|
|
|
|
class TestMainCallbackLineBreakCommands(unittest.TestCase):
|
|
"""Standalone line-break segments through the real main() callback (#955).
|
|
|
|
"new line" / "new paragraph" expand to leading newlines, which the old
|
|
bare ``lstrip()`` destroyed — the segment was dropped or only its stray
|
|
punctuation survived. Normalization now strips only spaces/tabs.
|
|
"""
|
|
|
|
def test_standalone_newline_is_injected(self) -> None:
|
|
"""A segment that is only a line break reaches the injector verbatim."""
|
|
boot = _boot_main_callbacks()
|
|
try:
|
|
boot.text_cb("\n").result(timeout=10)
|
|
boot.text_system.inject_text.assert_called_once_with("\n")
|
|
finally:
|
|
boot.stack.close()
|
|
|
|
def test_standalone_paragraph_break_is_injected(self) -> None:
|
|
"""A paragraph break reaches the injector verbatim, no trailing space."""
|
|
boot = _boot_main_callbacks()
|
|
try:
|
|
boot.text_cb("\n\n").result(timeout=10)
|
|
boot.text_system.inject_text.assert_called_once_with("\n\n")
|
|
finally:
|
|
boot.stack.close()
|
|
|
|
def test_leading_newline_before_text_is_preserved(self) -> None:
|
|
"""A command starting a mixed segment keeps its line break."""
|
|
boot = _boot_main_callbacks()
|
|
try:
|
|
boot.text_cb("\n hello").result(timeout=10)
|
|
boot.text_system.inject_text.assert_called_once_with("\n hello ")
|
|
finally:
|
|
boot.stack.close()
|
|
|
|
def test_horizontal_whitespace_still_stripped(self) -> None:
|
|
"""Spaces and tabs are still stripped on both sides."""
|
|
boot = _boot_main_callbacks()
|
|
try:
|
|
boot.text_cb(" \t hello \t ").result(timeout=10)
|
|
boot.text_system.inject_text.assert_called_once_with("hello ")
|
|
finally:
|
|
boot.stack.close()
|
|
|
|
def test_vosk_capitalize_skips_leading_newline(self) -> None:
|
|
"""Vosk auto-capitalize uppercases the first letter after "\n"."""
|
|
boot = _boot_main_callbacks()
|
|
try:
|
|
boot.mock_speech.engine = "vosk"
|
|
boot.mock_config.get.side_effect = lambda section, key, *a, **k: (section, key) == (
|
|
"text_injection",
|
|
"auto_capitalize",
|
|
)
|
|
boot.text_cb("\nhello").result(timeout=10)
|
|
boot.text_system.inject_text.assert_called_once_with("\nHello ")
|
|
finally:
|
|
boot.stack.close()
|
|
|
|
|
|
class TestPadRoutingCallbacks(unittest.TestCase):
|
|
"""Pad capture routing through the real main() callback wrappers (#726).
|
|
|
|
The pad path is the feature's main routing: these exercise the callbacks
|
|
actually registered on the speech engine rather than re-implementations.
|
|
"""
|
|
|
|
def test_capture_routes_transcription_to_pad_not_injector(self) -> None:
|
|
boot = _boot_main_callbacks(dictate_to_pad=True)
|
|
try:
|
|
boot.text_cb("hello").result(timeout=10)
|
|
boot.pad.append_text.assert_called_once_with("hello ")
|
|
boot.text_system.inject_text.assert_not_called()
|
|
finally:
|
|
boot.stack.close()
|
|
|
|
def test_capture_routes_delete_that_to_pad(self) -> None:
|
|
boot = _boot_main_callbacks(dictate_to_pad=True)
|
|
try:
|
|
boot.pad.last_segment = "hello "
|
|
boot.text_cb("hello").result(timeout=10)
|
|
boot.action_cb("delete_last").result(timeout=10)
|
|
boot.pad.delete_last_chars.assert_called_once_with(len("hello "))
|
|
boot.text_system.press_backspace.assert_not_called()
|
|
finally:
|
|
boot.stack.close()
|
|
|
|
def test_capture_routes_editing_commands_to_pad(self) -> None:
|
|
"""undo/select/cut/copy/paste must never reach the focused app."""
|
|
boot = _boot_main_callbacks(dictate_to_pad=True)
|
|
try:
|
|
for action in (
|
|
"undo",
|
|
"redo",
|
|
"select_all",
|
|
"select_line",
|
|
"select_word",
|
|
"select_paragraph",
|
|
"cut",
|
|
"copy",
|
|
"paste",
|
|
):
|
|
boot.pad.handle_action.reset_mock()
|
|
self.assertTrue(boot.action_cb(action).result(timeout=10))
|
|
boot.pad.handle_action.assert_called_once_with(action)
|
|
boot.text_system._inject_keyboard_shortcut.assert_not_called()
|
|
boot.text_system.press_backspace.assert_not_called()
|
|
finally:
|
|
boot.stack.close()
|
|
|
|
def test_delete_that_follows_destination_across_mode_change(self) -> None:
|
|
"""Deletion targets where the last segment went, not the live toggle."""
|
|
boot = _boot_main_callbacks(dictate_to_pad=False)
|
|
try:
|
|
# Dictated into the app; capture switched on afterwards: the
|
|
# backspaces must still go to the app that received the text.
|
|
boot.text_cb("into app").result(timeout=10)
|
|
boot.config.is_dictate_to_pad_enabled.return_value = True
|
|
boot.action_cb("delete_last").result(timeout=10)
|
|
boot.text_system.press_backspace.assert_called_once_with(len("into app "))
|
|
boot.pad.delete_last_chars.assert_not_called()
|
|
|
|
# Dictated into the pad; capture switched off afterwards: the
|
|
# pad's copy is still the one removed.
|
|
boot.config.is_dictate_to_pad_enabled.return_value = True
|
|
boot.pad.last_segment = "into pad "
|
|
boot.pad.delete_last_chars.return_value = len("into pad ")
|
|
boot.text_cb("into pad").result(timeout=10)
|
|
boot.config.is_dictate_to_pad_enabled.return_value = False
|
|
boot.action_cb("delete_last").result(timeout=10)
|
|
boot.pad.delete_last_chars.assert_called_once_with(len("into pad "))
|
|
self.assertEqual(boot.text_system.press_backspace.call_count, 1)
|
|
finally:
|
|
boot.stack.close()
|
|
|
|
def test_capture_toggle_flips_routing_without_restart(self) -> None:
|
|
"""Routing follows the shared manager's live value, not a snapshot."""
|
|
boot = _boot_main_callbacks(dictate_to_pad=False)
|
|
try:
|
|
boot.text_cb("to app").result(timeout=10)
|
|
boot.text_system.inject_text.assert_called_once_with("to app ")
|
|
|
|
boot.config.is_dictate_to_pad_enabled.return_value = True
|
|
boot.text_cb("to pad").result(timeout=10)
|
|
boot.pad.append_text.assert_called_once_with("to pad ")
|
|
self.assertEqual(boot.text_system.inject_text.call_count, 1)
|
|
finally:
|
|
boot.stack.close()
|
|
|
|
def test_delete_that_with_empty_history_is_swallowed(self) -> None:
|
|
boot = _boot_main_callbacks(dictate_to_pad=True)
|
|
try:
|
|
self.assertTrue(boot.action_cb("delete_last").result(timeout=10))
|
|
boot.pad.delete_last_chars.assert_not_called()
|
|
boot.text_system.press_backspace.assert_not_called()
|
|
finally:
|
|
boot.stack.close()
|
|
|
|
def test_idle_state_resets_delete_that_target(self) -> None:
|
|
boot = _boot_main_callbacks(dictate_to_pad=True)
|
|
try:
|
|
boot.text_cb("pad text").result(timeout=10)
|
|
boot.state_cb(RecognitionState.IDLE)
|
|
boot.action_cb("delete_last").result(timeout=10)
|
|
boot.pad.delete_last_chars.assert_not_called()
|
|
boot.text_system.press_backspace.assert_not_called()
|
|
finally:
|
|
boot.stack.close()
|
|
|
|
def test_pad_undo_retargets_delete_that(self) -> None:
|
|
"""Undoing the last pad segment must not leave a stale delete length."""
|
|
boot = _boot_main_callbacks(dictate_to_pad=True)
|
|
try:
|
|
boot.text_cb("hello").result(timeout=10)
|
|
boot.text_cb("world").result(timeout=10)
|
|
# The pad reports the surviving segment after its own undo pops
|
|
# "world "; the next delete targets it, not the stale segment.
|
|
boot.pad.last_segment = "hello "
|
|
boot.pad.handle_action.return_value = True
|
|
self.assertTrue(boot.action_cb("undo").result(timeout=10))
|
|
boot.pad.handle_action.assert_called_once_with("undo")
|
|
|
|
boot.action_cb("delete_last").result(timeout=10)
|
|
boot.pad.delete_last_chars.assert_called_once_with(len("hello "))
|
|
boot.text_system.press_backspace.assert_not_called()
|
|
finally:
|
|
boot.stack.close()
|
|
|
|
def test_pad_undo_to_empty_clears_delete_target(self) -> None:
|
|
"""When pad undo removes the last segment, "delete that" is a no-op."""
|
|
boot = _boot_main_callbacks(dictate_to_pad=True)
|
|
try:
|
|
boot.text_cb("only").result(timeout=10)
|
|
boot.pad.last_segment = None
|
|
boot.pad.handle_action.return_value = True
|
|
self.assertTrue(boot.action_cb("undo").result(timeout=10))
|
|
|
|
self.assertTrue(boot.action_cb("delete_last").result(timeout=10))
|
|
boot.pad.delete_last_chars.assert_not_called()
|
|
boot.text_system.press_backspace.assert_not_called()
|
|
finally:
|
|
boot.stack.close()
|
|
|
|
def test_delete_that_after_manual_pad_edit_is_a_no_op(self) -> None:
|
|
"""A manual edit blurs segment boundaries: "delete that" must not
|
|
erase user-typed text with the stale recorded length."""
|
|
boot = _boot_main_callbacks(dictate_to_pad=True)
|
|
try:
|
|
boot.text_cb("hello").result(timeout=10)
|
|
# The widget's own edit cleared the tracked segment boundaries.
|
|
boot.pad.last_segment = None
|
|
self.assertTrue(boot.action_cb("delete_last").result(timeout=10))
|
|
boot.pad.delete_last_chars.assert_not_called()
|
|
boot.text_system.press_backspace.assert_not_called()
|
|
# Tracking stays cleared, so a repeat cannot retry the stale length.
|
|
self.assertTrue(boot.action_cb("delete_last").result(timeout=10))
|
|
boot.pad.delete_last_chars.assert_not_called()
|
|
finally:
|
|
boot.stack.close()
|
|
|
|
def test_pad_bound_segment_survives_focus_change(self) -> None:
|
|
"""Pad-bound text is not dropped when focus moved since the dictate."""
|
|
# A configured script arms the submit-time focus probe.
|
|
boot = _boot_main_callbacks(dictate_to_pad=True, post_script="/bin/cat")
|
|
try:
|
|
app_a = MagicMock()
|
|
app_a.identity_blob.return_value = "app-a"
|
|
app_b = MagicMock()
|
|
app_b.identity_blob.return_value = "app-b"
|
|
with patch(
|
|
"vocalinux.text_injection.focused_window.get_focused_window",
|
|
side_effect=[app_a, app_b],
|
|
):
|
|
boot.text_cb("hello").result(timeout=10)
|
|
boot.pad.append_text.assert_called_once_with("hello ")
|
|
boot.text_system.inject_text.assert_not_called()
|
|
finally:
|
|
boot.stack.close()
|
|
|
|
def test_segment_destination_decided_once_between_check_and_delivery(self) -> None:
|
|
"""A toggle flip between the focus check and delivery cannot reroute
|
|
a pad-bound segment into whichever application holds focus."""
|
|
boot = _boot_main_callbacks(dictate_to_pad=True, post_script="/bin/cat")
|
|
try:
|
|
# The routing decision reads the toggle once; a stale re-read at
|
|
# delivery would see it off and inject into the focused app.
|
|
boot.config.is_dictate_to_pad_enabled.side_effect = [True, False]
|
|
app_a = MagicMock()
|
|
app_a.identity_blob.return_value = "app-a"
|
|
with patch(
|
|
"vocalinux.text_injection.focused_window.get_focused_window",
|
|
return_value=app_a,
|
|
):
|
|
boot.text_cb("hello").result(timeout=10)
|
|
boot.pad.append_text.assert_called_once_with("hello ")
|
|
boot.text_system.inject_text.assert_not_called()
|
|
finally:
|
|
boot.stack.close()
|
|
|
|
def test_action_destination_decided_once_between_check_and_delivery(self) -> None:
|
|
"""The same single decision binds a pad-targeted editing command."""
|
|
boot = _boot_main_callbacks(dictate_to_pad=True)
|
|
try:
|
|
boot.config.is_dictate_to_pad_enabled.side_effect = [True, False]
|
|
boot.pad.handle_action.return_value = True
|
|
app_a = MagicMock()
|
|
app_a.identity_blob.return_value = "app-a"
|
|
with patch(
|
|
"vocalinux.text_injection.focused_window.get_focused_window",
|
|
return_value=app_a,
|
|
):
|
|
self.assertTrue(boot.action_cb("select_all").result(timeout=10))
|
|
boot.pad.handle_action.assert_called_once_with("select_all")
|
|
finally:
|
|
boot.stack.close()
|
|
|
|
def test_app_bound_segment_drops_on_focus_change(self) -> None:
|
|
"""App-bound text is still dropped when focus moved since the dictate."""
|
|
# A configured script arms the submit-time focus probe.
|
|
boot = _boot_main_callbacks(dictate_to_pad=False, post_script="/bin/cat")
|
|
try:
|
|
app_a = MagicMock()
|
|
app_a.identity_blob.return_value = "app-a"
|
|
app_b = MagicMock()
|
|
app_b.identity_blob.return_value = "app-b"
|
|
with patch(
|
|
"vocalinux.text_injection.focused_window.get_focused_window",
|
|
side_effect=[app_a, app_b],
|
|
):
|
|
boot.text_cb("hello").result(timeout=10)
|
|
boot.text_system.inject_text.assert_not_called()
|
|
boot.pad.append_text.assert_not_called()
|
|
finally:
|
|
boot.stack.close()
|
|
|
|
def test_pad_bound_action_survives_focus_change(self) -> None:
|
|
"""Pad-bound editing commands are not dropped on a focus change."""
|
|
boot = _boot_main_callbacks(dictate_to_pad=True)
|
|
try:
|
|
boot.pad.handle_action.return_value = True
|
|
app_a = MagicMock()
|
|
app_a.identity_blob.return_value = "app-a"
|
|
app_b = MagicMock()
|
|
app_b.identity_blob.return_value = "app-b"
|
|
with patch(
|
|
"vocalinux.text_injection.focused_window.get_focused_window",
|
|
side_effect=[app_a, app_b],
|
|
):
|
|
self.assertTrue(boot.action_cb("undo").result(timeout=10))
|
|
boot.pad.handle_action.assert_called_once_with("undo")
|
|
finally:
|
|
boot.stack.close()
|
|
|
|
def test_pad_redo_restores_delete_target(self) -> None:
|
|
"""A redo that brings the segment back must re-arm "delete that"."""
|
|
boot = _boot_main_callbacks(dictate_to_pad=True)
|
|
try:
|
|
boot.text_cb("only").result(timeout=10)
|
|
boot.pad.handle_action.return_value = True
|
|
|
|
# Undo to an empty pad clears the target but keeps the pad as the
|
|
# last dictation destination.
|
|
boot.pad.last_segment = None
|
|
self.assertTrue(boot.action_cb("undo").result(timeout=10))
|
|
self.assertTrue(boot.action_cb("delete_last").result(timeout=10))
|
|
boot.pad.delete_last_chars.assert_not_called()
|
|
|
|
# Redo restores the segment; the next delete reaches the pad.
|
|
boot.pad.last_segment = "only"
|
|
self.assertTrue(boot.action_cb("redo").result(timeout=10))
|
|
self.assertTrue(boot.action_cb("delete_last").result(timeout=10))
|
|
boot.pad.delete_last_chars.assert_called_once_with(len("only"))
|
|
boot.text_system.press_backspace.assert_not_called()
|
|
finally:
|
|
boot.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
|
|
# Stashed so tests can drive engine-side attributes (capture floors,
|
|
# the live recognition thread) through the same instance the
|
|
# callbacks close over.
|
|
self._engine = 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 _worker_deliverer(
|
|
self, segment_cb: Callable[[str, float], None]
|
|
) -> Tuple[Callable[[str], None], Callable[[], None]]:
|
|
"""Deliver segments from one persistent worker thread, like the real engine.
|
|
|
|
Returns (deliver, close): ``deliver(text)`` runs the segment callback
|
|
on the worker synchronously; ``close()`` stops it.
|
|
"""
|
|
work: queue.Queue = queue.Queue()
|
|
errors: list[BaseException] = []
|
|
|
|
def worker() -> None:
|
|
while True:
|
|
job = work.get()
|
|
try:
|
|
if job is not None:
|
|
job()
|
|
except BaseException as error:
|
|
errors.append(error)
|
|
finally:
|
|
work.task_done()
|
|
if job is None:
|
|
return
|
|
|
|
thread = threading.Thread(target=worker)
|
|
thread.start()
|
|
|
|
def deliver(text: str) -> None:
|
|
work.put(lambda: segment_cb(text, time.monotonic()))
|
|
work.join()
|
|
if errors:
|
|
raise errors.pop(0)
|
|
|
|
def close() -> None:
|
|
work.put(None)
|
|
work.join()
|
|
thread.join()
|
|
|
|
return deliver, 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()
|
|
on_old_worker, close_old_worker = self._worker_deliverer(segment_cb)
|
|
try:
|
|
state_cb(RecognitionState.LISTENING)
|
|
on_old_worker("one")
|
|
state_cb(RecognitionState.IDLE)
|
|
|
|
state_cb(RecognitionState.LISTENING)
|
|
segment_cb("two", time.monotonic())
|
|
# The previous session's worker finally delivers on its own thread.
|
|
on_old_worker("trailing")
|
|
self.assertEqual(history.get_all(), ["one trailing"])
|
|
|
|
state_cb(RecognitionState.IDLE)
|
|
self.assertEqual(history.get_all(), ["two", "one trailing"])
|
|
finally:
|
|
close_old_worker()
|
|
stack.close()
|
|
|
|
def test_straggler_after_newer_session_forms_own_snippet(self) -> None:
|
|
"""An old worker finishing after the next session must not grow its entry."""
|
|
stack, _, segment_cb, state_cb, history = self._boot()
|
|
on_old_worker, close_old_worker = self._worker_deliverer(segment_cb)
|
|
try:
|
|
state_cb(RecognitionState.LISTENING)
|
|
on_old_worker("first")
|
|
state_cb(RecognitionState.IDLE)
|
|
|
|
state_cb(RecognitionState.LISTENING)
|
|
segment_cb("second", time.monotonic())
|
|
state_cb(RecognitionState.IDLE)
|
|
|
|
# The older session's worker decodes last: its text still belongs
|
|
# to the first session — worker ownership extends that session's
|
|
# own snippet rather than the currently-newest entry or an
|
|
# orphan one.
|
|
on_old_worker("late tail")
|
|
self.assertEqual(history.get_all(), ["second", "first late tail"])
|
|
|
|
# A later straggler on that same worker still extends the same
|
|
# snippet it owns — one session never splits across entries.
|
|
on_old_worker("more tail")
|
|
self.assertEqual(history.get_all(), ["second", "first late tail more tail"])
|
|
finally:
|
|
close_old_worker()
|
|
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()
|
|
|
|
def test_straggler_extends_its_own_snippet_during_next_session(self) -> None:
|
|
"""A late segment lands in its session's entry, not the newest one."""
|
|
stack, _, segment_cb, state_cb, history = self._boot()
|
|
try:
|
|
state_cb(RecognitionState.LISTENING)
|
|
w1 = threading.Thread(target=segment_cb, args=("one", time.monotonic()))
|
|
self._engine.recognition_thread = w1
|
|
w1.start()
|
|
w1.join()
|
|
state_cb(RecognitionState.IDLE)
|
|
|
|
state_cb(RecognitionState.LISTENING)
|
|
# Session one's leftover worker finally delivers while session
|
|
# two is open — it must grow "one", never the pending "two".
|
|
straggler = threading.Thread(target=segment_cb, args=("tail", time.monotonic()))
|
|
straggler.start()
|
|
straggler.join()
|
|
# Session two's own segment arrives on its live worker.
|
|
w2 = threading.Thread(target=segment_cb, args=("two", time.monotonic()))
|
|
self._engine.recognition_thread = w2
|
|
w2.start()
|
|
w2.join()
|
|
state_cb(RecognitionState.IDLE)
|
|
self.assertEqual(history.get_all(), ["two", "one tail"])
|
|
finally:
|
|
stack.close()
|
|
|
|
def test_pre_session_capture_is_not_attributed_to_open_session(self) -> None:
|
|
"""Audio captured before a session opened cannot join its snippet."""
|
|
stack, _, segment_cb, state_cb, history = self._boot()
|
|
try:
|
|
state_cb(RecognitionState.LISTENING)
|
|
session_start = time.monotonic()
|
|
segment_cb("new", time.monotonic())
|
|
# A queued leftover captured before this session began, delivered
|
|
# on the session's worker, still belongs to what came before.
|
|
segment_cb("old", session_start - 10.0)
|
|
state_cb(RecognitionState.IDLE)
|
|
self.assertEqual(history.get_all(), ["new", "old"])
|
|
finally:
|
|
stack.close()
|
|
|
|
def test_mic_test_window_drops_segments_after_test_ends(self) -> None:
|
|
"""Test speech still decoding after restore stays out of history."""
|
|
stack, _, segment_cb, state_cb, history = self._boot()
|
|
try:
|
|
floor = time.monotonic()
|
|
self._engine.test_capture_floor = floor
|
|
self._engine.test_capture_ceiling = floor + 60.0
|
|
|
|
state_cb(RecognitionState.LISTENING)
|
|
segment_cb("test speech", floor + 5.0)
|
|
segment_cb("real dictation", floor + 120.0)
|
|
state_cb(RecognitionState.IDLE)
|
|
self.assertEqual(history.get_all(), ["real dictation"])
|
|
finally:
|
|
stack.close()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|