d11bd1bf0d
* feat(post-processing): pipe transcriptions through user-defined script Add a configurable post-processing step that passes each transcription result to an external executable via stdin and uses its stdout as the text to type. Enables custom transforms (formatting, LLM rewriting, etc.) without modifying Vocalinux. Falls back to original text on script failure or timeout. * test(post-processor): add unit tests and extract apply_post_processing Move config-reading logic from main.py closure into apply_post_processing() in post_processor.py so it can be tested independently. Add 11 unit tests covering PostProcessor.process() and apply_post_processing(). * docs(post-processing): document post-processing script feature * fix formatting of one file * fix(post-processing): run the user script with a stripped host environment Rebase compat: the host_process AST guard requires every subprocess spawned for a host binary to pass env=host_env() so an AppImage's LD_LIBRARY_PATH/GI_TYPELIB_PATH/PYTHONHOME do not leak in. Also give apply_post_processing() a typed ConfigManager parameter and keep text_callback_wrapper's local variable typed as str for mypy. * fix(post-processing): preserve paragraph breaks and tighten script handling Take script stdout verbatim instead of rstripping every trailing newline, so the "new paragraph" voice command's break survives a passthrough script. A single trailing newline is still dropped, but only when the transcription had none — it is almost always just the line terminator echo/print-style tools append. Also stop writing transformed output to the log (it can carry sensitive dictation into retained logs), catch only the expected process/OS/encoding failures instead of every Exception, and add the missing signature annotations. * fix(main): run the post-processing script off the recognition thread A user script may take seconds per segment, and text callbacks run on the recognition thread that drains the bounded audio-segment queue; stalling it drops queued dictation once the queue fills. Submit post-processing and injection to a dedicated single-worker executor so the backlog waits instead of losing speech, while keeping injection order. On timeout or failure the worker falls back to the original text, and an unexpected error is logged with a traceback and still injects the original rather than silently dropping the segment. Also pin the callback tests' get_str mock to no-script so they exercise the synchronous path deterministically, and add a callback test proving transformed text (including preserved paragraph breaks) reaches the injector. * docs(post-processing): annotate new callbacks and clarify stdout contract Add the missing type annotations on the new Settings page callbacks per the repository guide, and document that script stdout is used verbatim (a lone trailing newline added by line-oriented tools is dropped only when the transcription had none) and that scripts run on a dedicated worker. * test(main): annotate new post-processing callback tests * fix(main): bind queued dictation to its target app and drain worker on quit Route every transcription segment through the single post-processing worker (a pass-through when no script is configured) so clearing the script path mid-queue can no longer let a later segment overtake an earlier one still running. A segment that can sit in the backlog — a script is configured or a job is still running — is bound to the application focused when it was dictated; the worker drops it if focus has verifiably moved to another application, so delayed dictation cannot land in the wrong app. Wire the worker into the tray quit path via an on_quit hook: it clears the acceptance flag, cancels pending jobs, and waits — bounded by the script's own timeout — so nothing injects after the injector stops. * fix(main): keep voice actions ordered and probe focus off the recognition thread - Queue voice-command actions on the post-processing worker so an action can no longer overtake dictated text still waiting to be injected. - Move the submit-time focus probe onto a dedicated executor; the recognition thread only submits the probe, and the job joins the result at inject time, so compositor tools can no longer stall consumption of the bounded audio-segment queue. - Re-check the quit flag after the inject-time focus re-probe, then shut both workers down without waiting on a running script so quitting does not freeze on the script timeout. * fix(main): capture submit-time focus on dedicated threads and serialize quit - Replace the shared focus-probe executor with a fresh daemon thread per submission so a slow earlier probe cannot delay a later one past the focus change it is meant to capture; jobs read the result off a one-slot queue at inject time. - Give queued voice actions the same submit-time focus binding as text segments so a queued shortcut is dropped instead of fired into a newly focused application. - Hold injection_lock across the accept-check plus inject/action call so quit waits only for an in-flight injection instead of racing it. - Queue the IDLE reset of last_injected_text behind pending worker jobs so a queued 'delete that' still sees the text it targets. * fix(main): keep actions instant on a free worker and bound quit wait - An action now starts a submit-time focus probe only when it actually queues behind other jobs; with a free worker there is no delay window to bind against, so the probe's compositor calls are skipped and the action stays immediate even when a script is configured. - Probe and executor submission now happen inside pending_jobs_lock so queue order matches callback order — a job that saw an empty queue can no longer slip behind one that arrived while it was submitting. - Every inject-path subprocess call (xdotool window/key/type, wtype and ydotool type) is bounded by a timeout, and the quit hook's injection_lock wait is capped, so quitting can no longer freeze on a stalled injector subprocess. * fix(main): bind every action to submit-time focus without slowing it down - Every voice action now starts a submit-time focus probe: submission never guarantees immediate execution, so skipping the binding left a focus-change window in which a shortcut could fire into another app. The worker waits on the probe for at most one second — milliseconds on a healthy desktop — so immediate actions stay immediate while a genuinely stalled probe falls back to the injector's own targeting. - Quit waits for an in-flight injection without a cap again: every inject-path subprocess is individually bounded, so the wait always ends with the injection and a long transcription is never cut off. - The wtype/ydotool typing timeout is now derived from the configured per-key delay (a generous multiple), so a legitimately slow type can never be cut mid-text; a timed-out type no longer falls through to the clipboard fallback, which would duplicate the partial fragment. * fix(main): drop unverified actions and make quit-time injection abortable - An action whose submit-time focus probe cannot answer inside its one-second bound is now dropped instead of running unverified: a stalled probe can no longer leave a shortcut pointing at whatever application happens to be focused. - Quit asks the injector to abort in-flight injections, and injection paths honour it: the xdotool chunk loop stops between chunks, the wtype/ydotool type call runs as a polling Popen that can be killed, and paste/shortcut/backspace entry points exit early. The quit hook still waits for an in-flight injection, but the wait is bounded by at most one already-running subprocess call — never the injection's full length — and no subprocess is ever stopped mid-call by a timeout cap on the wait itself. - Annotate the remaining untyped test helpers. * fix(main): wait out slow focus probes; restore run() in wayland typing - An action no longer races its submit-time focus probe on a one-second bound: the job waits for the probe's own answer, whose compositor calls already carry per-call timeouts and short-circuit absent tools. A slow but living probe delays the command instead of discarding it, so a valid "undo" can no longer be lost to timing alone, and focus verification still decides every drop. - Wayland typing goes back to subprocess.run so tests keep intercepting the same entry point. Abortability is preserved by typing in 200-character chunks with the shutdown check between calls — a whole-string call could not be interrupted, but no single chunk holds the injector for more than its own scaled budget. - The abort flag gains a class-level default so injectors built without __init__ in tests still answer the checks; each real instance gets its own event in __init__. - Update call-shape assertions for the typed-call timeout and finish annotating the test methods touched by the callback refactor. * fix(inject): never retype text a failed chunked call already sent - _PartiallyTyped carries the confirmed-delivered prefix length out of the chunked type loops, so every downstream fallback continues from the untyped remainder: the Wayland-to-xdotool switch types only the tail, and the clipboard fallback copies only the tail, never the whole string a second time. - The xdotool retry loop gains the same accounting — a retry now resumes at the first untyped chunk instead of restarting at the top and duplicating everything the failed attempt already delivered. - Annotate the remaining test signature the repository guide requires. * fix(inject): report partial typing as failure even when clipboard lands A chunked type call that dies mid-text leaves only a prefix on screen, and the remainder-fallback keeps it that way: it copies the tail for a manual paste but now returns False, so the caller no longer records the full transcription as injected and a following "delete that" cannot erase text that was never typed. * fix(inject): keep deletion state aligned with delivered text inject_text now records how many characters a call confirmed on screen as last_typed_count: the whole length on success, the typed prefix after a partial failure, -1 when the delivered amount is unknowable. The transcription caller uses it to keep last_injected_text honest — a following "delete that" erases the typed prefix or nothing, never the previous segment or text that never arrived. * fix(inject): never replay input a timed-out portal job may have delivered - RemoteDesktopPortalError.delivered=None marks an unreported job's count as unknowable; all three portal fallbacks demote and skip instead of replaying, and text injection surfaces as TimeoutExpired so no full-text clipboard copy is offered - _PartiallyTyped counts raised after a portal demote are rebased onto the original text (typed_prefix) so the caller clips the real prefix - evdev: a consumed pure-modifier press is withheld and replayed ahead of the first non-target press, so holding RightAlt for AltGr composition still reaches the application - history: extend_latest gains an expected_latest CAS guard so a late segment from an older worker merges only into its own session's snippet, never into a newer session's entry - quit path aborts in-flight injections before waiting on the lock so quitting cannot freeze behind a long transcription's typing budget - tests: stop _ensure_ydotoold leaking a real Popen in Wayland tests * fix(shortcut): replay withheld modifiers across devices and after SYN_DROPPED - Replay withheld modifier presses from every device, not only the fd the character arrived on: split keyboards deliver AltGr and the character on separate devices, and a per-fd check forwarded a plain character. - Let the post-SYN_DROPPED resync run when only withheld modifiers exist for the fd, so a press whose release the dropped burst ate is pruned instead of replaying a stale modifier on the next ordinary key. * fix(history,pads): keep late segments pad-bound deliveries and pad deletions correct - Track the worker behind the ended session's snippet so late segments from an older worker cannot merge into a newer session's entry - Skip the focus-change drop for dictation-pad segments and pad-bound editing commands, which never target the focused application - Refuse 'delete that' when a manual pad edit blurred the segment boundaries instead of deleting the stale recorded length * style: drop unneeded nonlocal for mutated worker set * fix(shortcut): share one evdev device layer across shortcut listeners Each KeyboardShortcutManager built its own EvdevKeyboardBackend, and on evdev every backend opened and grabbed the same physical keyboards: the first grabber received all events and the rest received none, so per-language shortcuts could never fire on Wayland. All backends now bind to a process-wide EvdevDeviceHub that opens each keyboard once, owns the grabs and the paired uinput clones, and fans every event out to each registered engine. An event is re-emitted to applications only when no engine consumed it, so suppression semantics stay the same while per-engine matching state (combo progress, withheld modifiers, swallowed presses) remains independent. * fix(history): own late segments per worker and decide pad routing once - ended_worker_snippets maps each ended session's worker to its own (snippet, commit epoch): late deliveries merge into their session's entry however many sessions have committed since, so one session's stragglers never split across Recent Snippets entries. - The leftover-segment commit reads the live history epoch inside the session lock: a mid-session clear() no longer makes the stale ended epoch refuse valid post-clear dictation. - The pad-bound routing decision is captured once per job and passed to the injector/action handler, so a toggle flip between the focus check and delivery cannot reroute text or shortcuts into the focused app. - Dead recognition workers are pruned from the tracking set and map so long runs don't accumulate finished threads. - The custom-terms switch now suppresses GTK's own state-set on save failure so it shows the value that actually persisted. * fix(shortcut): serialize evdev device layer teardown against restart The last listener's unregister released the engine lock before joining the monitor thread and closing devices, so a new listener registering in that gap opened keyboards the stale teardown then closed — it stayed active but dead, and briefly grabbed keyboards stopped forwarding input. A lifecycle lock now covers the whole close/open transition in register, unregister, and the monitor's unexpected-exit teardown, so a cold start can only open devices after the previous generation finished closing (also making the new generation's grabs succeed). A generation stamp tells a monitor thread that outlived its join() its incarnation is over even when running has gone True again. * fix(injection): log focus-probe failures and narrow clipboard fallback catch A crashed focus probe turned into an unknown result with no logged error, indistinguishable from unavailable focus information; the partial-typing clipboard fallback caught every exception. Log the probe failure with context and narrow the clipboard catch to the exceptions the fallback path can actually raise. * fix(shortcut): join the evdev monitor outside the lifecycle lock If the monitor exited unexpectedly while the last listener unregistered, its cleanup queued on the lifecycle lock that unregister() held for the join — stalling shutdown for the join's full 2s timeout. The close now finishes inside the lock and the join waits after it is released, so the monitor's own cleanup can run and exit instead of queueing behind it. Closing before the join also dies the monitor's read/select on the closed fds right away. A _closed gate on _open_keyboard_device keeps a rescan from opening a device between detach and close that nothing would ever close. * fix(injection): narrow focus-probe catch to the raisable exception set get_focused_window() reports Optional and never raises; the catch now covers only what can realistically escape the queue write and logs it with context, keeping the waiting job released. * fix(shortcut): tag evdev device fds with the generation that opened them A monitor that outlived join() could still touch a newer generation's devices: the loop re-checked the generation only at the top, so mid-iteration it could dispatch from, forward to, or remove devices a restarted listener had just opened. Every fd is now tagged with the generation that opened it, and the tag is verified at each point a monitor could mutate shared state — fd lookup for dispatch, clone forwarding, forwarder release, device removal, and rescan opens — while dispatch re-checks the generation between events and the per-fd loop breaks on it. Cold starts bump the generation before opening so fresh devices are tagged for the new generation. Devices an older generation opened stay tagged stale, so its reader leaves them alone. --------- Co-authored-by: Jatin K Malik <jatinkrmalik@gmail.com>
2393 lines
97 KiB
Python
2393 lines
97 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 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()
|