xavierk/voculinux
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 3 Wiki Activity
Files
5e97773ebb0857369738fd6511f4a6aa35cd2e86
voculinux/tests/test_settings_test_dictation.py
T
Yunus Emre KarabağGitHubJatin K Malik
 a89966a744 feat(recognition): record while an idle model reloads (#851) 
* feat(recognition): buffer audio during model reload

* fix(recognition): recover buffered reload failures

* test(recognition): cover buffered reload edge cases

* fix(recognition): narrow buffered reload errors

* fix(settings): wait for IDLE before restoring Test Dictation callbacks

Buffered reload's stop_recognition() returns while PROCESSING. Test
Dictation must keep its text callbacks until recognition reaches IDLE
so the test utterance cannot inject into the focused app.

* style: apply Black formatting to PR 851 files

* fix(settings): cancel buffered reload before Test Dictation timeout restore

On the ~3min idle-wait timeout, do not restore live injector callbacks while
a buffered reload worker may still deliver text. Cancel via
_cancel_reload_recording off the GTK thread, then restore; if cancel is
unavailable or fails, leave test callbacks installed and show an error.
Add return/param annotations and regression coverage.

* fix(settings): keep Test Dictation waiting if buffered cancel fails

After the idle-wait timeout, if cancel is unavailable or raises, keep
polling with test callbacks and _test_active instead of marking the test
done. Ending early left the test callback as the only text destination,
so later normal dictation could land in the hidden test buffer. Restore
live injectors only once recognition reaches IDLE.

* fix(settings): reset Test Dictation timeout-cancel flag per session

Clear _test_timeout_cancel_attempted when a test starts, when entering
the buffered idle-wait, and when restoring UI so a later Test Dictation
can still cancel after timeout. Narrow the cancel-worker except clause
to AttributeError/OSError/RuntimeError/TypeError so unrelated defects
are not treated as ordinary cancel failure.

* fix(recognition): signal capture completion when buffered capture dies

A dead microphone during a buffered model reload left should_record=false
with no stop_recognition handoff, so the reload worker blocked forever on
_capture_finished.wait(). Signal the event from the audio thread's failure
teardown paths instead.

Refs VocaHQ/vocalinux#851

---------

Co-authored-by: Jatin K Malik <jatinkrmalik@gmail.com>
2026-09-29 22:02:56 -07:00

493 lines
20 KiB
Python
Raw Blame History

"""Tests for Settings → Test Dictation start failure vs real no-speech."""
import importlib
import sys
from typing import Any, Optional
from unittest.mock import MagicMock, Mock, patch
from vocalinux.common_types import RecognitionState
def _load_settings_dialog() -> Any:
"""Import settings_dialog with real bases so unbound methods stay callable.
conftest swaps gi for a MagicMock, which leaves every ``class X(Gtk.Y)`` as
a mock. Handing the three bases a real class keeps the methods intact.
Restore both sys.modules and the vocalinux.ui.settings_dialog attribute.
Python 3.9/3.10 mock follows the package attribute, so leaving a private
copy there makes later tests patch a different module than they imported.
"""
import vocalinux.ui as ui_pkg
repository = sys.modules["gi.repository"]
bases = {name: type(name, (), {}) for name in ("Box", "ListBoxRow", "Dialog")}
saved = sys.modules.pop("vocalinux.ui.settings_dialog", None)
try:
with patch.object(repository, "Gtk", MagicMock(**bases)):
module = importlib.import_module("vocalinux.ui.settings_dialog")
finally:
sys.modules.pop("vocalinux.ui.settings_dialog", None)
if saved is not None:
sys.modules["vocalinux.ui.settings_dialog"] = saved
ui_pkg.settings_dialog = saved
else:
ui_pkg.__dict__.pop("settings_dialog", None)
return module
settings_dialog = _load_settings_dialog()
SettingsDialog = settings_dialog.SettingsDialog
def _text_buffer() -> Mock:
"""Minimal stand-in for Gtk.TextBuffer used by the test output pane."""
buf = Mock()
buf._text = ""
def set_text(text: str) -> None:
buf._text = text
def get_text(*_args: Any, **_kwargs: Any) -> str:
return buf._text
def insert(_iter: Any, text: str) -> None:
buf._text += text
buf.set_text.side_effect = set_text
buf.get_text.side_effect = get_text
buf.insert.side_effect = insert
buf.get_start_iter.return_value = Mock()
buf.get_end_iter.return_value = Mock()
buf.get_insert.return_value = Mock()
return buf
def _dialog_for_test(
*, start_return: bool, model_ready: bool = True, is_auto_paused: bool = False
) -> Mock:
dialog = Mock()
dialog._test_active = False
dialog._applying_settings = False
dialog.test_button = Mock()
dialog.test_output_revealer = Mock()
dialog.test_buffer = _text_buffer()
dialog.test_textview = Mock()
dialog.config_manager = Mock()
dialog.config_manager.get_settings.return_value = {
"speech_recognition": {
"engine": "whisper_cpp",
"model_size": "tiny",
"silence_timeout": 2.0,
"vad_sensitivity": 3,
}
}
dialog.get_selected_settings = Mock(
return_value={
"engine": "whisper_cpp",
"model_size": "tiny",
"silence_timeout": 2.0,
"vad_sensitivity": 3,
}
)
dialog.apply_settings = Mock(return_value=True)
dialog.connect_to_recognition_manager = Mock()
dialog.update_recognition_progress = Mock()
dialog._test_text_callback = Mock()
dialog._stop_test_after_delay = SettingsDialog._stop_test_after_delay.__get__(dialog)
engine = Mock()
engine.state = RecognitionState.IDLE
engine.engine = "whisper_cpp"
engine.model_size = "tiny"
engine.start_recognition = Mock(return_value=start_return)
engine.stop_recognition = Mock()
engine.get_text_callbacks = Mock(return_value=[])
engine.set_text_callbacks = Mock()
engine.model_ready = model_ready
engine.is_auto_paused = is_auto_paused
dialog.speech_engine = engine
return dialog
def test_test_dictation_missing_model_skips_listen_timer() -> None:
dialog = _dialog_for_test(start_return=False, model_ready=False)
with patch.object(settings_dialog.threading, "Thread") as thread_cls:
SettingsDialog._on_test_clicked(dialog, None)
thread_cls.assert_not_called()
assert dialog._test_active is False
assert "Speech Model page" in dialog.test_buffer._text
assert "No speech detected" not in dialog.test_buffer._text
dialog.speech_engine.set_text_callbacks.assert_any_call([])
dialog.test_button.set_label.assert_not_called()
dialog.test_output_revealer.set_reveal_child.assert_called_with(True)
def test_test_dictation_auto_paused_message() -> None:
dialog = _dialog_for_test(start_return=False, is_auto_paused=True)
with patch.object(settings_dialog.threading, "Thread") as thread_cls:
SettingsDialog._on_test_clicked(dialog, None)
thread_cls.assert_not_called()
assert "paused" in dialog.test_buffer._text.lower()
assert "No speech detected" not in dialog.test_buffer._text
def test_test_dictation_start_failure_generic_message() -> None:
dialog = _dialog_for_test(start_return=False, model_ready=True, is_auto_paused=False)
with patch.object(settings_dialog.threading, "Thread") as thread_cls:
SettingsDialog._on_test_clicked(dialog, None)
thread_cls.assert_not_called()
assert dialog.test_buffer._text == "Could not start recognition test."
def test_test_dictation_started_arms_listen_timer() -> None:
dialog = _dialog_for_test(start_return=True)
with patch.object(settings_dialog.threading, "Thread") as thread_cls:
SettingsDialog._on_test_clicked(dialog, None)
thread_cls.assert_called_once()
assert thread_cls.call_args.kwargs["args"] == (4.0,)
assert dialog._test_active is True
dialog.test_button.set_label.assert_called_with("Testing… Speak Now!")
dialog.apply_settings.assert_not_called()
def test_test_dictation_reconfigures_when_live_engine_differs_from_ui() -> None:
"""UI matches the file but the live manager is still on another engine."""
dialog = _dialog_for_test(start_return=True, model_ready=True)
dialog.speech_engine.engine = "vosk"
dialog.speech_engine.model_size = "small"
dialog.speech_engine.model_ready = False
def _apply_and_sync() -> bool:
dialog.speech_engine.engine = "whisper_cpp"
dialog.speech_engine.model_size = "tiny"
dialog.speech_engine.model_ready = True
return True
dialog.apply_settings.side_effect = _apply_and_sync
with patch.object(settings_dialog.threading, "Thread") as thread_cls:
SettingsDialog._on_test_clicked(dialog, None)
dialog.apply_settings.assert_called_once()
thread_cls.assert_called_once()
assert dialog._test_active is True
def test_check_test_result_empty_buffer_is_no_speech() -> None:
dialog = Mock()
dialog.test_buffer = _text_buffer()
dialog.test_buffer.set_text("")
SettingsDialog._check_test_result(dialog)
assert dialog.test_buffer._text == "(No speech detected during test)"
def test_check_test_result_keeps_captured_text() -> None:
dialog = Mock()
dialog.test_buffer = _text_buffer()
dialog.test_buffer.set_text("hello world")
SettingsDialog._check_test_result(dialog)
assert dialog.test_buffer._text == "hello world"
def test_append_test_result_shows_callback_text() -> None:
dialog = Mock()
dialog.test_buffer = _text_buffer()
dialog.test_textview = Mock()
SettingsDialog._append_test_result(dialog, "hello")
assert dialog.test_buffer._text == "hello"
def _dialog_for_finalize(
*,
state: RecognitionState,
buffered_reload: bool = False,
saved_callbacks: Optional[list[Any]] = None,
has_cancel: bool = True,
) -> Mock:
"""Minimal dialog wired for _finalize_test / idle-wait helpers."""
dialog = Mock()
dialog._test_active = True
dialog._test_idle_wait_ticks = 0
dialog._test_timeout_cancel_attempted = False
dialog.test_button = Mock()
dialog.test_buffer = _text_buffer()
dialog.update_recognition_progress = Mock()
dialog._saved_text_callbacks = (
list(saved_callbacks) if saved_callbacks is not None else [Mock(name="live")]
)
dialog._test_text_callback = Mock(name="test_cb")
dialog._restore_callbacks_and_check_result = (
SettingsDialog._restore_callbacks_and_check_result.__get__(dialog)
)
dialog._wait_for_idle_then_restore_callbacks = (
SettingsDialog._wait_for_idle_then_restore_callbacks.__get__(dialog)
)
dialog._on_test_idle_wait_timeout = SettingsDialog._on_test_idle_wait_timeout.__get__(dialog)
dialog._cancel_buffered_reload_then_restore = (
SettingsDialog._cancel_buffered_reload_then_restore.__get__(dialog)
)
dialog._on_buffered_cancel_failed = SettingsDialog._on_buffered_cancel_failed.__get__(dialog)
dialog._finish_test_after_buffered_cancel = (
SettingsDialog._finish_test_after_buffered_cancel.__get__(dialog)
)
dialog._keep_waiting_after_timeout = SettingsDialog._keep_waiting_after_timeout.__get__(dialog)
dialog._finish_test_restore_ui = SettingsDialog._finish_test_restore_ui.__get__(dialog)
dialog._check_test_result = Mock(return_value=False)
if has_cancel:
engine = Mock()
engine._cancel_reload_recording = Mock()
else:
# getattr(..., "_cancel_reload_recording", None) must miss.
engine = Mock(
spec=[
"state",
"_buffered_reload_session",
"stop_recognition",
"set_text_callbacks",
]
)
engine.state = state
engine._buffered_reload_session = buffered_reload
engine.stop_recognition = Mock()
engine.set_text_callbacks = Mock()
dialog.speech_engine = engine
return dialog
def test_finalize_test_buffered_reload_waits_for_idle_before_restore() -> None:
"""stop_recognition may return while PROCESSING; keep test callbacks."""
dialog = _dialog_for_finalize(state=RecognitionState.PROCESSING, buffered_reload=True)
live = dialog._saved_text_callbacks[0]
with patch.object(settings_dialog, "GLib") as glib:
SettingsDialog._finalize_test(dialog)
dialog.speech_engine.stop_recognition.assert_called_once()
# Must not restore live injector yet — worker still holds the utterance.
dialog.speech_engine.set_text_callbacks.assert_not_called()
assert dialog._saved_text_callbacks == [live]
assert dialog._test_active is True
dialog.test_button.set_label.assert_called_with("Processing…")
glib.timeout_add.assert_called_once_with(100, dialog._wait_for_idle_then_restore_callbacks)
def test_wait_for_idle_keeps_polling_while_processing() -> None:
dialog = _dialog_for_finalize(state=RecognitionState.PROCESSING, buffered_reload=True)
assert SettingsDialog._wait_for_idle_then_restore_callbacks(dialog) is True
dialog.speech_engine.set_text_callbacks.assert_not_called()
assert dialog._test_active is True
assert dialog._test_idle_wait_ticks == 1
def test_wait_for_idle_restores_only_after_idle() -> None:
dialog = _dialog_for_finalize(state=RecognitionState.IDLE, buffered_reload=False)
live = dialog._saved_text_callbacks[0]
with patch.object(settings_dialog, "GLib") as glib:
result = SettingsDialog._wait_for_idle_then_restore_callbacks(dialog)
assert result is False
dialog.speech_engine.set_text_callbacks.assert_called_once_with([live])
assert not hasattr(dialog, "_saved_text_callbacks")
assert dialog._test_active is False
dialog.test_button.set_label.assert_called_with("Test Dictation")
glib.timeout_add.assert_called_once_with(300, dialog._check_test_result)
def test_wait_for_idle_treats_buffered_flag_as_busy_even_if_state_idle() -> None:
"""Defensive: session flag can lag a frame behind state transitions."""
dialog = _dialog_for_finalize(state=RecognitionState.IDLE, buffered_reload=True)
assert SettingsDialog._wait_for_idle_then_restore_callbacks(dialog) is True
dialog.speech_engine.set_text_callbacks.assert_not_called()
def test_finalize_test_idle_path_uses_settle_delay() -> None:
"""Non-buffered stop already reached IDLE — keep the legacy 500ms settle."""
dialog = _dialog_for_finalize(state=RecognitionState.IDLE, buffered_reload=False)
with patch.object(settings_dialog, "GLib") as glib:
SettingsDialog._finalize_test(dialog)
dialog.speech_engine.stop_recognition.assert_called_once()
assert dialog._test_active is False
dialog.test_button.set_label.assert_called_with("Test Dictation")
glib.timeout_add.assert_called_once_with(500, dialog._restore_callbacks_and_check_result)
# Live callbacks still saved until the settle timeout fires.
assert hasattr(dialog, "_saved_text_callbacks")
def test_timeout_does_not_restore_live_while_buffered_starts_cancel() -> None:
"""At ~3min timeout while buffered, cancel off-thread; do not restore yet."""
dialog = _dialog_for_finalize(state=RecognitionState.PROCESSING, buffered_reload=True)
live = dialog._saved_text_callbacks[0]
dialog._test_idle_wait_ticks = 1799
with patch.object(settings_dialog.threading, "Thread") as thread_cls:
result = SettingsDialog._wait_for_idle_then_restore_callbacks(dialog)
assert result is False
dialog.speech_engine.set_text_callbacks.assert_not_called()
assert dialog._saved_text_callbacks == [live]
assert dialog._test_active is True
dialog.test_button.set_label.assert_called_with("Cancelling…")
thread_cls.assert_called_once()
assert thread_cls.call_args.kwargs["target"] == dialog._cancel_buffered_reload_then_restore
assert thread_cls.call_args.kwargs["daemon"] is True
def test_timeout_without_cancel_keeps_waiting_active() -> None:
"""If cancel API is missing, keep polling with test callbacks / _test_active."""
dialog = _dialog_for_finalize(
state=RecognitionState.PROCESSING, buffered_reload=True, has_cancel=False
)
live = dialog._saved_text_callbacks[0]
dialog._test_idle_wait_ticks = 1799
with patch.object(settings_dialog.threading, "Thread") as thread_cls:
result = SettingsDialog._wait_for_idle_then_restore_callbacks(dialog)
assert result is True
thread_cls.assert_not_called()
dialog.speech_engine.set_text_callbacks.assert_not_called()
assert dialog._saved_text_callbacks == [live]
assert dialog._test_active is True
assert dialog._test_timeout_cancel_attempted is True
dialog.test_button.set_sensitive.assert_called_with(False)
dialog.test_button.set_label.assert_called_with("Still processing…")
assert "Timed out" in dialog.test_buffer._text
assert "Still waiting" in dialog.test_buffer._text
# Subsequent ticks past timeout must keep waiting, not restore.
result2 = SettingsDialog._wait_for_idle_then_restore_callbacks(dialog)
assert result2 is True
dialog.speech_engine.set_text_callbacks.assert_not_called()
def test_finish_after_buffered_cancel_restores_live_callbacks() -> None:
"""Only after cancel completes is it safe to put live injectors back."""
dialog = _dialog_for_finalize(state=RecognitionState.IDLE, buffered_reload=False)
live = dialog._saved_text_callbacks[0]
with patch.object(settings_dialog, "GLib") as glib:
result = SettingsDialog._finish_test_after_buffered_cancel(dialog)
assert result is False
dialog.speech_engine.set_text_callbacks.assert_called_once_with([live])
assert not hasattr(dialog, "_saved_text_callbacks")
glib.timeout_add.assert_called_once_with(300, dialog._check_test_result)
def test_cancel_buffered_reload_then_restore_schedules_finish() -> None:
dialog = _dialog_for_finalize(state=RecognitionState.PROCESSING, buffered_reload=True)
with patch.object(settings_dialog, "GLib") as glib:
SettingsDialog._cancel_buffered_reload_then_restore(dialog)
dialog.speech_engine._cancel_reload_recording.assert_called_once_with()
glib.idle_add.assert_called_once_with(dialog._finish_test_after_buffered_cancel)
def test_cancel_buffered_reload_failure_resumes_idle_wait() -> None:
dialog = _dialog_for_finalize(state=RecognitionState.PROCESSING, buffered_reload=True)
live = dialog._saved_text_callbacks[0]
dialog._test_timeout_cancel_attempted = True
dialog.speech_engine._cancel_reload_recording.side_effect = RuntimeError("boom")
with patch.object(settings_dialog, "GLib") as glib:
SettingsDialog._cancel_buffered_reload_then_restore(dialog)
glib.idle_add.assert_called_once_with(dialog._on_buffered_cancel_failed)
with patch.object(settings_dialog, "GLib") as glib2:
SettingsDialog._on_buffered_cancel_failed(dialog)
dialog.speech_engine.set_text_callbacks.assert_not_called()
assert dialog._saved_text_callbacks == [live]
assert dialog._test_active is True
dialog.test_button.set_sensitive.assert_called_with(False)
dialog.test_button.set_label.assert_called_with("Still processing…")
assert "cancel" in dialog.test_buffer._text.lower()
glib2.timeout_add.assert_called_once_with(100, dialog._wait_for_idle_then_restore_callbacks)
# Resumed poll must not restore while still busy, and must not re-cancel.
with patch.object(settings_dialog.threading, "Thread") as thread_cls:
assert SettingsDialog._wait_for_idle_then_restore_callbacks(dialog) is True
thread_cls.assert_not_called()
dialog.speech_engine.set_text_callbacks.assert_not_called()
def test_keep_waiting_then_idle_restores_live_callbacks() -> None:
"""After a no-cancel timeout, IDLE eventually restores live injectors."""
dialog = _dialog_for_finalize(
state=RecognitionState.PROCESSING, buffered_reload=True, has_cancel=False
)
live = dialog._saved_text_callbacks[0]
dialog._test_idle_wait_ticks = 1799
SettingsDialog._wait_for_idle_then_restore_callbacks(dialog)
assert dialog._test_active is True
dialog.speech_engine.state = RecognitionState.IDLE
dialog.speech_engine._buffered_reload_session = False
with patch.object(settings_dialog, "GLib") as glib:
result = SettingsDialog._wait_for_idle_then_restore_callbacks(dialog)
assert result is False
dialog.speech_engine.set_text_callbacks.assert_called_once_with([live])
assert not hasattr(dialog, "_saved_text_callbacks")
assert dialog._test_active is False
glib.timeout_add.assert_called_once_with(300, dialog._check_test_result)
def test_timeout_cancel_flag_resets_on_new_finalize_wait() -> None:
"""A later Test Dictation session must be allowed to cancel again."""
dialog = _dialog_for_finalize(state=RecognitionState.PROCESSING, buffered_reload=True)
dialog._test_timeout_cancel_attempted = True
dialog._test_idle_wait_ticks = 0
with patch.object(settings_dialog, "GLib") as glib:
SettingsDialog._finalize_test(dialog)
assert dialog._test_timeout_cancel_attempted is False
assert dialog._test_idle_wait_ticks == 0
glib.timeout_add.assert_called_once_with(100, dialog._wait_for_idle_then_restore_callbacks)
dialog._test_idle_wait_ticks = 1799
with patch.object(settings_dialog.threading, "Thread") as thread_cls:
assert SettingsDialog._wait_for_idle_then_restore_callbacks(dialog) is False
thread_cls.assert_called_once()
assert thread_cls.call_args.kwargs["target"] == dialog._cancel_buffered_reload_then_restore
def test_cancel_worker_reraises_unexpected_defect() -> None:
"""Programming defects must not be swallowed as cancel-failed recovery."""
dialog = _dialog_for_finalize(state=RecognitionState.PROCESSING, buffered_reload=True)
dialog.speech_engine._cancel_reload_recording.side_effect = KeyError("bug")
with patch.object(settings_dialog, "GLib") as glib:
try:
SettingsDialog._cancel_buffered_reload_then_restore(dialog)
raised = False
except KeyError:
raised = True
assert raised is True
glib.idle_add.assert_not_called()
Reference in New Issue View Git Blame Copy Permalink