feat(ui): in-app Dictation Pad for Wayland-safe dictation (#887)
* feat(ui): add in-app Dictation Pad for Wayland-safe dictation Dictated text can now be routed into a text box inside Vocalinux's own window instead of being injected into other applications — an explicit fallback for desktops where cross-app injection is unreliable or blocked (notably Wayland). The pad opens from the tray menu, captures dictation when enabled via Settings or its own checkbox, and lets the user copy text out by hand. 'delete that' removes the last segment in-pad. Closes #726. * fix(ui): consult live config for pad auto-reveal and Copy All The reveal gate used the controller's enabled flag, which only syncs on pad show or its own checkbox — enabling capture via the Settings switch while the pad was closed silently buffered dictation with no reveal. _apply_append now reads dictate_to_pad from the shared config. Also: Copy All copies the widget text (dictation plus in-pad edits), and re-showing the pad no longer does a wholesale set_text — the view stays incrementally in sync, so manual edits survive hide/reopen. * fix(ui): keep dictation pad capture and commands inside the pad Address all Greptile findings on PR #887: - Read the capture toggle from the shared ConfigManager instead of re-parsing config.json per segment, so a torn mid-write read can no longer fail-open and send dictation to the focused app - Sync edits made directly in the pad widget back into the controller, keeping delete semantics consistent with the visible text - Stamp queued widget operations with a generation tag so a Clear drops appends still sitting in the GTK idle queue - Route undo/redo, select all/line/word/paragraph and cut/copy/paste to pad-local implementations while capturing, so their shortcuts never reach whichever application holds focus - Track where the last segment was delivered so "delete that" follows the text rather than the current capture toggle - Annotate every new signature and add callback-level tests that pin the routing, queue, and undo behavior * fix(ui): keep pad controller, queued ops, and delete target in sync Manual edits could overwrite queued dictation in the controller because GTK input events outrank idle callbacks; the changed handler now replays pending widget ops before syncing the view. The controller also tracks dictated segments across undo snapshots, so a pad undo pops the "delete that" target instead of leaving a stale length. Catches in the pad are narrowed to the GTK/config error set per repo style, and missing type annotations are added. * fix(ui): close flush races in pad idle queue and delete retarget Flushed idle ops stayed scheduled with GLib and ran a second time, duplicating appends; each queued op now only acts while its entry is still pending. A queued full-view refresh could also erase a manual edit that landed first, so the changed-handler flush now drops stale set-text ops instead of replaying them. And clearing the pad destination on undo-to-empty kept a redo from restoring the "delete that" target; the destination flag is preserved so redo re-arms it. * fix(ui): replay pad idle ops before edits land, not after Flushing in the changed handler still let queued appends leapfrog the keystroke that fired it, and a stale dispatch could cancel a later queued op with identical text. Pending entries now match by identity, and the replay moved to insert-text/delete-range — signals that fire before the edit is applied — so queued dictation and deletes keep true chronological order against manual edits. * fix(history): target late segments to their own snippet by id A straggler from a just-ended session could merge into whichever snippet was newest when it arrived, and audio captured before a session opened was attributed to that session while decoding on its worker. - TranscriptionHistory.add returns the snippet id and the new extend_entry() extends a specific entry, replacing the positional extend_latest target in the late-segment path - session_started_floor stamps when a session opens; in-session attribution also requires the capture to postdate it - test_capture_ceiling closes the mic-test window at callback restore so test speech still decoding can no longer leak into history * fix(settings): keep deferred edits from overwriting a newer choice The deferred-persist filter treated "config still equals the baseline" as untouched, so a reopened dialog that applied a key back to its baseline value could still lose it to the closed dialog worker. - Record the generation of the apply that last wrote each advanced key; a pending edit now survives only when no newer apply wrote that key - A worker apply whose snapshot a newer apply predates now skips its reconfigure and save entirely, so collection order — not lock order — decides which snapshot wins * fix(settings): skip stale applies before stopping dictation The superseded-apply check ran after stop_recognition, so a stale worker still interrupted a live dictation session. Recognition now stops inside the apply lock, after the generation check. * fix(shortcuts): keep language edits from breaking dictation Two per-language shortcut fixes (#805): - A binding edit landing while a session is live no longer stops recognition. The dialog's per-keystroke persist still saves the new binding, but the tray defers the listener rebuild to the next IDLE transition, so a held push-to-talk keeps its release callback and a toggle session is never cut off mid-utterance. - A row edited onto a key another row holds can no longer displace that row's binding. Each row's remembered shortcut is reserved for its own row before claims run in row order, so the stealing edit is rejected and the held binding survives. - Persists that resolve to the already-saved bindings (mid-edit keystrokes, rejected duplicates, reverted fields) skip the save and the listener rebuild entirely. --------- Co-authored-by: Jatin K Malik <jatinkrmalik@gmail.com>
This commit is contained in:
devin-ai-integration[bot]
co-authored by
Jatin K Malik
Jatin K Malik
parent
dbe0215aad
commit
9898bb8dd9
+119
-31
@@ -8,6 +8,7 @@ import atexit
|
||||
import logging
|
||||
import sys
|
||||
import threading
|
||||
import time
|
||||
from typing import Optional
|
||||
|
||||
from .utils.vosk_model_info import SUPPORTED_LANGUAGES
|
||||
@@ -352,8 +353,6 @@ def main():
|
||||
if not single_instance.acquire_lock():
|
||||
# Another instance is already running - show notification and exit
|
||||
try:
|
||||
import time
|
||||
|
||||
from gi.repository import Notify
|
||||
|
||||
Notify.init("Vocalinux")
|
||||
@@ -601,16 +600,45 @@ def main():
|
||||
session_lock = threading.Lock()
|
||||
session_segments: list[tuple[str, float]] = []
|
||||
session_open = False
|
||||
# Monotonic time the open session started; a segment delivered on the
|
||||
# live worker whose capture predates it belongs to an earlier session.
|
||||
session_started_floor = 0.0
|
||||
# Worker thread that produced the open session's segments; used to
|
||||
# detect callbacks from a previous session's still-running worker.
|
||||
session_worker: Optional[threading.Thread] = None
|
||||
# True when the newest history entry is the just-closed session's
|
||||
# snippet, so late segments can still merge into it.
|
||||
latest_snippet_extendable = False
|
||||
# Id of the just-closed session's snippet (from
|
||||
# TranscriptionHistory.add), so late segments extend that entry
|
||||
# rather than whichever snippet happens to be newest.
|
||||
latest_snippet_id: Optional[int] = None
|
||||
# Clear epoch the most-recently-ended session was committed under;
|
||||
# late segments merging into its snippet are judged against it.
|
||||
ended_session_epoch = transcription_history.epoch
|
||||
|
||||
# In-app dictation pad: the Wayland-proof fallback target that
|
||||
# receives dictated text when "dictate to pad" capture is enabled
|
||||
# (#726). Constructed before the tray so the tray menu can open it.
|
||||
from .ui.dictation_pad import DictationPad
|
||||
|
||||
dictation_pad = DictationPad(
|
||||
enabled=config_manager.is_dictate_to_pad_enabled(),
|
||||
config_manager=config_manager,
|
||||
)
|
||||
|
||||
def dictate_to_pad_enabled() -> bool:
|
||||
"""Read the live capture toggle from the shared config manager.
|
||||
|
||||
The shared manager's in-memory cache is updated by Settings and
|
||||
the pad checkbox before save_config rewrites config.json, so this
|
||||
can never flip on a torn mid-write disk read and send dictated
|
||||
text to whichever application holds focus.
|
||||
"""
|
||||
return bool(config_manager.is_dictate_to_pad_enabled())
|
||||
|
||||
# Where the last delivered segment went. "delete that" follows the
|
||||
# text, not the current toggle: the capture setting may have flipped
|
||||
# since the segment was delivered.
|
||||
last_injected = {"to_pad": False}
|
||||
|
||||
# --- Callback wiring ---------------------------------------------------
|
||||
# The speech engine emits four kinds of events, each handled by a
|
||||
# dedicated callback registered below:
|
||||
@@ -667,7 +695,7 @@ def main():
|
||||
captured after the clear are kept, so dictation that continues
|
||||
across a clear is still recoverable.
|
||||
"""
|
||||
nonlocal session_worker, latest_snippet_extendable
|
||||
nonlocal session_worker, latest_snippet_id
|
||||
if not transcription_history.enabled:
|
||||
return
|
||||
segment = _normalize_segment_text(segment)
|
||||
@@ -682,35 +710,49 @@ def main():
|
||||
# Captured before the last clear — must not re-enter.
|
||||
return
|
||||
# Segments captured while the Settings mic test runs are test
|
||||
# speech, not dictation; the floor is stamped before the test
|
||||
# starts recognition, so leftovers from an earlier session —
|
||||
# whose capture began before it — still file normally.
|
||||
# speech, not dictation. The [floor, ceiling) window survives
|
||||
# the test's end, so segments still decoding after it cannot
|
||||
# leak into history.
|
||||
test_floor = getattr(speech_engine, "test_capture_floor", None)
|
||||
if isinstance(test_floor, (int, float)) and started_at >= test_floor:
|
||||
test_ceiling = getattr(speech_engine, "test_capture_ceiling", None)
|
||||
if (
|
||||
isinstance(test_floor, (int, float))
|
||||
and started_at >= test_floor
|
||||
and (not isinstance(test_ceiling, (int, float)) or started_at < test_ceiling)
|
||||
):
|
||||
return
|
||||
if session_open and (
|
||||
worker is session_worker
|
||||
or worker is current_worker
|
||||
# When the engine exposes no worker (tests, mocks), the
|
||||
# first segment of a session tags it.
|
||||
or (session_worker is None and not isinstance(current_worker, threading.Thread))
|
||||
if (
|
||||
session_open
|
||||
and started_at >= session_started_floor
|
||||
and (
|
||||
worker is session_worker
|
||||
or worker is current_worker
|
||||
# When the engine exposes no worker (tests, mocks), the
|
||||
# first segment of a session tags it.
|
||||
or (
|
||||
session_worker is None
|
||||
and not isinstance(current_worker, threading.Thread)
|
||||
)
|
||||
)
|
||||
):
|
||||
session_worker = worker
|
||||
session_segments.append((segment, started_at))
|
||||
return
|
||||
# Late segment from a session that already ended: merge into
|
||||
# its committed snippet when there is one. Both writes are
|
||||
# guarded by the ended session's epoch, so a clear() landing
|
||||
# between that session's commit and this delivery still
|
||||
# refuses the text.
|
||||
if latest_snippet_extendable and transcription_history.extend_latest(
|
||||
segment, expected_epoch=ended_session_epoch
|
||||
# its own snippet by id — a newer session may already have
|
||||
# committed on top, so the newest entry is not the target.
|
||||
# Both writes are guarded by the ended session's epoch, so a
|
||||
# clear() landing between that commit and this delivery
|
||||
# still refuses the text.
|
||||
if latest_snippet_id is not None and transcription_history.extend_entry(
|
||||
latest_snippet_id, segment, expected_epoch=ended_session_epoch
|
||||
):
|
||||
return
|
||||
# Otherwise the late segments are the session's only output
|
||||
# and form their own snippet.
|
||||
if transcription_history.add(segment, expected_epoch=ended_session_epoch):
|
||||
latest_snippet_extendable = True
|
||||
snippet_id = transcription_history.add(segment, expected_epoch=ended_session_epoch)
|
||||
if snippet_id is not None:
|
||||
latest_snippet_id = snippet_id
|
||||
|
||||
def text_callback_wrapper(text: str) -> None:
|
||||
"""Bridge between speech engine text events and the text injector.
|
||||
@@ -746,9 +788,17 @@ def main():
|
||||
text_to_inject = " " + text_to_inject
|
||||
logger.debug("Added space separator before new segment")
|
||||
|
||||
success = text_system.inject_text(text_to_inject)
|
||||
captured = dictate_to_pad_enabled()
|
||||
if captured:
|
||||
# In-app capture: skip cross-application injection entirely
|
||||
# and land the text in the pad instead (#726).
|
||||
dictation_pad.append_text(text_to_inject)
|
||||
success = True
|
||||
else:
|
||||
success = text_system.inject_text(text_to_inject)
|
||||
if success:
|
||||
action_handler.set_last_injected_text(text_to_inject)
|
||||
last_injected["to_pad"] = captured
|
||||
|
||||
def on_state_change(state: RecognitionState) -> None:
|
||||
"""Reset the last-injected buffer when a listening session ends.
|
||||
@@ -756,11 +806,12 @@ def main():
|
||||
Also commits the just-finished dictation session to the
|
||||
transcription history as a single snippet.
|
||||
"""
|
||||
nonlocal session_open, session_worker, latest_snippet_extendable
|
||||
nonlocal ended_session_epoch
|
||||
nonlocal session_open, session_worker, latest_snippet_id
|
||||
nonlocal ended_session_epoch, session_started_floor
|
||||
if state in (RecognitionState.IDLE, RecognitionState.ERROR):
|
||||
if state == RecognitionState.IDLE:
|
||||
action_handler.set_last_injected_text("")
|
||||
last_injected["to_pad"] = False
|
||||
with session_lock:
|
||||
session_open = False
|
||||
session_worker = None
|
||||
@@ -775,10 +826,10 @@ def main():
|
||||
session_segments.clear()
|
||||
# Guarded by the epoch observed here: a clear() landing
|
||||
# between this read and the add still refuses the
|
||||
# snippet. The committed snippet stays open to late
|
||||
# snippet. The committed snippet's id stays open to late
|
||||
# segments still trickling out of the worker; a session
|
||||
# that produced no text leaves no entry to merge into.
|
||||
latest_snippet_extendable = transcription_history.add(
|
||||
latest_snippet_id = transcription_history.add(
|
||||
joined, expected_epoch=ended_session_epoch
|
||||
)
|
||||
else:
|
||||
@@ -786,13 +837,14 @@ def main():
|
||||
if not session_open:
|
||||
session_open = True
|
||||
session_worker = None
|
||||
session_started_floor = time.monotonic()
|
||||
# Segments left over by a session that ended without a
|
||||
# closing state commit as their own snippet rather
|
||||
# than leaking into the new session's — still only
|
||||
# those captured after the last clear().
|
||||
if session_segments:
|
||||
cleared_at = transcription_history.cleared_at
|
||||
latest_snippet_extendable = transcription_history.add(
|
||||
latest_snippet_id = transcription_history.add(
|
||||
" ".join(
|
||||
text
|
||||
for text, started_at in session_segments
|
||||
@@ -807,8 +859,43 @@ def main():
|
||||
# session clear() can separate pre-clear speech from new dictation;
|
||||
# injection keeps the plain text callback.
|
||||
speech_engine.register_segment_callback(record_history_segment)
|
||||
|
||||
def action_callback_wrapper(action: str) -> bool:
|
||||
"""Route editing voice commands to the pad while capture is on.
|
||||
|
||||
"delete that" follows the segment it removes: the pad when the
|
||||
last delivered text went there (even if capture was toggled off
|
||||
since) and the focused application when it was injected (even if
|
||||
capture was toggled on). Every other editing command is handled
|
||||
pad-side while capturing so its shortcuts never leak into
|
||||
whichever application holds focus.
|
||||
"""
|
||||
if action == "delete_last":
|
||||
if not action_handler.last_injected_text:
|
||||
return True
|
||||
if last_injected["to_pad"]:
|
||||
# The pad's own segment bookkeeping wins over the recorded
|
||||
# text: a pad "undo" may have popped that segment, leaving
|
||||
# its length stale.
|
||||
target = dictation_pad.last_segment or action_handler.last_injected_text
|
||||
deleted = dictation_pad.delete_last_chars(len(target))
|
||||
if deleted:
|
||||
action_handler.set_last_injected_text("")
|
||||
last_injected["to_pad"] = False
|
||||
return True
|
||||
return bool(action_handler.handle_action(action))
|
||||
if dictate_to_pad_enabled():
|
||||
handled = bool(dictation_pad.handle_action(action))
|
||||
if handled and action in ("undo", "redo") and last_injected["to_pad"]:
|
||||
# Pad history moved: retarget "delete that" at the segment
|
||||
# now at the pad's tail. Keep the pad destination even
|
||||
# when the tail is empty — a redo can restore it.
|
||||
action_handler.set_last_injected_text(dictation_pad.last_segment or "")
|
||||
return handled
|
||||
return bool(action_handler.handle_action(action))
|
||||
|
||||
speech_engine.register_text_callback(text_callback_wrapper)
|
||||
speech_engine.register_action_callback(action_handler.handle_action)
|
||||
speech_engine.register_action_callback(action_callback_wrapper)
|
||||
speech_engine.register_state_callback(on_state_change)
|
||||
|
||||
# Initialize and start the system tray indicator
|
||||
@@ -816,6 +903,7 @@ def main():
|
||||
speech_engine=speech_engine,
|
||||
text_injector=text_system,
|
||||
transcription_history=transcription_history,
|
||||
dictation_pad=dictation_pad,
|
||||
)
|
||||
|
||||
# Start the GTK main loop
|
||||
|
||||
@@ -415,8 +415,12 @@ class SpeechRecognitionManager:
|
||||
self.segment_callbacks: list[Callable[[str, float], None]] = []
|
||||
# While the Settings microphone test runs, segments whose capture
|
||||
# began at or after this monotonic time are test speech; consumers
|
||||
# (e.g. history) read the floor to skip them.
|
||||
# (e.g. history) read the floor to skip them. When the test ends the
|
||||
# ceiling closes the window — segments captured during it but still
|
||||
# decoding must keep failing the check, so the floor alone is never
|
||||
# lifted back to None.
|
||||
self.test_capture_floor: Optional[float] = None
|
||||
self.test_capture_ceiling: Optional[float] = None
|
||||
self.state_callbacks: list[Callable[[RecognitionState], None]] = []
|
||||
self.action_callbacks: list[Callable[[str], None]] = []
|
||||
|
||||
|
||||
@@ -199,6 +199,9 @@ DEFAULT_CONFIG = {
|
||||
# "ydotool"/"xdotool" when autodetection is wrong (#476).
|
||||
# VOCALINUX_FORCE_BACKEND overrides this for a single run.
|
||||
"backend": "auto",
|
||||
# Route dictation into the in-app Dictation Pad window instead of
|
||||
# injecting into other apps — the Wayland-safe fallback (#726).
|
||||
"dictate_to_pad": False,
|
||||
},
|
||||
"history": {
|
||||
"enabled": True, # Keep recent dictation snippets in the tray menu
|
||||
@@ -881,6 +884,14 @@ class ConfigManager:
|
||||
"""Enable or disable the floating dictation overlay."""
|
||||
self.set("ui", "show_overlay", bool(enabled))
|
||||
|
||||
def is_dictate_to_pad_enabled(self) -> bool:
|
||||
"""Check if dictation is routed into the in-app pad (default False)."""
|
||||
return self.get_bool("text_injection", "dictate_to_pad", False)
|
||||
|
||||
def set_dictate_to_pad(self, enabled: bool) -> None:
|
||||
"""Route dictation into the in-app Dictation Pad instead of injecting."""
|
||||
self.set("text_injection", "dictate_to_pad", bool(enabled))
|
||||
|
||||
def _update_dict_recursive(self, target: dict, source: dict):
|
||||
"""
|
||||
Update a dictionary recursively.
|
||||
|
||||
@@ -0,0 +1,677 @@
|
||||
"""
|
||||
In-app dictation pad for Vocalinux.
|
||||
|
||||
A small persistent window with an editable text view that receives dictated
|
||||
text directly, bypassing cross-application text injection entirely. It is the
|
||||
explicit fallback path for desktops where injecting keystrokes into another
|
||||
window is unreliable or blocked (notably Wayland compositors): the user
|
||||
dictates into the pad, then selects and copies the text into the real target.
|
||||
|
||||
Like DictationOverlay, the pure buffer logic lives in DictationPadController
|
||||
so unit tests can drive it without a display; the GTK wrapper degrades to a
|
||||
headless no-op when GTK is unavailable.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from typing import TYPE_CHECKING, Any, Callable, Optional
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from .config_manager import ConfigManager
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
_PAD_WIDTH = 480
|
||||
_PAD_HEIGHT = 360
|
||||
_COPIED_FEEDBACK_MS = 1200
|
||||
_UNDO_LIMIT = 200
|
||||
|
||||
|
||||
def _ui_errors() -> tuple[type[BaseException], ...]:
|
||||
"""Exception types tolerated by the pad's UI glue.
|
||||
|
||||
GTK and config calls fail with a bounded set of runtime exceptions —
|
||||
naming them (instead of ``Exception``) keeps real bugs visible while
|
||||
the pad still degrades gracefully on systems without a usable display.
|
||||
``GLib.Error`` joins only when gi is importable, so the module stays
|
||||
GTK-free for tests.
|
||||
"""
|
||||
errors: tuple[type[BaseException], ...] = (
|
||||
ImportError,
|
||||
RuntimeError,
|
||||
ValueError,
|
||||
TypeError,
|
||||
AttributeError,
|
||||
OSError,
|
||||
)
|
||||
try:
|
||||
from gi.repository import GLib
|
||||
except (ImportError, ValueError):
|
||||
return errors
|
||||
# Under gi-mocked test runs GLib.Error is a MagicMock, not a class —
|
||||
# only real exception types may join the tuple.
|
||||
glib_error = getattr(GLib, "Error", None)
|
||||
if isinstance(glib_error, type) and issubclass(glib_error, BaseException):
|
||||
return errors + (glib_error,)
|
||||
return errors
|
||||
|
||||
|
||||
# Voice commands the pad performs itself while capture mode is on. They are
|
||||
# the same actions ActionHandler would otherwise send as keystrokes into
|
||||
# whichever application happens to hold focus.
|
||||
_HISTORY_ACTIONS = frozenset({"undo", "redo"})
|
||||
_WIDGET_ACTIONS = frozenset(
|
||||
{"select_all", "select_line", "select_word", "select_paragraph", "cut", "copy", "paste"}
|
||||
)
|
||||
|
||||
|
||||
class DictationPadController:
|
||||
"""
|
||||
Pure buffer state for the dictation pad: the accumulated text plus a
|
||||
bounded undo history.
|
||||
|
||||
Separated from GTK so unit tests can exercise append/delete semantics
|
||||
without a display.
|
||||
"""
|
||||
|
||||
def __init__(self, enabled: bool = False) -> None:
|
||||
self._enabled = bool(enabled)
|
||||
self._text = ""
|
||||
# Dictated segments still present in the buffer, oldest first. Manual
|
||||
# widget edits blur the boundaries (the list clears); undo snapshots
|
||||
# restore them, so "delete that" can retarget after history moves.
|
||||
self._segments: list[str] = []
|
||||
self._undo_stack: list[tuple[str, list[str]]] = []
|
||||
self._redo_stack: list[tuple[str, list[str]]] = []
|
||||
|
||||
@property
|
||||
def enabled(self) -> bool:
|
||||
"""Whether dictation is routed into the pad instead of injected."""
|
||||
return self._enabled
|
||||
|
||||
def set_enabled(self, enabled: bool) -> None:
|
||||
"""Enable or disable capture mode (does not touch the buffer)."""
|
||||
self._enabled = bool(enabled)
|
||||
|
||||
@property
|
||||
def text(self) -> str:
|
||||
"""The full contents of the pad."""
|
||||
return self._text
|
||||
|
||||
@property
|
||||
def last_segment(self) -> Optional[str]:
|
||||
"""The last dictated segment still in the buffer.
|
||||
|
||||
``None`` when the buffer holds no tracked dictation — empty, cleared,
|
||||
or rewritten by a manual edit. Undo/redo restore the tracked tail
|
||||
along with the text snapshot.
|
||||
"""
|
||||
return self._segments[-1] if self._segments else None
|
||||
|
||||
def append(self, text: str) -> None:
|
||||
"""Append a transcription segment to the end of the buffer."""
|
||||
if not text:
|
||||
return
|
||||
self._record_undo()
|
||||
self._text += text
|
||||
self._segments.append(text)
|
||||
|
||||
def set_text(self, text: str) -> None:
|
||||
"""Replace the buffer (e.g. edits made directly in the widget)."""
|
||||
if text == self._text:
|
||||
return
|
||||
self._record_undo()
|
||||
self._text = text
|
||||
# An arbitrary replacement blurs segment boundaries; undo restores
|
||||
# them from the snapshot.
|
||||
self._segments = []
|
||||
|
||||
def delete_last(self, count: int) -> int:
|
||||
"""
|
||||
Delete up to ``count`` characters from the end of the buffer.
|
||||
|
||||
Returns the number of characters actually removed.
|
||||
"""
|
||||
if count <= 0 or not self._text:
|
||||
return 0
|
||||
deleted = min(count, len(self._text))
|
||||
self._record_undo()
|
||||
self._text = self._text[:-deleted]
|
||||
self._trim_segments(deleted)
|
||||
return deleted
|
||||
|
||||
def clear(self) -> None:
|
||||
"""Drop everything in the pad."""
|
||||
if not self._text:
|
||||
return
|
||||
self._record_undo()
|
||||
self._text = ""
|
||||
self._segments = []
|
||||
|
||||
def undo(self) -> bool:
|
||||
"""Revert the last buffer mutation. Returns False with empty history."""
|
||||
if not self._undo_stack:
|
||||
return False
|
||||
self._redo_stack.append((self._text, list(self._segments)))
|
||||
self._text, self._segments = self._undo_stack.pop()
|
||||
return True
|
||||
|
||||
def redo(self) -> bool:
|
||||
"""Re-apply the last undone mutation. Returns False with no redo lane."""
|
||||
if not self._redo_stack:
|
||||
return False
|
||||
self._undo_stack.append((self._text, list(self._segments)))
|
||||
self._text, self._segments = self._redo_stack.pop()
|
||||
return True
|
||||
|
||||
def _trim_segments(self, count: int) -> None:
|
||||
"""Drop ``count`` characters from the tracked segment tail."""
|
||||
remaining = count
|
||||
while remaining > 0 and self._segments:
|
||||
tail = self._segments[-1]
|
||||
if len(tail) <= remaining:
|
||||
self._segments.pop()
|
||||
remaining -= len(tail)
|
||||
else:
|
||||
self._segments[-1] = tail[:-remaining]
|
||||
remaining = 0
|
||||
|
||||
def _record_undo(self) -> None:
|
||||
"""Snapshot the buffer before a mutation; new edits drop the redo lane."""
|
||||
self._undo_stack.append((self._text, list(self._segments)))
|
||||
if len(self._undo_stack) > _UNDO_LIMIT:
|
||||
del self._undo_stack[0]
|
||||
self._redo_stack.clear()
|
||||
|
||||
|
||||
class DictationPad:
|
||||
"""
|
||||
The dictation pad window: a GTK window backed by DictationPadController.
|
||||
|
||||
Text arrives on the recognition thread, so every mutation of the visible
|
||||
widget is marshalled onto the GTK main loop with GLib.idle_add. Closing
|
||||
the window hides it — the buffer survives for the session.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
enabled: bool = False,
|
||||
config_manager: Optional["ConfigManager"] = None,
|
||||
) -> None:
|
||||
self.controller = DictationPadController(enabled=enabled)
|
||||
self._config_manager = config_manager
|
||||
self._window: Any = None
|
||||
self._textview: Any = None
|
||||
self._buffer: Any = None
|
||||
self._capture_check: Any = None
|
||||
self._copy_button: Any = None
|
||||
self._gtk_ready = False
|
||||
self._syncing_capture_check = False
|
||||
self._syncing_widget = False
|
||||
# Monotonic tag stamped on every queued widget op. A Clear (or a full
|
||||
# refresh) bumps it, so appends still waiting in the GTK idle queue
|
||||
# can tell they are stale and must not resurrect removed text.
|
||||
self._generation = 0
|
||||
# Widget ops waiting in the GTK idle queue, in queue order. GTK input
|
||||
# events outrank idle callbacks, so a manual edit can land before a
|
||||
# queued append — _flush_idle_ops replays this list to keep the view
|
||||
# in the order the controller already applied.
|
||||
self._pending_idle: list[tuple[Callable[..., None], tuple[Any, ...]]] = []
|
||||
self._copied_feedback_id: Optional[int] = None
|
||||
|
||||
try:
|
||||
self._init_gtk_window()
|
||||
self._gtk_ready = True
|
||||
except _ui_errors() as e:
|
||||
# Headless / missing display: keep the controller usable for tests.
|
||||
logger.warning("Dictation pad window unavailable: %s", e)
|
||||
|
||||
def _init_gtk_window(self) -> None:
|
||||
"""Construct the pad window, text view and action buttons."""
|
||||
import gi
|
||||
|
||||
gi.require_version("Gtk", "3.0")
|
||||
gi.require_version("Gdk", "3.0")
|
||||
from gi.repository import Gdk, GLib, Gtk # noqa: F401
|
||||
|
||||
self._GLib = GLib
|
||||
self._Gdk = Gdk
|
||||
self._Gtk = Gtk
|
||||
|
||||
window = Gtk.Window(type=Gtk.WindowType.TOPLEVEL)
|
||||
window.set_title("Vocalinux Dictation Pad")
|
||||
window.set_default_size(_PAD_WIDTH, _PAD_HEIGHT)
|
||||
# Closing only hides: the pad keeps its buffer for the whole session
|
||||
# and must not take the app down with it.
|
||||
window.connect("delete-event", self._on_delete_event)
|
||||
|
||||
outer = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=8)
|
||||
outer.set_margin_start(12)
|
||||
outer.set_margin_end(12)
|
||||
outer.set_margin_top(12)
|
||||
outer.set_margin_bottom(12)
|
||||
window.add(outer)
|
||||
|
||||
self._capture_check = Gtk.CheckButton(
|
||||
label="Dictate into this window instead of injecting into other apps"
|
||||
)
|
||||
self._capture_check.set_tooltip_text(
|
||||
"Fallback for desktops where text injection does not work "
|
||||
"(e.g. Wayland): dictated text lands in this box, then you copy "
|
||||
"it out by hand."
|
||||
)
|
||||
self._capture_check.set_active(self.controller.enabled)
|
||||
self._capture_check.connect("toggled", self._on_capture_toggled)
|
||||
outer.pack_start(self._capture_check, False, False, 0)
|
||||
|
||||
scrolled = Gtk.ScrolledWindow()
|
||||
scrolled.set_policy(Gtk.PolicyType.AUTOMATIC, Gtk.PolicyType.AUTOMATIC)
|
||||
scrolled.set_hexpand(True)
|
||||
scrolled.set_vexpand(True)
|
||||
outer.pack_start(scrolled, True, True, 0)
|
||||
|
||||
self._textview = Gtk.TextView()
|
||||
self._textview.set_editable(True)
|
||||
self._textview.set_wrap_mode(Gtk.WrapMode.WORD_CHAR)
|
||||
self._textview.set_margin_start(8)
|
||||
self._textview.set_margin_end(8)
|
||||
self._textview.set_margin_top(8)
|
||||
self._textview.set_margin_bottom(8)
|
||||
self._buffer = self._textview.get_buffer()
|
||||
# Manual edits in the widget feed back into the controller so
|
||||
# deletion history never computes against stale dictated text.
|
||||
self._buffer.connect("changed", self._on_buffer_changed)
|
||||
# insert-text/delete-range fire BEFORE the edit is applied, so the
|
||||
# queued dictation ops replay ahead of it in true chronological order.
|
||||
self._buffer.connect("insert-text", self._on_buffer_user_edit)
|
||||
self._buffer.connect("delete-range", self._on_buffer_user_edit)
|
||||
scrolled.add(self._textview)
|
||||
|
||||
button_row = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=8)
|
||||
button_row.set_halign(Gtk.Align.END)
|
||||
outer.pack_start(button_row, False, False, 0)
|
||||
|
||||
self._copy_button = Gtk.Button(label="Copy All")
|
||||
self._copy_button.set_tooltip_text("Copy the entire pad to the clipboard")
|
||||
self._copy_button.connect("clicked", self._on_copy_all_clicked)
|
||||
button_row.pack_start(self._copy_button, False, False, 0)
|
||||
|
||||
clear_button = Gtk.Button(label="Clear")
|
||||
clear_button.set_tooltip_text("Erase everything in the pad")
|
||||
clear_button.connect("clicked", self._on_clear_clicked)
|
||||
button_row.pack_start(clear_button, False, False, 0)
|
||||
|
||||
self._window = window
|
||||
|
||||
# -- public API ---------------------------------------------------------
|
||||
|
||||
def append_text(self, text: str) -> None:
|
||||
"""
|
||||
Append a transcription segment (safe to call from any thread).
|
||||
|
||||
The buffer is updated immediately so a headless pad still captures
|
||||
text; the widget refresh and auto-show happen on the GTK main loop,
|
||||
tagged with the current generation so a later Clear can drop it.
|
||||
"""
|
||||
self.controller.append(text)
|
||||
self._idle_add(self._apply_append, text, self._generation)
|
||||
|
||||
def delete_last_chars(self, count: int) -> int:
|
||||
"""
|
||||
Delete up to ``count`` characters from the end (any thread).
|
||||
|
||||
Used by the "delete that" voice command while capture mode is on.
|
||||
Returns the number of characters actually removed.
|
||||
"""
|
||||
deleted = self.controller.delete_last(count)
|
||||
if deleted:
|
||||
self._idle_add(self._apply_delete, deleted, self._generation)
|
||||
return deleted
|
||||
|
||||
@property
|
||||
def last_segment(self) -> Optional[str]:
|
||||
"""The last dictated segment still in the pad (None if untracked)."""
|
||||
return self.controller.last_segment
|
||||
|
||||
def show_pad(self) -> None:
|
||||
"""Show the pad window, resyncing capture state from live config."""
|
||||
if not self._gtk_ready or self._window is None:
|
||||
return
|
||||
self._sync_capture_check()
|
||||
if not self._window.get_visible():
|
||||
self._window.show_all()
|
||||
try:
|
||||
self._window.present_with_time(self._Gtk.get_current_event_time())
|
||||
except _ui_errors():
|
||||
pass
|
||||
|
||||
def set_capture_enabled(self, enabled: bool) -> None:
|
||||
"""Programmatically flip capture mode (keeps the checkbox in sync)."""
|
||||
self.controller.set_enabled(enabled)
|
||||
self._sync_capture_check()
|
||||
|
||||
def destroy(self) -> None:
|
||||
"""Tear down the window and any pending feedback timer."""
|
||||
self._bump_generation()
|
||||
self._buffer = None
|
||||
self._textview = None
|
||||
if self._copied_feedback_id is not None:
|
||||
try:
|
||||
self._GLib.source_remove(self._copied_feedback_id)
|
||||
except _ui_errors():
|
||||
pass
|
||||
self._copied_feedback_id = None
|
||||
if self._window is not None:
|
||||
try:
|
||||
self._window.destroy()
|
||||
except _ui_errors():
|
||||
pass
|
||||
self._window = None
|
||||
|
||||
# -- internals ----------------------------------------------------------
|
||||
|
||||
def _idle_add(self, func: Callable[..., None], *args: Any) -> None:
|
||||
"""Schedule ``func`` on the GTK main loop when GTK is available."""
|
||||
glib = getattr(self, "_GLib", None)
|
||||
if glib is None:
|
||||
return
|
||||
entry = (func, args)
|
||||
|
||||
def _call() -> bool:
|
||||
# A flush may have already run this op; GLib still dispatches the
|
||||
# callback, so only act while this entry is still pending. Match
|
||||
# by identity — an equal payload must not cancel a later twin.
|
||||
idx = next(
|
||||
(i for i, pending in enumerate(self._pending_idle) if pending is entry),
|
||||
None,
|
||||
)
|
||||
if idx is not None:
|
||||
del self._pending_idle[idx]
|
||||
func(*args)
|
||||
return False
|
||||
|
||||
self._pending_idle.append(entry)
|
||||
glib.idle_add(_call)
|
||||
|
||||
def _bump_generation(self) -> None:
|
||||
"""Invalidate queued widget ops and forget them.
|
||||
|
||||
Stale ops still in the GLib queue drop themselves by generation;
|
||||
clearing the list keeps a later flush from replaying them first.
|
||||
"""
|
||||
self._generation += 1
|
||||
self._pending_idle = []
|
||||
|
||||
def _flush_idle_ops(self, keep_user_view: bool = False) -> None:
|
||||
"""Replay queued widget ops so the view catches up with the controller.
|
||||
|
||||
GTK input events outrank idle callbacks: a keystroke can land in the
|
||||
buffer before appends queued earlier, and the "changed" handler would
|
||||
then write the pre-append view back over the controller — losing the
|
||||
queued dictation. Running the queue first keeps the view (and the
|
||||
controller sync) in the order the controller already applied.
|
||||
|
||||
With ``keep_user_view`` — a manual edit being synced back — ops that
|
||||
replace the whole buffer are dropped instead of run: the keystroke
|
||||
already landed on the newer view, so a stale refresh would erase it.
|
||||
"""
|
||||
pending = self._pending_idle
|
||||
self._pending_idle = []
|
||||
for func, args in pending:
|
||||
if keep_user_view and func == self._apply_set_text:
|
||||
continue
|
||||
func(*args)
|
||||
|
||||
def _apply_append(self, text: str, generation: int) -> None:
|
||||
"""Insert a segment at the end of the widget and keep the tail visible."""
|
||||
if self._buffer is None or generation != self._generation:
|
||||
# Cleared or refreshed while this insert sat in the idle queue.
|
||||
return
|
||||
try:
|
||||
self._syncing_widget = True
|
||||
self._buffer.insert(self._buffer.get_end_iter(), text)
|
||||
# Scroll to the end without place_cursor: moving the caret would
|
||||
# collapse a selection the user is making for copy-out.
|
||||
self._textview.scroll_to_iter(self._buffer.get_end_iter(), 0.0, True, 0.0, 1.0)
|
||||
except _ui_errors() as e:
|
||||
logger.debug("Could not append text to dictation pad: %s", e)
|
||||
finally:
|
||||
self._syncing_widget = False
|
||||
# Reveal the pad on the first incoming segment; once visible, further
|
||||
# appends update silently so the window never re-raises mid-selection.
|
||||
if self._capture_enabled() and self._window is not None and not self._window.get_visible():
|
||||
self.show_pad()
|
||||
|
||||
def _apply_delete(self, deleted: int, generation: int) -> None:
|
||||
"""Remove ``deleted`` characters from the end of the widget."""
|
||||
if self._buffer is None or generation != self._generation:
|
||||
return
|
||||
try:
|
||||
self._syncing_widget = True
|
||||
end = self._buffer.get_end_iter()
|
||||
start = end.copy()
|
||||
start.backward_chars(deleted)
|
||||
self._buffer.delete(start, end)
|
||||
except _ui_errors() as e:
|
||||
logger.debug("Could not delete text in dictation pad: %s", e)
|
||||
finally:
|
||||
self._syncing_widget = False
|
||||
|
||||
def _apply_set_text(self, text: str, generation: int) -> None:
|
||||
"""Replace the widget contents with ``text``."""
|
||||
if self._buffer is None or generation != self._generation:
|
||||
return
|
||||
try:
|
||||
self._syncing_widget = True
|
||||
self._buffer.set_text(text)
|
||||
except _ui_errors() as e:
|
||||
logger.debug("Could not update dictation pad view: %s", e)
|
||||
finally:
|
||||
self._syncing_widget = False
|
||||
|
||||
def _apply_action(self, action: str, generation: int) -> None:
|
||||
"""Execute a selection or clipboard command on the GTK main loop."""
|
||||
if self._buffer is None or generation != self._generation:
|
||||
return
|
||||
try:
|
||||
if action == "cut":
|
||||
self._buffer.cut_clipboard(self._clipboard(), True)
|
||||
elif action == "copy":
|
||||
self._buffer.copy_clipboard(self._clipboard())
|
||||
elif action == "paste":
|
||||
self._buffer.paste_clipboard(self._clipboard(), None, True)
|
||||
else:
|
||||
start, end = self._selection_bounds(action)
|
||||
if start is not None:
|
||||
self._buffer.select_range(start, end)
|
||||
except _ui_errors() as e:
|
||||
logger.debug("Could not perform %s in dictation pad: %s", action, e)
|
||||
|
||||
def _clipboard(self) -> Any:
|
||||
"""Return the shared CLIPBOARD selection."""
|
||||
return self._Gtk.Clipboard.get(self._Gdk.SELECTION_CLIPBOARD)
|
||||
|
||||
def _selection_bounds(self, action: str) -> tuple[Optional[Any], Optional[Any]]:
|
||||
"""Iter pair for a selection ``action`` taken at the insert mark.
|
||||
|
||||
Mirrors the shortcuts ActionHandler would inject: select_line is
|
||||
Home+Shift+End, select_word is Ctrl+Shift+Right, and
|
||||
select_paragraph extends to the next blank line (``\\n\\n`` delimits
|
||||
paragraphs, matching the "new paragraph" voice command).
|
||||
"""
|
||||
if action == "select_all":
|
||||
return self._buffer.get_start_iter(), self._buffer.get_end_iter()
|
||||
it = self._buffer.get_iter_at_mark(self._buffer.get_insert())
|
||||
start = it.copy()
|
||||
end = it.copy()
|
||||
if action == "select_line":
|
||||
start.set_line_offset(0)
|
||||
if not end.ends_line():
|
||||
end.forward_to_line_end()
|
||||
return start, end
|
||||
if action == "select_word":
|
||||
end.forward_word_end()
|
||||
return start, end
|
||||
if action == "select_paragraph":
|
||||
text = self._buffer.get_text(
|
||||
self._buffer.get_start_iter(), self._buffer.get_end_iter(), False
|
||||
)
|
||||
boundary = text.find("\n\n", it.get_offset())
|
||||
end = self._buffer.get_iter_at_offset(len(text) if boundary < 0 else boundary)
|
||||
return start, end
|
||||
return None, None
|
||||
|
||||
def _sync_widget_from_controller(self) -> None:
|
||||
"""Queue a full widget refresh, dropping appends queued before it."""
|
||||
self._bump_generation()
|
||||
self._idle_add(self._apply_set_text, self.controller.text, self._generation)
|
||||
|
||||
def _on_buffer_user_edit(self, *_args: Any) -> None:
|
||||
"""Replay queued dictation ops before a manual edit reaches the buffer.
|
||||
|
||||
GTK input events outrank idle callbacks, so a keystroke can land
|
||||
before an append queued earlier. insert-text/delete-range fire before
|
||||
the edit is applied, so flushing here keeps the widget (and the
|
||||
controller sync that follows) in the order the controller applied.
|
||||
"""
|
||||
if self._syncing_widget:
|
||||
return
|
||||
self._flush_idle_ops()
|
||||
|
||||
def _on_buffer_changed(self, buffer: Any) -> None:
|
||||
"""Mirror edits made directly in the widget back into the controller."""
|
||||
if self._syncing_widget:
|
||||
return
|
||||
# Fallback for edit paths that bypass insert-text/delete-range. A
|
||||
# stale whole-view refresh must not erase the edit that just landed.
|
||||
self._flush_idle_ops(keep_user_view=True)
|
||||
try:
|
||||
text = buffer.get_text(buffer.get_start_iter(), buffer.get_end_iter(), False)
|
||||
except _ui_errors() as e:
|
||||
logger.debug("Could not sync dictation pad edits: %s", e)
|
||||
return
|
||||
self.controller.set_text(text)
|
||||
|
||||
def _capture_enabled(self) -> bool:
|
||||
"""Return whether dictation is currently routed into the pad.
|
||||
|
||||
Prefers the live config value so a Settings change applies even
|
||||
before the pad is ever shown; falls back to the controller flag
|
||||
when no config manager is attached.
|
||||
"""
|
||||
if self._config_manager is None:
|
||||
return self.controller.enabled
|
||||
try:
|
||||
enabled = bool(self._config_manager.get_bool("text_injection", "dictate_to_pad", False))
|
||||
except (AttributeError, TypeError, KeyError):
|
||||
return self.controller.enabled
|
||||
self.controller.set_enabled(enabled)
|
||||
return enabled
|
||||
|
||||
def _sync_capture_check(self) -> None:
|
||||
"""Mirror the persisted capture setting onto the checkbox."""
|
||||
if self._capture_check is None:
|
||||
return
|
||||
try:
|
||||
enabled = self._capture_enabled()
|
||||
self._syncing_capture_check = True
|
||||
self._capture_check.set_active(enabled)
|
||||
except _ui_errors() as e:
|
||||
logger.debug("Could not sync dictation pad toggle: %s", e)
|
||||
finally:
|
||||
self._syncing_capture_check = False
|
||||
|
||||
def _on_delete_event(self, *_args: Any) -> bool:
|
||||
"""Hide on close instead of destroying — the buffer is session-scoped."""
|
||||
if self._window is not None:
|
||||
self._window.hide()
|
||||
return True
|
||||
|
||||
def _on_capture_toggled(self, widget: Any) -> None:
|
||||
"""Persist the pad's own capture checkbox."""
|
||||
if self._syncing_capture_check:
|
||||
return
|
||||
enabled = bool(widget.get_active())
|
||||
self.controller.set_enabled(enabled)
|
||||
if self._config_manager is not None:
|
||||
try:
|
||||
self._config_manager.set("text_injection", "dictate_to_pad", enabled)
|
||||
self._config_manager.save_settings()
|
||||
except (AttributeError, TypeError) as e:
|
||||
logger.warning("Could not save dictation pad setting: %s", e)
|
||||
logger.info("Dictation pad capture %s", "enabled" if enabled else "disabled")
|
||||
|
||||
def _on_copy_all_clicked(self, *_args: Any) -> None:
|
||||
"""Copy the entire buffer to the clipboard and flash a confirmation."""
|
||||
# The widget's text may include the user's own in-pad edits; fall back
|
||||
# to the dictated buffer when there is no widget (headless).
|
||||
text = self.controller.text
|
||||
if self._buffer is not None:
|
||||
# Land queued appends first so Copy All never misses the latest
|
||||
# dictated segment still sitting in the idle queue.
|
||||
self._flush_idle_ops()
|
||||
try:
|
||||
text = self._buffer.get_text(
|
||||
self._buffer.get_start_iter(), self._buffer.get_end_iter(), False
|
||||
)
|
||||
except _ui_errors():
|
||||
pass
|
||||
try:
|
||||
clipboard = self._clipboard()
|
||||
clipboard.set_text(text, -1)
|
||||
clipboard.store()
|
||||
except _ui_errors() as e:
|
||||
logger.warning("Could not copy dictation pad contents: %s", e)
|
||||
return
|
||||
if self._copy_button is None or self._copied_feedback_id is not None:
|
||||
return
|
||||
self._copy_button.set_label("Copied")
|
||||
self._copied_feedback_id = self._GLib.timeout_add(
|
||||
_COPIED_FEEDBACK_MS, self._reset_copy_button
|
||||
)
|
||||
|
||||
def _reset_copy_button(self) -> bool:
|
||||
"""Restore the Copy All label after the brief confirmation flash."""
|
||||
if self._copy_button is not None:
|
||||
self._copy_button.set_label("Copy All")
|
||||
self._copied_feedback_id = None
|
||||
return False
|
||||
|
||||
def _on_clear_clicked(self, *_args: Any) -> None:
|
||||
"""Erase the buffer and drop appends still queued for the widget."""
|
||||
# Bump the generation first: any append or delete already sitting in
|
||||
# the GTK idle queue becomes stale and cannot resurrect cleared text.
|
||||
self._bump_generation()
|
||||
self.controller.clear()
|
||||
if self._buffer is not None:
|
||||
try:
|
||||
self._syncing_widget = True
|
||||
self._buffer.set_text("")
|
||||
except _ui_errors() as e:
|
||||
logger.debug("Could not clear dictation pad view: %s", e)
|
||||
finally:
|
||||
self._syncing_widget = False
|
||||
|
||||
def handle_action(self, action: str) -> bool:
|
||||
"""
|
||||
Perform an editing voice command on the pad (any thread).
|
||||
|
||||
While capture mode is on, the pad owns every editing action — they
|
||||
must never reach the application that happens to hold keyboard
|
||||
focus. History commands run on the controller; selection and
|
||||
clipboard commands are marshalled onto the GTK main loop.
|
||||
"""
|
||||
if action in _HISTORY_ACTIONS:
|
||||
changed = self.controller.undo() if action == "undo" else self.controller.redo()
|
||||
if changed:
|
||||
self._sync_widget_from_controller()
|
||||
return changed
|
||||
if action in _WIDGET_ACTIONS:
|
||||
if self._buffer is None:
|
||||
logger.debug("Dictation pad cannot %s without a text view", action)
|
||||
return False
|
||||
self._idle_add(self._apply_action, action, self._generation)
|
||||
return True
|
||||
logger.warning("Dictation pad does not handle action: %s", action)
|
||||
return False
|
||||
@@ -102,6 +102,7 @@ from .config_manager import ( # noqa: E402
|
||||
DEFAULT_SOUND_EFFECT_TONE,
|
||||
PASTE_SHORTCUTS,
|
||||
SOUND_EFFECT_TONES,
|
||||
normalize_language_shortcuts,
|
||||
resolve_whispercpp_variant,
|
||||
)
|
||||
from .keyboard_backends import ( # noqa: E402
|
||||
@@ -140,6 +141,11 @@ logger = logging.getLogger(__name__)
|
||||
# must agree on whose snapshot is newest.
|
||||
_apply_settings_lock = threading.Lock()
|
||||
_apply_settings_generation = 0
|
||||
# Generation of the apply that last wrote each advanced key. A deferred
|
||||
# persist uses it to tell "a newer apply rewrote this key" apart from "the
|
||||
# config still holds the baseline" — a newer user choice can legally equal
|
||||
# that baseline, so comparing values alone is not enough.
|
||||
_apply_settings_written: dict[str, int] = {}
|
||||
|
||||
|
||||
def _raw_audio_device_name(device_name: Optional[str]) -> Optional[str]:
|
||||
@@ -3735,6 +3741,18 @@ class SettingsDialog(Gtk.Dialog):
|
||||
logger.info(f"Copy to clipboard {'enabled' if enabled else 'disabled'}")
|
||||
return False
|
||||
|
||||
def _on_dictation_pad_toggled(self, widget: Gtk.Switch, state: bool) -> bool:
|
||||
"""Handle toggle of the in-app dictation pad switch."""
|
||||
if _handlers_suppressed(self):
|
||||
return False
|
||||
|
||||
enabled = bool(state)
|
||||
logger.info(f"Dictate to pad toggled: {enabled}")
|
||||
self.config_manager.set("text_injection", "dictate_to_pad", enabled)
|
||||
self.config_manager.save_settings()
|
||||
logger.info(f"Dictate to pad {'enabled' if enabled else 'disabled'}")
|
||||
return False
|
||||
|
||||
def _on_auto_capitalize_toggled(self, widget, state):
|
||||
"""Handle toggle of the auto-capitalize switch."""
|
||||
if _handlers_suppressed(self):
|
||||
@@ -4289,6 +4307,19 @@ class SettingsDialog(Gtk.Dialog):
|
||||
),
|
||||
)
|
||||
|
||||
self.dictation_pad_switch = _add_switch_row(
|
||||
output_group,
|
||||
title="Dictation Pad",
|
||||
subtitle="Capture dictation in an in-app text box instead of other apps",
|
||||
keywords=("wayland", "pad", "fallback", "text box"),
|
||||
tooltip=(
|
||||
"Type dictation into Vocalinux's own Dictation Pad window instead "
|
||||
"of injecting it into other apps. The pad opens from the tray menu; "
|
||||
"copy text out of it by hand. Useful on Wayland, where injecting "
|
||||
"keystrokes into other windows is restricted."
|
||||
),
|
||||
)
|
||||
|
||||
self.append_trailing_space_switch = _add_switch_row(
|
||||
output_group,
|
||||
title="Trailing Space After Dictation",
|
||||
@@ -4323,6 +4354,7 @@ class SettingsDialog(Gtk.Dialog):
|
||||
|
||||
self.recognition_settings_tab.pack_start(output_group, False, False, 0)
|
||||
self.copy_to_clipboard_switch.connect("state-set", self._on_copy_to_clipboard_toggled)
|
||||
self.dictation_pad_switch.connect("state-set", self._on_dictation_pad_toggled)
|
||||
self.auto_capitalize_switch.connect("state-set", self._on_auto_capitalize_toggled)
|
||||
self.append_trailing_space_switch.connect(
|
||||
"state-set", self._on_append_trailing_space_toggled
|
||||
@@ -4755,17 +4787,33 @@ class SettingsDialog(Gtk.Dialog):
|
||||
"""
|
||||
if self._initializing:
|
||||
return
|
||||
# Each row's remembered binding belongs to that row first, so a
|
||||
# shortcut typed into one row can never displace a binding another
|
||||
# row already holds — the stealing edit is rejected instead.
|
||||
owners: dict[str, int] = {}
|
||||
for index, refs in enumerate(self._language_shortcut_rows):
|
||||
owned = (refs.get("last_valid_shortcut") or "").strip().lower()
|
||||
if owned and owned not in owners:
|
||||
owners[owned] = index
|
||||
entries = []
|
||||
rejected = []
|
||||
claimed = set()
|
||||
for refs in self._language_shortcut_rows:
|
||||
for index, refs in enumerate(self._language_shortcut_rows):
|
||||
language = refs["language_picker"].get_active_id() or "auto"
|
||||
shortcut = refs["shortcut_entry"].get_text().strip().lower()
|
||||
last_valid = refs.get("last_valid_shortcut", "")
|
||||
if shortcut and is_valid_shortcut(shortcut) and shortcut not in claimed:
|
||||
owner = owners.get(shortcut)
|
||||
if (
|
||||
shortcut
|
||||
and is_valid_shortcut(shortcut)
|
||||
and shortcut not in claimed
|
||||
and (owner is None or owner == index)
|
||||
):
|
||||
refs["last_valid_shortcut"] = shortcut
|
||||
claimed.add(shortcut)
|
||||
elif last_valid and last_valid not in claimed:
|
||||
elif (
|
||||
last_valid and last_valid not in claimed and owners.get(last_valid, index) == index
|
||||
):
|
||||
if shortcut and is_valid_shortcut(shortcut):
|
||||
rejected.append(shortcut)
|
||||
shortcut = last_valid
|
||||
@@ -4775,6 +4823,12 @@ class SettingsDialog(Gtk.Dialog):
|
||||
rejected.append(shortcut)
|
||||
continue
|
||||
entries.append({"shortcut": shortcut, "language": language})
|
||||
if normalize_language_shortcuts(entries) == self.config_manager.get_language_shortcuts():
|
||||
# Nothing effective changed — a mid-edit keystroke, a rejected
|
||||
# duplicate, or a reverted field — so the live listeners do not
|
||||
# need another rebuild (and dictation never pauses for one).
|
||||
self._report_language_shortcut_rejections(rejected)
|
||||
return
|
||||
self.config_manager.set_language_shortcuts(entries)
|
||||
self.config_manager.save_settings()
|
||||
if self.language_shortcuts_update_callback:
|
||||
@@ -6475,11 +6529,16 @@ class SettingsDialog(Gtk.Dialog):
|
||||
# A newer apply already reconfigured the shared engine
|
||||
# and saved newer values. Keep only the keys it left
|
||||
# untouched — the rest of the snapshot is stale and
|
||||
# must not reach the engine or the config again.
|
||||
# must not reach the engine or the config again. A key
|
||||
# counts as touched only when the newer apply wrote
|
||||
# it: its choice can legally equal the baseline, and
|
||||
# the older edit must not overwrite it.
|
||||
surviving = {
|
||||
key: value
|
||||
for key, value in pending.items()
|
||||
if self.config_manager.get("advanced", key) == baseline.get(key)
|
||||
if _apply_settings_written.get(key, self._pending_apply_generation)
|
||||
<= self._pending_apply_generation
|
||||
and self.config_manager.get("advanced", key) == baseline.get(key)
|
||||
}
|
||||
else:
|
||||
surviving = dict(pending)
|
||||
@@ -6494,6 +6553,7 @@ class SettingsDialog(Gtk.Dialog):
|
||||
self.speech_engine.reconfigure(**surviving)
|
||||
for key, value in surviving.items():
|
||||
self.config_manager.set("advanced", key, value)
|
||||
_apply_settings_written[key] = _apply_settings_generation
|
||||
self.config_manager.save_settings()
|
||||
except (OSError, ValueError, TypeError, RuntimeError) as e:
|
||||
logger.warning(
|
||||
@@ -6573,6 +6633,7 @@ class SettingsDialog(Gtk.Dialog):
|
||||
show_missing_tray_warning = ui_settings.get("show_missing_tray_warning", True)
|
||||
show_overlay = ui_settings.get("show_overlay", True)
|
||||
copy_to_clipboard = text_injection_settings.get("copy_to_clipboard", False)
|
||||
dictate_to_pad = text_injection_settings.get("dictate_to_pad", False)
|
||||
auto_capitalize = text_injection_settings.get("auto_capitalize", True)
|
||||
append_trailing_space = text_injection_settings.get("append_trailing_space", True)
|
||||
paste_shortcut = self.config_manager.get_paste_shortcut()
|
||||
@@ -6586,6 +6647,7 @@ class SettingsDialog(Gtk.Dialog):
|
||||
self.missing_tray_warning_switch.set_active(show_missing_tray_warning)
|
||||
self.show_overlay_switch.set_active(show_overlay)
|
||||
self.copy_to_clipboard_switch.set_active(copy_to_clipboard)
|
||||
self.dictation_pad_switch.set_active(dictate_to_pad)
|
||||
self.auto_capitalize_switch.set_active(auto_capitalize)
|
||||
self.append_trailing_space_switch.set_active(append_trailing_space)
|
||||
if not self.paste_shortcut_combo.set_active_id(paste_shortcut):
|
||||
@@ -8101,6 +8163,7 @@ class SettingsDialog(Gtk.Dialog):
|
||||
# generation that already includes this in-flight apply.
|
||||
global _apply_settings_generation
|
||||
_apply_settings_generation += 1
|
||||
apply_generation = _apply_settings_generation
|
||||
self._applying_settings = True
|
||||
# Already-downloaded apply is handed to a worker that clears this flag
|
||||
# via GLib.idle_add. Download still holds it for the modal run() below.
|
||||
@@ -8203,7 +8266,9 @@ class SettingsDialog(Gtk.Dialog):
|
||||
|
||||
def apply_already_downloaded() -> None:
|
||||
try:
|
||||
self._apply_settings_internal(settings, raise_errors=True)
|
||||
self._apply_settings_internal(
|
||||
settings, raise_errors=True, apply_generation=apply_generation
|
||||
)
|
||||
logger.info("Settings auto-applied successfully")
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to auto-apply settings: {e}")
|
||||
@@ -8438,13 +8503,15 @@ class SettingsDialog(Gtk.Dialog):
|
||||
self._saved_text_callbacks = self.speech_engine.get_text_callbacks()
|
||||
self.speech_engine.set_text_callbacks([self._test_text_callback])
|
||||
# History recording runs on segment callbacks keyed by capture time:
|
||||
# stamp the floor so test speech stays out of Recent Snippets while
|
||||
# stamp the window so test speech stays out of Recent Snippets while
|
||||
# leftover segments from a dictation still file normally.
|
||||
self.speech_engine.test_capture_ceiling = None
|
||||
self.speech_engine.test_capture_floor = time.monotonic()
|
||||
|
||||
if not self.speech_engine.start_recognition():
|
||||
self.speech_engine.set_text_callbacks(self._saved_text_callbacks)
|
||||
self.speech_engine.test_capture_floor = None
|
||||
self.speech_engine.test_capture_ceiling = None
|
||||
del self._saved_text_callbacks
|
||||
self.test_output_revealer.set_reveal_child(True)
|
||||
if getattr(self.speech_engine, "is_auto_paused", False):
|
||||
@@ -8650,7 +8717,10 @@ class SettingsDialog(Gtk.Dialog):
|
||||
if hasattr(self, "_saved_text_callbacks"):
|
||||
self.speech_engine.set_text_callbacks(self._saved_text_callbacks)
|
||||
del self._saved_text_callbacks
|
||||
self.speech_engine.test_capture_floor = None
|
||||
# Close the test-capture window rather than reopening history for
|
||||
# everything after the floor: segments captured during the test but
|
||||
# still decoding must keep failing the window check.
|
||||
self.speech_engine.test_capture_ceiling = time.monotonic()
|
||||
|
||||
# Check result after giving time for final callbacks to complete
|
||||
GLib.timeout_add(300, self._check_test_result)
|
||||
@@ -8802,7 +8872,11 @@ For now, the engine has been reverted to VOSK."""
|
||||
dialog.destroy()
|
||||
|
||||
def _apply_settings_internal(
|
||||
self, settings: dict, raise_errors: bool = False, force_reinit: bool = False
|
||||
self,
|
||||
settings: dict,
|
||||
raise_errors: bool = False,
|
||||
force_reinit: bool = False,
|
||||
apply_generation: Optional[int] = None,
|
||||
) -> bool:
|
||||
"""Internal method to apply settings.
|
||||
|
||||
@@ -8816,20 +8890,37 @@ For now, the engine has been reverted to VOSK."""
|
||||
matches its live state. The download threads pass True: they
|
||||
only run because the model is missing on disk, and a no-op
|
||||
reconfigure would report success for a download that never ran.
|
||||
apply_generation: The apply generation captured when this snapshot
|
||||
was collected. When a newer apply began since, this snapshot is
|
||||
stale and its write is skipped — otherwise lock order, not
|
||||
collection order, would decide which snapshot wins.
|
||||
"""
|
||||
try:
|
||||
was_running = self.speech_engine.state != RecognitionState.IDLE
|
||||
if was_running:
|
||||
self.speech_engine.stop_recognition()
|
||||
time.sleep(0.5)
|
||||
|
||||
# Persist only once the engine really runs these settings: this call
|
||||
# downloads missing models, and a config saved up front would keep
|
||||
# pointing at a model that never made it to disk. The lock orders
|
||||
# this snapshot against _persist_pending_text_edits' deferred one.
|
||||
# this snapshot against _persist_pending_text_edits' deferred one —
|
||||
# and the staleness check must precede stop_recognition, or a
|
||||
# superseded apply would still interrupt a live dictation session.
|
||||
with _apply_settings_lock:
|
||||
if apply_generation is not None and _apply_settings_generation != apply_generation:
|
||||
logger.info(
|
||||
"Settings apply superseded by a newer one; " "skipping the stale snapshot"
|
||||
)
|
||||
return True
|
||||
was_running = self.speech_engine.state != RecognitionState.IDLE
|
||||
if was_running:
|
||||
self.speech_engine.stop_recognition()
|
||||
time.sleep(0.5)
|
||||
|
||||
self.speech_engine.reconfigure(force_reinit=force_reinit, **settings)
|
||||
self._save_selected_settings(settings)
|
||||
written_gen = (
|
||||
apply_generation if apply_generation is not None else _apply_settings_generation
|
||||
)
|
||||
for key in settings:
|
||||
if key.startswith("whispercpp_"):
|
||||
_apply_settings_written[key] = written_gen
|
||||
|
||||
logger.info("Settings applied successfully.")
|
||||
return True
|
||||
|
||||
@@ -58,9 +58,12 @@ class TranscriptionHistory:
|
||||
def __init__(self, max_items: int = DEFAULT_MAX_ITEMS, enabled: bool = True) -> None:
|
||||
self._max_items = sanitize_max_items(max_items)
|
||||
self._enabled = bool(enabled)
|
||||
self._entries: deque = deque(maxlen=self._max_items)
|
||||
self._entries: deque[tuple[int, str]] = deque(maxlen=self._max_items)
|
||||
self._lock = threading.Lock()
|
||||
self._change_callback: Optional[Callable[[], None]] = None
|
||||
# Monotonic id stamped on each snippet so late segments can extend
|
||||
# the session they belong to even after newer snippets arrive.
|
||||
self._next_id = 0
|
||||
# Bumped every time the entries are wiped; lets callers refuse text
|
||||
# produced before a clear so it cannot reappear afterwards.
|
||||
self._epoch = 0
|
||||
@@ -130,8 +133,8 @@ class TranscriptionHistory:
|
||||
self._cleared_at = time.monotonic()
|
||||
self._notify()
|
||||
|
||||
def add(self, text: str, *, expected_epoch: Optional[int] = None) -> bool:
|
||||
"""Add a snippet, returning True when it was recorded.
|
||||
def add(self, text: str, *, expected_epoch: Optional[int] = None) -> Optional[int]:
|
||||
"""Add a snippet, returning its id when recorded (``None`` on refusal).
|
||||
|
||||
No-op when disabled or when text is empty. With ``expected_epoch``
|
||||
the add is also refused once the epoch has advanced — i.e. the
|
||||
@@ -139,16 +142,42 @@ class TranscriptionHistory:
|
||||
produced before the clear cannot reappear as a new entry.
|
||||
"""
|
||||
if not text:
|
||||
return False
|
||||
return None
|
||||
text = text.strip()
|
||||
if not text:
|
||||
return False
|
||||
return None
|
||||
with self._lock:
|
||||
if not self._enabled:
|
||||
return None
|
||||
if expected_epoch is not None and expected_epoch != self._epoch:
|
||||
return None
|
||||
self._next_id += 1
|
||||
self._entries.append((self._next_id, text))
|
||||
self._notify()
|
||||
return self._next_id
|
||||
|
||||
def extend_entry(
|
||||
self, snippet_id: int, text: str, *, expected_epoch: Optional[int] = None
|
||||
) -> bool:
|
||||
"""Append a late-arriving segment to the snippet with ``snippet_id``.
|
||||
|
||||
Unlike ``extend_latest`` the target is the entry's own id, so a
|
||||
straggler still lands in the session that produced it even when a
|
||||
newer session has since committed its snippet on top.
|
||||
"""
|
||||
if not text or not text.strip():
|
||||
return False
|
||||
with self._lock:
|
||||
if not self._enabled or not self._entries:
|
||||
return False
|
||||
if expected_epoch is not None and expected_epoch != self._epoch:
|
||||
return False
|
||||
self._entries.append(text)
|
||||
for index, (entry_id, existing) in enumerate(self._entries):
|
||||
if entry_id == snippet_id:
|
||||
self._entries[index] = (entry_id, f"{existing} {text.strip()}")
|
||||
break
|
||||
else:
|
||||
return False
|
||||
self._notify()
|
||||
return True
|
||||
|
||||
@@ -173,14 +202,15 @@ class TranscriptionHistory:
|
||||
return False
|
||||
if expected_epoch is not None and expected_epoch != self._epoch:
|
||||
return False
|
||||
self._entries[-1] = f"{self._entries[-1]} {text.strip()}"
|
||||
entry_id, existing = self._entries[-1]
|
||||
self._entries[-1] = (entry_id, f"{existing} {text.strip()}")
|
||||
self._notify()
|
||||
return True
|
||||
|
||||
def get_all(self) -> List[str]:
|
||||
"""Return all snippets, newest first."""
|
||||
with self._lock:
|
||||
return list(reversed(self._entries))
|
||||
return [text for _, text in reversed(self._entries)]
|
||||
|
||||
def clear(self) -> None:
|
||||
"""Remove all snippets.
|
||||
|
||||
@@ -149,6 +149,7 @@ class TrayIndicator:
|
||||
speech_engine: SpeechRecognitionManagerProtocol,
|
||||
text_injector: TextInjectorProtocol,
|
||||
transcription_history: Optional[TranscriptionHistory] = None,
|
||||
dictation_pad: Optional[Any] = None,
|
||||
) -> None:
|
||||
"""
|
||||
Initialize the system tray indicator.
|
||||
@@ -158,10 +159,13 @@ class TrayIndicator:
|
||||
text_injector: The text injector instance
|
||||
transcription_history: Optional in-memory store of recent dictation
|
||||
snippets. When provided, a "Recent Snippets" submenu is shown.
|
||||
dictation_pad: Optional in-app Dictation Pad window the tray menu
|
||||
can open (the Wayland-safe dictation fallback, #726)
|
||||
"""
|
||||
self.speech_engine = speech_engine
|
||||
self.text_injector = text_injector
|
||||
self.transcription_history = transcription_history
|
||||
self.dictation_pad = dictation_pad
|
||||
# Shared with main() and the settings dialog: separate instances would
|
||||
# overwrite each other's saves with stale in-memory copies.
|
||||
self.config_manager = get_shared_config_manager()
|
||||
@@ -190,6 +194,9 @@ class TrayIndicator:
|
||||
# One listener per bound language shortcut (#805); rebuilt by
|
||||
# _setup_language_shortcuts alongside the main one.
|
||||
self._language_shortcut_managers: list[KeyboardShortcutManager] = []
|
||||
# A settings edit landed while a dictation session was live; the
|
||||
# listener rebuild is deferred to the next IDLE transition (#805).
|
||||
self._language_shortcuts_refresh_pending = False
|
||||
# The binding whose press started the live push-to-talk session, so an
|
||||
# unrelated key's release cannot end it (#805).
|
||||
self._ptt_owner: Optional[KeyboardShortcutManager] = None
|
||||
@@ -466,12 +473,20 @@ class TrayIndicator:
|
||||
self._language_shortcut_managers = []
|
||||
|
||||
def refresh_language_shortcuts(self) -> None:
|
||||
"""Rebuild the per-language listeners after a settings change (#805)."""
|
||||
"""Rebuild the per-language listeners after a settings change (#805).
|
||||
|
||||
While a dictation session is live the rebuild is deferred to the
|
||||
next IDLE transition instead: rebuilding stops recognition first,
|
||||
and a keystroke-level settings edit must never cut an utterance off.
|
||||
"""
|
||||
if self._external_activation_active():
|
||||
# The internal listener family is off; a binding edit must not
|
||||
# arm language listeners behind the external-activation setting.
|
||||
self._stop_language_shortcut_managers()
|
||||
return
|
||||
if self.speech_engine.state != RecognitionState.IDLE:
|
||||
self._language_shortcuts_refresh_pending = True
|
||||
return
|
||||
self._setup_language_shortcuts()
|
||||
|
||||
def _init_icons(self):
|
||||
@@ -570,6 +585,8 @@ class TrayIndicator:
|
||||
self._update_autostart_checkbox()
|
||||
|
||||
self._add_menu_separator()
|
||||
if self.dictation_pad is not None:
|
||||
self._add_menu_item("Dictation Pad", self._on_dictation_pad_clicked)
|
||||
self._add_menu_item("Settings", self._on_settings_clicked)
|
||||
self._add_menu_item("View Logs", self._on_logs_clicked)
|
||||
self._gateway_stop_menu_item = self._add_menu_item(
|
||||
@@ -1198,13 +1215,20 @@ class TrayIndicator:
|
||||
state wins: GLib.idle_add can deliver a captured PROCESSING
|
||||
argument after stop already went IDLE (#739).
|
||||
"""
|
||||
if not hasattr(self, "indicator"):
|
||||
return False
|
||||
|
||||
engine_state = getattr(self.speech_engine, "state", None)
|
||||
if isinstance(engine_state, RecognitionState):
|
||||
state = engine_state
|
||||
|
||||
if state == RecognitionState.IDLE and getattr(
|
||||
self, "_language_shortcuts_refresh_pending", False
|
||||
):
|
||||
# The session the listener rebuild was deferred for has ended.
|
||||
self._language_shortcuts_refresh_pending = False
|
||||
self.refresh_language_shortcuts()
|
||||
|
||||
if not hasattr(self, "indicator"):
|
||||
return False
|
||||
|
||||
if state == RecognitionState.IDLE:
|
||||
self._set_indicator_icon(self._icon_keys["default"], "Microphone off")
|
||||
self._set_menu_item_enabled("Start Voice Typing", True)
|
||||
@@ -1325,6 +1349,12 @@ class TrayIndicator:
|
||||
if self.transcription_history is not None:
|
||||
self.transcription_history.clear()
|
||||
|
||||
def _on_dictation_pad_clicked(self, widget: Gtk.MenuItem) -> None:
|
||||
"""Handle click on the Dictation Pad menu item."""
|
||||
logger.debug("Dictation Pad clicked")
|
||||
if self.dictation_pad is not None:
|
||||
self.dictation_pad.show_pad()
|
||||
|
||||
def _on_settings_clicked(self, widget):
|
||||
"""Handle click on the Settings menu item."""
|
||||
logger.debug("Settings clicked")
|
||||
|
||||
@@ -0,0 +1,806 @@
|
||||
"""
|
||||
Tests for the in-app dictation pad (issue #726).
|
||||
|
||||
Drives the shipped DictationPadController (and config helpers) so capture
|
||||
mode routes text into the pad's buffer instead of cross-app injection.
|
||||
|
||||
Important: these tests must not import real GTK / tray_indicator into
|
||||
sys.modules — that breaks later tests that mock gi (see CI isolation).
|
||||
"""
|
||||
|
||||
import os
|
||||
import tempfile
|
||||
import unittest
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
from vocalinux.ui.config_manager import DEFAULT_CONFIG, ConfigManager
|
||||
from vocalinux.ui.dictation_pad import DictationPad, DictationPadController
|
||||
|
||||
|
||||
def _ensure_test_config_dir(path: str) -> None:
|
||||
parent_dir = os.path.dirname(path)
|
||||
if not os.path.exists(parent_dir):
|
||||
os.mkdir(parent_dir)
|
||||
if not os.path.exists(path):
|
||||
os.mkdir(path)
|
||||
|
||||
|
||||
def _pad_without_gtk(enabled: bool = False) -> DictationPad:
|
||||
"""
|
||||
Build DictationPad without initializing a real GTK window.
|
||||
|
||||
Patches window construction so tests never load gi.repository (which
|
||||
would pollute the process for later gi-mocked tests).
|
||||
"""
|
||||
with patch.object(
|
||||
DictationPad,
|
||||
"_init_gtk_window",
|
||||
side_effect=RuntimeError("gtk disabled in unit tests"),
|
||||
):
|
||||
return DictationPad(enabled=enabled)
|
||||
|
||||
|
||||
class TestDictationPadController(unittest.TestCase):
|
||||
"""Unit tests for the pure buffer controller."""
|
||||
|
||||
def test_default_disabled_and_empty(self) -> None:
|
||||
ctrl = DictationPadController()
|
||||
self.assertFalse(ctrl.enabled)
|
||||
self.assertEqual(ctrl.text, "")
|
||||
|
||||
def test_append_accumulates_segments(self) -> None:
|
||||
ctrl = DictationPadController(enabled=True)
|
||||
ctrl.append("Hello ")
|
||||
ctrl.append("world.")
|
||||
self.assertEqual(ctrl.text, "Hello world.")
|
||||
|
||||
def test_append_newlines_preserved(self) -> None:
|
||||
ctrl = DictationPadController()
|
||||
ctrl.append("first line")
|
||||
ctrl.append("\n")
|
||||
ctrl.append("second")
|
||||
self.assertEqual(ctrl.text, "first line\nsecond")
|
||||
|
||||
def test_delete_last_removes_tail_chars(self) -> None:
|
||||
ctrl = DictationPadController()
|
||||
ctrl.append("Hello world")
|
||||
self.assertEqual(ctrl.delete_last(6), 6)
|
||||
self.assertEqual(ctrl.text, "Hello")
|
||||
|
||||
def test_delete_last_clamps_to_buffer_length(self) -> None:
|
||||
ctrl = DictationPadController()
|
||||
ctrl.append("hi")
|
||||
self.assertEqual(ctrl.delete_last(10), 2)
|
||||
self.assertEqual(ctrl.text, "")
|
||||
|
||||
def test_delete_last_on_empty_returns_zero(self) -> None:
|
||||
ctrl = DictationPadController()
|
||||
self.assertEqual(ctrl.delete_last(5), 0)
|
||||
ctrl.append("x")
|
||||
self.assertEqual(ctrl.delete_last(0), 0)
|
||||
self.assertEqual(ctrl.delete_last(-3), 0)
|
||||
self.assertEqual(ctrl.text, "x")
|
||||
|
||||
def test_clear_empties_buffer(self) -> None:
|
||||
ctrl = DictationPadController()
|
||||
ctrl.append("some dictation")
|
||||
ctrl.clear()
|
||||
self.assertEqual(ctrl.text, "")
|
||||
|
||||
def test_set_text_replaces_buffer(self) -> None:
|
||||
ctrl = DictationPadController()
|
||||
ctrl.append("dictated")
|
||||
ctrl.set_text("dictated plus edits")
|
||||
self.assertEqual(ctrl.text, "dictated plus edits")
|
||||
# Identical content is a no-op, not a mutation: the first undo goes
|
||||
# straight back to the pre-edit text.
|
||||
ctrl.set_text("dictated plus edits")
|
||||
self.assertTrue(ctrl.undo())
|
||||
self.assertEqual(ctrl.text, "dictated")
|
||||
|
||||
def test_undo_restores_last_mutation_and_redo_replays(self) -> None:
|
||||
ctrl = DictationPadController()
|
||||
ctrl.append("one ")
|
||||
ctrl.append("two ")
|
||||
self.assertTrue(ctrl.undo())
|
||||
self.assertEqual(ctrl.text, "one ")
|
||||
self.assertTrue(ctrl.undo())
|
||||
self.assertEqual(ctrl.text, "")
|
||||
self.assertFalse(ctrl.undo())
|
||||
self.assertTrue(ctrl.redo())
|
||||
self.assertEqual(ctrl.text, "one ")
|
||||
self.assertTrue(ctrl.redo())
|
||||
self.assertEqual(ctrl.text, "one two ")
|
||||
self.assertFalse(ctrl.redo())
|
||||
|
||||
def test_new_edit_clears_redo_lane(self) -> None:
|
||||
ctrl = DictationPadController()
|
||||
ctrl.append("a")
|
||||
ctrl.undo()
|
||||
ctrl.append("b")
|
||||
self.assertFalse(ctrl.redo())
|
||||
|
||||
def test_delete_last_and_clear_are_undoable(self) -> None:
|
||||
ctrl = DictationPadController()
|
||||
ctrl.append("hello world")
|
||||
ctrl.delete_last(6)
|
||||
self.assertEqual(ctrl.text, "hello")
|
||||
self.assertTrue(ctrl.undo())
|
||||
self.assertEqual(ctrl.text, "hello world")
|
||||
ctrl.clear()
|
||||
self.assertEqual(ctrl.text, "")
|
||||
self.assertTrue(ctrl.undo())
|
||||
self.assertEqual(ctrl.text, "hello world")
|
||||
|
||||
def test_set_enabled_does_not_touch_buffer(self) -> None:
|
||||
ctrl = DictationPadController()
|
||||
ctrl.append("keep me")
|
||||
ctrl.set_enabled(True)
|
||||
ctrl.set_enabled(False)
|
||||
self.assertEqual(ctrl.text, "keep me")
|
||||
|
||||
def test_last_segment_tracks_appends(self) -> None:
|
||||
ctrl = DictationPadController()
|
||||
self.assertIsNone(ctrl.last_segment)
|
||||
ctrl.append("hello ")
|
||||
ctrl.append("world ")
|
||||
self.assertEqual(ctrl.last_segment, "world ")
|
||||
|
||||
def test_last_segment_tracks_undo_and_redo(self) -> None:
|
||||
"""Undo pops the deletion target with the text it removed."""
|
||||
ctrl = DictationPadController()
|
||||
ctrl.append("hello ")
|
||||
ctrl.append("world ")
|
||||
self.assertTrue(ctrl.undo())
|
||||
self.assertEqual(ctrl.last_segment, "hello ")
|
||||
self.assertTrue(ctrl.undo())
|
||||
self.assertIsNone(ctrl.last_segment)
|
||||
self.assertTrue(ctrl.redo())
|
||||
self.assertEqual(ctrl.last_segment, "hello ")
|
||||
self.assertTrue(ctrl.redo())
|
||||
self.assertEqual(ctrl.last_segment, "world ")
|
||||
|
||||
def test_last_segment_survives_delete_and_drops_on_set_text(self) -> None:
|
||||
ctrl = DictationPadController()
|
||||
ctrl.append("hello ")
|
||||
ctrl.append("world ")
|
||||
ctrl.delete_last(2)
|
||||
self.assertEqual(ctrl.last_segment, "worl")
|
||||
ctrl.delete_last(10)
|
||||
self.assertIsNone(ctrl.last_segment)
|
||||
ctrl.append("again ")
|
||||
ctrl.set_text("manual rewrite")
|
||||
self.assertIsNone(ctrl.last_segment)
|
||||
# Undo restores the boundaries the manual rewrite blurred.
|
||||
self.assertTrue(ctrl.undo())
|
||||
self.assertEqual(ctrl.last_segment, "again ")
|
||||
|
||||
def test_clear_drops_segment_tracking(self) -> None:
|
||||
ctrl = DictationPadController()
|
||||
ctrl.append("gone ")
|
||||
ctrl.clear()
|
||||
self.assertIsNone(ctrl.last_segment)
|
||||
|
||||
|
||||
class TestDictationPadConfig(unittest.TestCase):
|
||||
"""ConfigManager helpers for the dictate_to_pad preference."""
|
||||
|
||||
def setUp(self) -> None:
|
||||
self.temp_dir = tempfile.TemporaryDirectory()
|
||||
self.temp_config_dir = os.path.join(self.temp_dir.name, ".config/vocalinux")
|
||||
_ensure_test_config_dir(self.temp_config_dir)
|
||||
self.temp_config_file = os.path.join(self.temp_config_dir, "config.json")
|
||||
|
||||
self.config_dir_patcher = patch(
|
||||
"vocalinux.ui.config_manager.CONFIG_DIR", self.temp_config_dir
|
||||
)
|
||||
self.config_file_patcher = patch(
|
||||
"vocalinux.ui.config_manager.CONFIG_FILE", self.temp_config_file
|
||||
)
|
||||
self.makedirs_patcher = patch(
|
||||
"vocalinux.ui.config_manager.os.makedirs",
|
||||
side_effect=lambda path, exist_ok=True: _ensure_test_config_dir(path),
|
||||
)
|
||||
self.config_dir_patcher.start()
|
||||
self.config_file_patcher.start()
|
||||
self.makedirs_patcher.start()
|
||||
_ensure_test_config_dir(self.temp_config_dir)
|
||||
|
||||
def tearDown(self) -> None:
|
||||
self.config_dir_patcher.stop()
|
||||
self.config_file_patcher.stop()
|
||||
self.makedirs_patcher.stop()
|
||||
self.temp_dir.cleanup()
|
||||
|
||||
def test_default_dictate_to_pad_disabled(self) -> None:
|
||||
self.assertFalse(DEFAULT_CONFIG["text_injection"]["dictate_to_pad"])
|
||||
cm = ConfigManager()
|
||||
self.assertFalse(cm.is_dictate_to_pad_enabled())
|
||||
|
||||
def test_set_dictate_to_pad_persists(self) -> None:
|
||||
cm = ConfigManager()
|
||||
cm.set_dictate_to_pad(True)
|
||||
self.assertTrue(cm.is_dictate_to_pad_enabled())
|
||||
cm.save_config()
|
||||
|
||||
cm2 = ConfigManager()
|
||||
self.assertTrue(cm2.is_dictate_to_pad_enabled())
|
||||
|
||||
def test_set_dictate_to_pad_false(self) -> None:
|
||||
cm = ConfigManager()
|
||||
cm.set_dictate_to_pad(True)
|
||||
cm.set_dictate_to_pad(False)
|
||||
self.assertFalse(cm.is_dictate_to_pad_enabled())
|
||||
|
||||
def test_corrupt_config_file_keeps_in_memory_capture_state(self) -> None:
|
||||
"""A torn config.json write must not flip a running session's routing.
|
||||
|
||||
The live toggle is read from the shared manager's in-memory cache,
|
||||
never re-parsed per segment: a partial save on disk cannot misroute
|
||||
dictation back into whichever application holds focus.
|
||||
"""
|
||||
cm = ConfigManager()
|
||||
cm.set_dictate_to_pad(True)
|
||||
# Simulate a torn mid-write file.
|
||||
with open(self.temp_config_file, "w") as f:
|
||||
f.write('{"text_injection": {"dictate_to_pad": tr')
|
||||
self.assertTrue(cm.is_dictate_to_pad_enabled())
|
||||
# A fresh manager loading the torn file fails closed to the default.
|
||||
self.assertFalse(ConfigManager().is_dictate_to_pad_enabled())
|
||||
|
||||
|
||||
class TestDictationPadFacade(unittest.TestCase):
|
||||
"""
|
||||
Drive the shipped DictationPad API without a display.
|
||||
|
||||
Window construction is patched out so these tests never load GTK; the
|
||||
controller must still capture text so nothing is silently dropped.
|
||||
"""
|
||||
|
||||
def test_gtk_init_failure_keeps_buffer_usable(self) -> None:
|
||||
pad = _pad_without_gtk(enabled=True)
|
||||
self.assertFalse(pad._gtk_ready)
|
||||
pad.append_text("dictated words ")
|
||||
pad.append_text("more")
|
||||
self.assertEqual(pad.controller.text, "dictated words more")
|
||||
pad.destroy()
|
||||
|
||||
def test_delete_last_chars_headless(self) -> None:
|
||||
pad = _pad_without_gtk()
|
||||
try:
|
||||
pad.append_text("hello ")
|
||||
self.assertEqual(pad.delete_last_chars(6), 6)
|
||||
self.assertEqual(pad.controller.text, "")
|
||||
finally:
|
||||
pad.destroy()
|
||||
|
||||
def test_show_pad_is_safe_headless(self) -> None:
|
||||
pad = _pad_without_gtk()
|
||||
try:
|
||||
pad.show_pad() # must not raise
|
||||
finally:
|
||||
pad.destroy()
|
||||
|
||||
def test_set_capture_enabled_updates_controller(self) -> None:
|
||||
pad = _pad_without_gtk()
|
||||
try:
|
||||
pad.set_capture_enabled(True)
|
||||
self.assertTrue(pad.controller.enabled)
|
||||
pad.set_capture_enabled(False)
|
||||
self.assertFalse(pad.controller.enabled)
|
||||
finally:
|
||||
pad.destroy()
|
||||
|
||||
def test_capture_checkbox_persists_config(self) -> None:
|
||||
"""The pad checkbox writes dictate_to_pad via the config manager."""
|
||||
config_manager = MagicMock()
|
||||
with patch.object(
|
||||
DictationPad,
|
||||
"_init_gtk_window",
|
||||
side_effect=RuntimeError("gtk disabled in unit tests"),
|
||||
):
|
||||
pad = DictationPad(enabled=False, config_manager=config_manager)
|
||||
try:
|
||||
widget = MagicMock()
|
||||
widget.get_active.return_value = True
|
||||
pad._on_capture_toggled(widget)
|
||||
config_manager.set.assert_called_once_with("text_injection", "dictate_to_pad", True)
|
||||
config_manager.save_settings.assert_called_once()
|
||||
self.assertTrue(pad.controller.enabled)
|
||||
finally:
|
||||
pad.destroy()
|
||||
|
||||
def test_apply_append_inserts_at_end_and_shows(self) -> None:
|
||||
"""Widget path: append inserts at buffer end; enabled capture shows."""
|
||||
pad = _pad_without_gtk(enabled=True)
|
||||
try:
|
||||
pad._gtk_ready = True
|
||||
pad._window = MagicMock()
|
||||
pad._window.get_visible.return_value = False
|
||||
pad._textview = MagicMock()
|
||||
pad._buffer = MagicMock()
|
||||
pad._Gtk = MagicMock()
|
||||
|
||||
pad.controller.append("segment ")
|
||||
pad._apply_append("segment ", pad._generation)
|
||||
|
||||
pad._buffer.insert.assert_called_once()
|
||||
args = pad._buffer.insert.call_args[0]
|
||||
self.assertEqual(args[1], "segment ")
|
||||
pad._window.show_all.assert_called_once()
|
||||
finally:
|
||||
pad._window = None
|
||||
pad.destroy()
|
||||
|
||||
def test_apply_append_does_not_show_when_disabled(self) -> None:
|
||||
pad = _pad_without_gtk(enabled=False)
|
||||
try:
|
||||
pad._gtk_ready = True
|
||||
pad._window = MagicMock()
|
||||
pad._window.get_visible.return_value = False
|
||||
pad._textview = MagicMock()
|
||||
pad._buffer = MagicMock()
|
||||
|
||||
pad.controller.append("segment ")
|
||||
pad._apply_append("segment ", pad._generation)
|
||||
|
||||
pad._window.show_all.assert_not_called()
|
||||
finally:
|
||||
pad._window = None
|
||||
pad.destroy()
|
||||
|
||||
def test_apply_append_reveals_when_enabled_via_settings(self) -> None:
|
||||
"""Regression: capture toggled in Settings must reveal the pad too.
|
||||
|
||||
The reveal gate consults live config, not just the controller flag,
|
||||
so a dictate_to_pad=True written by the Settings switch while the
|
||||
pad is closed still surfaces the window on the first segment.
|
||||
"""
|
||||
config_manager = MagicMock()
|
||||
config_manager.get_bool.return_value = True
|
||||
with patch.object(
|
||||
DictationPad,
|
||||
"_init_gtk_window",
|
||||
side_effect=RuntimeError("gtk disabled in unit tests"),
|
||||
):
|
||||
pad = DictationPad(enabled=False, config_manager=config_manager)
|
||||
try:
|
||||
pad._gtk_ready = True
|
||||
pad._window = MagicMock()
|
||||
pad._window.get_visible.return_value = False
|
||||
pad._textview = MagicMock()
|
||||
pad._buffer = MagicMock()
|
||||
pad._Gtk = MagicMock()
|
||||
|
||||
pad.controller.append("segment ")
|
||||
pad._apply_append("segment ", pad._generation)
|
||||
|
||||
config_manager.get_bool.assert_called_with("text_injection", "dictate_to_pad", False)
|
||||
self.assertTrue(pad.controller.enabled)
|
||||
pad._window.show_all.assert_called_once()
|
||||
finally:
|
||||
pad._window = None
|
||||
pad.destroy()
|
||||
|
||||
def test_apply_append_stays_hidden_when_settings_off(self) -> None:
|
||||
"""A stale controller flag must not override live config."""
|
||||
config_manager = MagicMock()
|
||||
config_manager.get_bool.return_value = False
|
||||
with patch.object(
|
||||
DictationPad,
|
||||
"_init_gtk_window",
|
||||
side_effect=RuntimeError("gtk disabled in unit tests"),
|
||||
):
|
||||
pad = DictationPad(enabled=True, config_manager=config_manager)
|
||||
try:
|
||||
pad._gtk_ready = True
|
||||
pad._window = MagicMock()
|
||||
pad._window.get_visible.return_value = False
|
||||
pad._textview = MagicMock()
|
||||
pad._buffer = MagicMock()
|
||||
|
||||
pad.controller.append("segment ")
|
||||
pad._apply_append("segment ", pad._generation)
|
||||
|
||||
pad._window.show_all.assert_not_called()
|
||||
self.assertFalse(pad.controller.enabled)
|
||||
finally:
|
||||
pad._window = None
|
||||
pad.destroy()
|
||||
|
||||
def test_copy_all_prefers_widget_text(self) -> None:
|
||||
"""Copy All copies the widget contents (dictation + in-pad edits)."""
|
||||
pad = _pad_without_gtk()
|
||||
try:
|
||||
pad._gtk_ready = True
|
||||
pad._Gtk = MagicMock()
|
||||
pad._Gdk = MagicMock()
|
||||
pad._GLib = MagicMock()
|
||||
pad._copy_button = MagicMock()
|
||||
pad._buffer = MagicMock()
|
||||
pad._buffer.get_text.return_value = "dictated plus edits"
|
||||
pad.controller.append("dictated")
|
||||
|
||||
pad._on_copy_all_clicked()
|
||||
|
||||
clipboard = pad._Gtk.Clipboard.get.return_value
|
||||
clipboard.set_text.assert_called_once_with("dictated plus edits", -1)
|
||||
finally:
|
||||
pad.destroy()
|
||||
|
||||
def test_show_pad_does_not_clobber_widget_edits(self) -> None:
|
||||
"""Re-showing the pad must not overwrite in-pad manual edits."""
|
||||
pad = _pad_without_gtk()
|
||||
try:
|
||||
pad._gtk_ready = True
|
||||
pad._window = MagicMock()
|
||||
pad._window.get_visible.return_value = True
|
||||
pad._buffer = MagicMock()
|
||||
pad._Gtk = MagicMock()
|
||||
|
||||
pad.show_pad()
|
||||
|
||||
# Never a wholesale set_text on (re)show: widget content is kept
|
||||
# incrementally in sync by _apply_append/_apply_delete.
|
||||
pad._buffer.set_text.assert_not_called()
|
||||
finally:
|
||||
pad._window = None
|
||||
pad.destroy()
|
||||
|
||||
def test_apply_delete_removes_tail_from_widget(self) -> None:
|
||||
pad = _pad_without_gtk()
|
||||
try:
|
||||
pad._gtk_ready = True
|
||||
pad._textview = MagicMock()
|
||||
pad._buffer = MagicMock()
|
||||
end_iter = MagicMock()
|
||||
pad._buffer.get_end_iter.return_value = end_iter
|
||||
|
||||
pad.controller.append("hello")
|
||||
pad._apply_delete(5, pad._generation)
|
||||
|
||||
pad._buffer.delete.assert_called_once()
|
||||
start_iter = pad._buffer.delete.call_args[0][0]
|
||||
start_iter.backward_chars.assert_called_once_with(5)
|
||||
finally:
|
||||
pad.destroy()
|
||||
|
||||
def test_clear_drops_appends_still_queued_for_widget(self) -> None:
|
||||
"""A Clear between append_text and its idle callback must win."""
|
||||
pad = _pad_without_gtk(enabled=True)
|
||||
try:
|
||||
pad._gtk_ready = True
|
||||
pad._GLib = MagicMock()
|
||||
pad._buffer = MagicMock()
|
||||
pad._textview = MagicMock()
|
||||
pad._window = MagicMock()
|
||||
pad._window.get_visible.return_value = True
|
||||
|
||||
pad.append_text("stale segment ")
|
||||
pad._on_clear_clicked()
|
||||
self.assertEqual(pad.controller.text, "")
|
||||
|
||||
# Run the queued idle callback — the stale generation skips it.
|
||||
queued = pad._GLib.idle_add.call_args.args[0]
|
||||
queued()
|
||||
pad._buffer.insert.assert_not_called()
|
||||
pad._buffer.set_text.assert_called_once_with("")
|
||||
finally:
|
||||
pad._window = None
|
||||
pad.destroy()
|
||||
|
||||
def test_queued_append_before_clear_still_applies(self) -> None:
|
||||
"""Appends queued before the bump run normally; only stale ones die."""
|
||||
pad = _pad_without_gtk(enabled=True)
|
||||
try:
|
||||
pad._gtk_ready = True
|
||||
pad._GLib = MagicMock()
|
||||
pad._buffer = MagicMock()
|
||||
pad._textview = MagicMock()
|
||||
pad._window = MagicMock()
|
||||
pad._window.get_visible.return_value = True
|
||||
|
||||
pad.append_text("kept ")
|
||||
queued = pad._GLib.idle_add.call_args.args[0]
|
||||
queued()
|
||||
pad._buffer.insert.assert_called_once()
|
||||
|
||||
pad.append_text("post clear ")
|
||||
pad._on_clear_clicked()
|
||||
stale = pad._GLib.idle_add.call_args.args[0]
|
||||
stale()
|
||||
self.assertEqual(pad._buffer.insert.call_count, 1)
|
||||
finally:
|
||||
pad._window = None
|
||||
pad.destroy()
|
||||
|
||||
def test_widget_edits_sync_back_to_controller(self) -> None:
|
||||
"""Manual edits in the text view update the controller's buffer."""
|
||||
pad = _pad_without_gtk()
|
||||
try:
|
||||
pad._buffer = MagicMock()
|
||||
pad._buffer.get_text.return_value = "user edited text"
|
||||
pad._on_buffer_changed(pad._buffer)
|
||||
self.assertEqual(pad.controller.text, "user edited text")
|
||||
# delete_last now computes against the edited content.
|
||||
self.assertEqual(pad.delete_last_chars(4), 4)
|
||||
self.assertEqual(pad.controller.text, "user edited ")
|
||||
finally:
|
||||
pad.destroy()
|
||||
|
||||
def test_buffer_changed_skips_programmatic_sync(self) -> None:
|
||||
"""The pad's own widget writes must not feed back into the controller."""
|
||||
pad = _pad_without_gtk()
|
||||
try:
|
||||
pad.controller.append("dictated")
|
||||
pad._syncing_widget = True
|
||||
pad._on_buffer_changed(MagicMock())
|
||||
self.assertEqual(pad.controller.text, "dictated")
|
||||
finally:
|
||||
pad.destroy()
|
||||
|
||||
def test_widget_edit_flushes_queued_ops_before_sync(self) -> None:
|
||||
"""A manual edit must not overwrite queued dictation in the controller.
|
||||
|
||||
Input events outrank idle callbacks, so a keystroke can land before a
|
||||
queued append. The changed handler replays the queue first, keeping
|
||||
the view — and the controller sync — in delivery order.
|
||||
"""
|
||||
pad = _pad_without_gtk(enabled=True)
|
||||
try:
|
||||
pad._gtk_ready = True
|
||||
pad._GLib = MagicMock()
|
||||
pad._buffer = MagicMock()
|
||||
pad._textview = MagicMock()
|
||||
pad._window = MagicMock()
|
||||
pad._window.get_visible.return_value = True
|
||||
|
||||
pad.append_text("queued dictation ")
|
||||
# The user edits before the idle callback runs; the queued op must
|
||||
# be flushed into the widget before the view is read back.
|
||||
pad._buffer.get_text.return_value = "user queued dictation "
|
||||
pad._on_buffer_changed(pad._buffer)
|
||||
|
||||
calls = [c[0] for c in pad._buffer.mock_calls]
|
||||
self.assertLess(calls.index("insert"), calls.index("get_text"))
|
||||
self.assertEqual(pad.controller.text, "user queued dictation ")
|
||||
self.assertEqual(pad._pending_idle, [])
|
||||
finally:
|
||||
pad._window = None
|
||||
pad.destroy()
|
||||
|
||||
def test_flushed_ops_do_not_run_again_on_glib_dispatch(self) -> None:
|
||||
"""An op a flush already ran must no-op when GLib still dispatches it."""
|
||||
pad = _pad_without_gtk(enabled=True)
|
||||
try:
|
||||
pad._gtk_ready = True
|
||||
pad._GLib = MagicMock()
|
||||
pad._buffer = MagicMock()
|
||||
pad._textview = MagicMock()
|
||||
pad._window = MagicMock()
|
||||
pad._window.get_visible.return_value = True
|
||||
|
||||
pad.append_text("queued ")
|
||||
queued = pad._GLib.idle_add.call_args.args[0]
|
||||
pad._flush_idle_ops()
|
||||
pad._buffer.insert.assert_called_once()
|
||||
|
||||
# GLib still fires the callback later; without the pending-entry
|
||||
# guard the insert would land a second time.
|
||||
queued()
|
||||
pad._buffer.insert.assert_called_once()
|
||||
finally:
|
||||
pad._window = None
|
||||
pad.destroy()
|
||||
|
||||
def test_queued_append_lands_before_user_edit(self) -> None:
|
||||
"""Dictation queued before a keystroke must precede it in the buffer."""
|
||||
pad = _pad_without_gtk(enabled=True)
|
||||
try:
|
||||
pad._gtk_ready = True
|
||||
pad._GLib = MagicMock()
|
||||
pad._buffer = MagicMock()
|
||||
pad._textview = MagicMock()
|
||||
pad._window = MagicMock()
|
||||
pad._window.get_visible.return_value = True
|
||||
|
||||
pad.append_text("dictated ")
|
||||
# insert-text/delete-range fire before the edit is applied, so
|
||||
# the queued append lands first and cannot leapfrog new text.
|
||||
pad._on_buffer_user_edit()
|
||||
pad._buffer.insert.assert_called_once()
|
||||
self.assertEqual(pad._pending_idle, [])
|
||||
finally:
|
||||
pad._window = None
|
||||
pad.destroy()
|
||||
|
||||
def test_same_text_appends_do_not_cancel_each_other(self) -> None:
|
||||
"""A stale dispatch must not consume a later entry with the same text."""
|
||||
pad = _pad_without_gtk(enabled=True)
|
||||
try:
|
||||
pad._gtk_ready = True
|
||||
pad._GLib = MagicMock()
|
||||
pad._buffer = MagicMock()
|
||||
pad._textview = MagicMock()
|
||||
pad._window = MagicMock()
|
||||
pad._window.get_visible.return_value = True
|
||||
|
||||
pad.append_text("y")
|
||||
pad._flush_idle_ops()
|
||||
pad.append_text("z")
|
||||
pad.append_text("y")
|
||||
calls = pad._GLib.idle_add.call_args_list
|
||||
calls[0].args[0]() # late dispatch of the already-flushed "y"
|
||||
calls[1].args[0]() # "z"
|
||||
calls[2].args[0]() # "y"
|
||||
|
||||
inserted = [c.args[1] for c in pad._buffer.insert.call_args_list]
|
||||
self.assertEqual(inserted, ["y", "z", "y"])
|
||||
finally:
|
||||
pad._window = None
|
||||
pad.destroy()
|
||||
|
||||
def test_widget_edit_drops_stale_full_refresh(self) -> None:
|
||||
"""A queued full refresh must not erase a manual edit.
|
||||
|
||||
After a voice undo the post-undo refresh waits in the idle queue; a
|
||||
keystroke landing first makes that refresh stale — running it would
|
||||
wipe the edit before it ever reaches the controller.
|
||||
"""
|
||||
pad = _pad_without_gtk(enabled=True)
|
||||
try:
|
||||
pad._gtk_ready = True
|
||||
pad._GLib = MagicMock()
|
||||
pad._buffer = MagicMock()
|
||||
pad._textview = MagicMock()
|
||||
pad._window = MagicMock()
|
||||
pad._window.get_visible.return_value = True
|
||||
|
||||
pad.append_text("one ")
|
||||
pad.append_text("two ")
|
||||
pad.handle_action("undo")
|
||||
# The undo refresh is still queued when the user types.
|
||||
pad._buffer.get_text.return_value = "one two edited"
|
||||
pad._on_buffer_changed(pad._buffer)
|
||||
|
||||
pad._buffer.set_text.assert_not_called()
|
||||
self.assertEqual(pad.controller.text, "one two edited")
|
||||
self.assertEqual(pad._pending_idle, [])
|
||||
|
||||
# GLib still dispatches the dropped refresh later — it must no-op.
|
||||
queued = pad._GLib.idle_add.call_args.args[0]
|
||||
queued()
|
||||
pad._buffer.set_text.assert_not_called()
|
||||
finally:
|
||||
pad._window = None
|
||||
pad.destroy()
|
||||
|
||||
def test_copy_all_flushes_queued_appends(self) -> None:
|
||||
"""Copy All sees segments still waiting in the idle queue."""
|
||||
pad = _pad_without_gtk()
|
||||
try:
|
||||
pad._gtk_ready = True
|
||||
pad._Gtk = MagicMock()
|
||||
pad._Gdk = MagicMock()
|
||||
pad._GLib = MagicMock()
|
||||
pad._copy_button = MagicMock()
|
||||
pad._buffer = MagicMock()
|
||||
pad._buffer.get_text.return_value = "whole pad "
|
||||
|
||||
pad.append_text("queued ")
|
||||
pad._on_copy_all_clicked()
|
||||
|
||||
calls = [c[0] for c in pad._buffer.mock_calls]
|
||||
self.assertLess(calls.index("insert"), calls.index("get_text"))
|
||||
clipboard = pad._Gtk.Clipboard.get.return_value
|
||||
clipboard.set_text.assert_called_once_with("whole pad ", -1)
|
||||
finally:
|
||||
pad.destroy()
|
||||
|
||||
def test_last_segment_passthrough_to_controller(self) -> None:
|
||||
pad = _pad_without_gtk()
|
||||
try:
|
||||
self.assertIsNone(pad.last_segment)
|
||||
pad.append_text("dictated ")
|
||||
self.assertEqual(pad.last_segment, "dictated ")
|
||||
finally:
|
||||
pad.destroy()
|
||||
|
||||
def test_apply_append_stale_generation_is_dropped(self) -> None:
|
||||
pad = _pad_without_gtk()
|
||||
try:
|
||||
pad._buffer = MagicMock()
|
||||
pad._apply_append("old", pad._generation - 1)
|
||||
pad._buffer.insert.assert_not_called()
|
||||
finally:
|
||||
pad.destroy()
|
||||
|
||||
def test_handle_action_undo_redo_sync_widget(self) -> None:
|
||||
pad = _pad_without_gtk()
|
||||
try:
|
||||
pad._GLib = MagicMock()
|
||||
pad._buffer = MagicMock()
|
||||
pad.controller.append("segment ")
|
||||
|
||||
self.assertTrue(pad.handle_action("undo"))
|
||||
self.assertEqual(pad.controller.text, "")
|
||||
pad._GLib.idle_add.call_args.args[0]()
|
||||
pad._buffer.set_text.assert_called_once_with("")
|
||||
|
||||
self.assertTrue(pad.handle_action("redo"))
|
||||
self.assertEqual(pad.controller.text, "segment ")
|
||||
finally:
|
||||
pad.destroy()
|
||||
|
||||
def test_handle_action_with_empty_history_returns_false(self) -> None:
|
||||
pad = _pad_without_gtk()
|
||||
try:
|
||||
self.assertFalse(pad.handle_action("undo"))
|
||||
self.assertFalse(pad.handle_action("redo"))
|
||||
finally:
|
||||
pad.destroy()
|
||||
|
||||
def test_handle_action_selection_commands(self) -> None:
|
||||
pad = _pad_without_gtk()
|
||||
try:
|
||||
pad._GLib = MagicMock()
|
||||
pad._buffer = MagicMock()
|
||||
it = MagicMock()
|
||||
it.get_offset.return_value = 0
|
||||
pad._buffer.get_iter_at_mark.return_value = it
|
||||
pad._buffer.get_text.return_value = "para one\n\npara two"
|
||||
|
||||
for action in (
|
||||
"select_all",
|
||||
"select_line",
|
||||
"select_word",
|
||||
"select_paragraph",
|
||||
):
|
||||
self.assertTrue(pad.handle_action(action))
|
||||
self.assertEqual(pad._GLib.idle_add.call_count, 4)
|
||||
|
||||
# Run the queued select_all and select_line applies.
|
||||
pad._GLib.idle_add.call_args_list[0].args[0]()
|
||||
pad._buffer.select_range.assert_called_once_with(
|
||||
pad._buffer.get_start_iter.return_value,
|
||||
pad._buffer.get_end_iter.return_value,
|
||||
)
|
||||
pad._GLib.idle_add.call_args_list[1].args[0]()
|
||||
it.copy.return_value.set_line_offset.assert_called_once_with(0)
|
||||
pad._GLib.idle_add.call_args_list[3].args[0]()
|
||||
# "para one\n\npara two" — next blank line after offset 0 is at 8.
|
||||
pad._buffer.get_iter_at_offset.assert_called_with(8)
|
||||
finally:
|
||||
pad.destroy()
|
||||
|
||||
def test_handle_action_clipboard_commands(self) -> None:
|
||||
pad = _pad_without_gtk()
|
||||
try:
|
||||
pad._GLib = MagicMock()
|
||||
pad._buffer = MagicMock()
|
||||
pad._Gtk = MagicMock()
|
||||
pad._Gdk = MagicMock()
|
||||
|
||||
for action in ("cut", "copy", "paste"):
|
||||
self.assertTrue(pad.handle_action(action))
|
||||
pad._GLib.idle_add.call_args.args[0]()
|
||||
|
||||
clipboard = pad._Gtk.Clipboard.get.return_value
|
||||
pad._buffer.cut_clipboard.assert_called_once_with(clipboard, True)
|
||||
pad._buffer.copy_clipboard.assert_called_once_with(clipboard)
|
||||
pad._buffer.paste_clipboard.assert_called_once_with(clipboard, None, True)
|
||||
finally:
|
||||
pad.destroy()
|
||||
|
||||
def test_handle_action_rejects_unknown_and_headless(self) -> None:
|
||||
"""Unknown actions and widget commands without a view are refused."""
|
||||
pad = _pad_without_gtk()
|
||||
try:
|
||||
self.assertFalse(pad.handle_action("select_all"))
|
||||
self.assertFalse(pad.handle_action("nonexistent"))
|
||||
finally:
|
||||
pad.destroy()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -574,7 +574,12 @@ def test_refresh_does_not_rearm_under_external_activation() -> None:
|
||||
|
||||
|
||||
def test_refresh_stops_a_held_push_to_talk_first() -> None:
|
||||
"""Rebuilding listeners cannot strand a session waiting on its release."""
|
||||
"""Rebuilding listeners cannot strand a session waiting on its release.
|
||||
|
||||
While a push-to-talk hold is live the rebuild is deferred instead of
|
||||
stopping the session: the manager whose release callback will end the
|
||||
hold is kept alive until the IDLE transition drains the refresh.
|
||||
"""
|
||||
from vocalinux.ui.tray_indicator import TrayIndicator
|
||||
|
||||
tray = _tray_stub(
|
||||
@@ -582,11 +587,17 @@ def test_refresh_stops_a_held_push_to_talk_first() -> None:
|
||||
entries=[{"shortcut": "alt+d", "language": "de"}],
|
||||
)
|
||||
tray.speech_engine.state = RecognitionState.LISTENING
|
||||
tray._language_shortcuts_refresh_pending = False
|
||||
armed = MagicMock()
|
||||
tray._language_shortcut_managers = [armed]
|
||||
|
||||
with patch("vocalinux.ui.tray_indicator.KeyboardShortcutManager"):
|
||||
TrayIndicator.refresh_language_shortcuts(tray)
|
||||
|
||||
tray.speech_engine.stop_recognition.assert_called_once()
|
||||
tray.speech_engine.stop_recognition.assert_not_called()
|
||||
assert tray._language_shortcuts_refresh_pending is True
|
||||
assert tray._language_shortcut_managers == [armed] # release callback survives
|
||||
armed.stop.assert_not_called()
|
||||
|
||||
|
||||
def test_refresh_rebuilds_the_language_listeners() -> None:
|
||||
@@ -909,3 +920,87 @@ def test_begin_recording_disarms_the_previous_target(dialog_class: type[Any]) ->
|
||||
)
|
||||
|
||||
dialog._stop_recording_shortcut.assert_called_once()
|
||||
|
||||
|
||||
def test_persist_rejects_an_edit_that_claims_another_rows_binding(
|
||||
dialog_class: type[Any],
|
||||
) -> None:
|
||||
"""Editing a row onto another row's key must not displace that row."""
|
||||
dialog = _dialog_stub()
|
||||
first_picker, second_picker = MagicMock(), MagicMock()
|
||||
first_entry, second_entry = MagicMock(), MagicMock()
|
||||
first_picker.get_active_id.return_value = "de"
|
||||
second_picker.get_active_id.return_value = "fr"
|
||||
first_entry.get_text.return_value = "alt+f" # typed onto row two's key
|
||||
second_entry.get_text.return_value = "alt+f" # row two still holds it
|
||||
dialog._language_shortcut_rows = [
|
||||
{
|
||||
"language_picker": first_picker,
|
||||
"shortcut_entry": first_entry,
|
||||
"last_valid_shortcut": "alt+d",
|
||||
},
|
||||
{
|
||||
"language_picker": second_picker,
|
||||
"shortcut_entry": second_entry,
|
||||
"last_valid_shortcut": "alt+f",
|
||||
},
|
||||
]
|
||||
dialog._report_language_shortcut_rejections = (
|
||||
lambda rejected: dialog_class._report_language_shortcut_rejections(dialog, rejected)
|
||||
)
|
||||
dialog.config_manager.get_language_shortcuts.return_value = [
|
||||
{"shortcut": "alt+d", "language": "de"},
|
||||
{"shortcut": "alt+f", "language": "fr"},
|
||||
]
|
||||
|
||||
dialog_class._persist_language_shortcuts(dialog)
|
||||
|
||||
# The resolved rows match the saved bindings — the stealing edit lost —
|
||||
# so nothing is rewritten and no listener rebuild is triggered.
|
||||
dialog.config_manager.set_language_shortcuts.assert_not_called()
|
||||
dialog.language_shortcuts_update_callback.assert_not_called()
|
||||
dialog.language_shortcuts_info_label.set_markup.assert_called_once()
|
||||
assert dialog._language_shortcut_rows[0]["last_valid_shortcut"] == "alt+d"
|
||||
|
||||
|
||||
def test_refresh_defers_listener_rebuild_while_dictating() -> None:
|
||||
"""A binding edit during a session waits for IDLE, it must not stop it."""
|
||||
from vocalinux.ui.tray_indicator import TrayIndicator
|
||||
|
||||
tray = _tray_stub(entries=[{"shortcut": "alt+d", "language": "de"}])
|
||||
tray.speech_engine.state = RecognitionState.LISTENING
|
||||
tray._language_shortcuts_refresh_pending = False
|
||||
|
||||
TrayIndicator.refresh_language_shortcuts(tray)
|
||||
|
||||
assert tray._language_shortcuts_refresh_pending is True
|
||||
tray.speech_engine.stop_recognition.assert_not_called()
|
||||
assert tray._language_shortcut_managers == []
|
||||
|
||||
# The next IDLE transition drains the deferred rebuild.
|
||||
tray.speech_engine.state = RecognitionState.IDLE
|
||||
with patch("vocalinux.ui.tray_indicator.KeyboardShortcutManager") as manager_class:
|
||||
manager = MagicMock()
|
||||
manager_class.return_value = manager
|
||||
TrayIndicator._update_ui(tray, RecognitionState.IDLE)
|
||||
|
||||
assert tray._language_shortcuts_refresh_pending is False
|
||||
manager_class.assert_called_once_with(shortcut="alt+d", mode="toggle")
|
||||
manager.start.assert_called_once()
|
||||
|
||||
|
||||
def test_refresh_rebuilds_immediately_when_idle() -> None:
|
||||
"""The normal path still rebuilds the listeners right away."""
|
||||
from vocalinux.ui.tray_indicator import TrayIndicator
|
||||
|
||||
tray = _tray_stub(entries=[{"shortcut": "alt+d", "language": "de"}])
|
||||
tray._language_shortcuts_refresh_pending = False
|
||||
|
||||
with patch("vocalinux.ui.tray_indicator.KeyboardShortcutManager") as manager_class:
|
||||
manager = MagicMock()
|
||||
manager_class.return_value = manager
|
||||
TrayIndicator.refresh_language_shortcuts(tray)
|
||||
|
||||
assert tray._language_shortcuts_refresh_pending is False
|
||||
manager_class.assert_called_once_with(shortcut="alt+d", mode="toggle")
|
||||
manager.start.assert_called_once()
|
||||
|
||||
+359
-92
@@ -8,6 +8,7 @@ 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
|
||||
|
||||
@@ -162,6 +163,7 @@ class TestMainModule(unittest.TestCase):
|
||||
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")
|
||||
@@ -178,6 +180,7 @@ class TestMainModule(unittest.TestCase):
|
||||
mock_text,
|
||||
mock_speech,
|
||||
mock_action_handler,
|
||||
mock_dictation_pad,
|
||||
mock_check_deps,
|
||||
):
|
||||
"""Test that main initializes all the required components."""
|
||||
@@ -192,6 +195,7 @@ class TestMainModule(unittest.TestCase):
|
||||
"general": {"first_run": False},
|
||||
}
|
||||
mock_config_instance.get_model_size_for_engine.return_value = "medium"
|
||||
mock_config_instance.is_dictate_to_pad_enabled.return_value = False
|
||||
mock_config_manager.return_value = mock_config_instance
|
||||
|
||||
# Mock objects
|
||||
@@ -199,12 +203,14 @@ class TestMainModule(unittest.TestCase):
|
||||
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:
|
||||
@@ -253,14 +259,21 @@ class TestMainModule(unittest.TestCase):
|
||||
speech_engine=mock_speech_instance,
|
||||
text_injector=mock_text_instance,
|
||||
transcription_history=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_with(
|
||||
mock_action_instance.handle_action
|
||||
)
|
||||
mock_speech_instance.register_action_callback.assert_called_once()
|
||||
# The registered wrapper delegates non-pad actions to ActionHandler.
|
||||
action_callback = mock_speech_instance.register_action_callback.call_args.args[0]
|
||||
action_callback("undo")
|
||||
mock_action_instance.handle_action.assert_called_once_with("undo")
|
||||
mock_speech_instance.register_state_callback.assert_called_once()
|
||||
|
||||
# Verify the tray indicator was started
|
||||
@@ -289,6 +302,7 @@ class TestMainModule(unittest.TestCase):
|
||||
"speech_recognition": {},
|
||||
"general": {"first_run": False},
|
||||
}
|
||||
mock_config_instance.is_dictate_to_pad_enabled.return_value = False
|
||||
mock_config_manager.return_value = mock_config_instance
|
||||
|
||||
mock_speech_instance = MagicMock()
|
||||
@@ -352,6 +366,7 @@ class TestMainModule(unittest.TestCase):
|
||||
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.is_dictate_to_pad_enabled.return_value = False
|
||||
mock_config_manager.return_value = mock_config_instance
|
||||
|
||||
mock_speech_instance = MagicMock()
|
||||
@@ -1255,111 +1270,301 @@ class TestShouldAppendTrailingSpace(unittest.TestCase):
|
||||
self.assertTrue(_should_append_trailing_space())
|
||||
|
||||
|
||||
def _boot_main_callbacks(
|
||||
*,
|
||||
append_trailing_space: bool = True,
|
||||
inject_ok: bool = True,
|
||||
dictate_to_pad: bool = False,
|
||||
) -> 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.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,
|
||||
)
|
||||
)
|
||||
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,
|
||||
)
|
||||
|
||||
|
||||
class TestMainCallbackTrailingSpaceEdges(unittest.TestCase):
|
||||
"""Exercise trailing-space edge paths through the real main() callback."""
|
||||
|
||||
def _boot_under_patches(self, *, append_trailing_space: bool = True, inject_ok: bool = True):
|
||||
"""Return (exit_stack, text_cb, mock_text) with patches still active."""
|
||||
from contextlib import ExitStack
|
||||
|
||||
stack = ExitStack()
|
||||
stack.enter_context(patch("vocalinux.main.check_dependencies", return_value=True))
|
||||
mock_config_cls = stack.enter_context(
|
||||
patch("vocalinux.ui.config_manager.get_shared_config_manager")
|
||||
)
|
||||
mock_config = MagicMock()
|
||||
mock_config.get_settings.return_value = {
|
||||
"speech_recognition": {},
|
||||
"general": {"first_run": False},
|
||||
}
|
||||
mock_config.get.return_value = False # auto_capitalize off
|
||||
mock_config_cls.return_value = mock_config
|
||||
|
||||
mock_speech_cls = stack.enter_context(
|
||||
patch("vocalinux.speech_recognition.recognition_manager.SpeechRecognitionManager")
|
||||
)
|
||||
mock_speech = MagicMock()
|
||||
mock_speech.engine = "whisper_cpp"
|
||||
mock_speech_cls.return_value = mock_speech
|
||||
|
||||
mock_text_cls = stack.enter_context(
|
||||
patch("vocalinux.text_injection.text_injector.TextInjector")
|
||||
)
|
||||
mock_text = MagicMock()
|
||||
mock_text.inject_text.return_value = inject_ok
|
||||
mock_text_cls.return_value = mock_text
|
||||
|
||||
mock_tray_cls = stack.enter_context(patch("vocalinux.ui.tray_indicator.TrayIndicator"))
|
||||
mock_tray_cls.return_value = MagicMock()
|
||||
|
||||
stack.enter_context(patch("vocalinux.ui.logging_manager.initialize_logging"))
|
||||
stack.enter_context(
|
||||
patch(
|
||||
"vocalinux.main._should_append_trailing_space",
|
||||
return_value=append_trailing_space,
|
||||
)
|
||||
)
|
||||
mock_parse = stack.enter_context(patch("vocalinux.main.parse_arguments"))
|
||||
stack.enter_context(patch("sys.argv", ["vocalinux"]))
|
||||
|
||||
mock_args = MagicMock()
|
||||
mock_args.debug = False
|
||||
mock_args.model = "tiny"
|
||||
mock_args.engine = "whisper_cpp"
|
||||
mock_args.language = "en-us"
|
||||
mock_args.wayland = False
|
||||
mock_args.start_minimized = False
|
||||
mock_parse.return_value = mock_args
|
||||
def test_whitespace_only_is_skipped_through_main(self) -> None:
|
||||
boot = _boot_main_callbacks()
|
||||
try:
|
||||
main()
|
||||
text_cb = mock_speech.register_text_callback.call_args.args[0]
|
||||
except BaseException:
|
||||
# The caller only closes the stack once it gets one back, so a
|
||||
# failure here would leak these patches into every later test.
|
||||
stack.close()
|
||||
raise
|
||||
|
||||
return stack, text_cb, mock_text
|
||||
|
||||
def test_whitespace_only_is_skipped_through_main(self):
|
||||
stack, text_cb, mock_text = self._boot_under_patches()
|
||||
try:
|
||||
text_cb(" \t ")
|
||||
mock_text.inject_text.assert_not_called()
|
||||
boot.text_cb(" \t ")
|
||||
boot.text_system.inject_text.assert_not_called()
|
||||
finally:
|
||||
stack.close()
|
||||
boot.stack.close()
|
||||
|
||||
def test_newline_segment_skips_trailing_space_through_main(self):
|
||||
stack, text_cb, mock_text = self._boot_under_patches()
|
||||
def test_newline_segment_skips_trailing_space_through_main(self) -> None:
|
||||
boot = _boot_main_callbacks()
|
||||
try:
|
||||
text_cb("Hello.\n")
|
||||
mock_text.inject_text.assert_called_once_with("Hello.\n")
|
||||
boot.text_cb("Hello.\n")
|
||||
boot.text_system.inject_text.assert_called_once_with("Hello.\n")
|
||||
finally:
|
||||
stack.close()
|
||||
boot.stack.close()
|
||||
|
||||
def test_legacy_mode_adds_leading_space_in_session(self):
|
||||
stack, text_cb, mock_text = self._boot_under_patches(append_trailing_space=False)
|
||||
def test_legacy_mode_adds_leading_space_in_session(self) -> None:
|
||||
boot = _boot_main_callbacks(append_trailing_space=False)
|
||||
try:
|
||||
text_cb("Hello.")
|
||||
text_cb("World")
|
||||
calls = [c.args[0] for c in mock_text.inject_text.call_args_list]
|
||||
boot.text_cb("Hello.")
|
||||
boot.text_cb("World")
|
||||
calls = [c.args[0] for c in boot.text_system.inject_text.call_args_list]
|
||||
self.assertEqual(calls, ["Hello.", " World"])
|
||||
finally:
|
||||
stack.close()
|
||||
boot.stack.close()
|
||||
|
||||
def test_failed_inject_does_not_remember_text(self):
|
||||
stack, text_cb, mock_text = self._boot_under_patches(
|
||||
append_trailing_space=False, inject_ok=False
|
||||
)
|
||||
def test_failed_inject_does_not_remember_text(self) -> None:
|
||||
boot = _boot_main_callbacks(append_trailing_space=False, inject_ok=False)
|
||||
try:
|
||||
text_cb("Hello.")
|
||||
mock_text.inject_text.reset_mock()
|
||||
mock_text.inject_text.return_value = True
|
||||
text_cb("World")
|
||||
boot.text_cb("Hello.")
|
||||
boot.text_system.inject_text.reset_mock()
|
||||
boot.text_system.inject_text.return_value = True
|
||||
boot.text_cb("World")
|
||||
# Failure means last_injected stays empty; next segment has no leading space
|
||||
mock_text.inject_text.assert_called_once_with("World")
|
||||
boot.text_system.inject_text.assert_called_once_with("World")
|
||||
finally:
|
||||
stack.close()
|
||||
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")
|
||||
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")
|
||||
boot.action_cb("delete_last")
|
||||
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))
|
||||
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")
|
||||
boot.config.is_dictate_to_pad_enabled.return_value = True
|
||||
boot.action_cb("delete_last")
|
||||
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")
|
||||
boot.config.is_dictate_to_pad_enabled.return_value = False
|
||||
boot.action_cb("delete_last")
|
||||
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")
|
||||
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")
|
||||
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"))
|
||||
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")
|
||||
boot.state_cb(RecognitionState.IDLE)
|
||||
boot.action_cb("delete_last")
|
||||
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")
|
||||
boot.text_cb("world")
|
||||
# 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"))
|
||||
boot.pad.handle_action.assert_called_once_with("undo")
|
||||
|
||||
boot.action_cb("delete_last")
|
||||
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")
|
||||
boot.pad.last_segment = None
|
||||
boot.pad.handle_action.return_value = True
|
||||
self.assertTrue(boot.action_cb("undo"))
|
||||
|
||||
self.assertTrue(boot.action_cb("delete_last"))
|
||||
boot.pad.delete_last_chars.assert_not_called()
|
||||
boot.text_system.press_backspace.assert_not_called()
|
||||
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")
|
||||
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"))
|
||||
self.assertTrue(boot.action_cb("delete_last"))
|
||||
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"))
|
||||
self.assertTrue(boot.action_cb("delete_last"))
|
||||
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):
|
||||
@@ -1402,6 +1607,10 @@ class TestSessionHistoryRecording(unittest.TestCase):
|
||||
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")
|
||||
@@ -1613,6 +1822,64 @@ class TestSessionHistoryRecording(unittest.TestCase):
|
||||
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()
|
||||
|
||||
@@ -213,7 +213,9 @@ def test_already_downloaded_auto_apply_runs_on_a_worker(settings_dialog, dialog_
|
||||
):
|
||||
dialog_class._auto_apply_settings(dialog)
|
||||
|
||||
dialog._apply_settings_internal.assert_called_once_with(settings, raise_errors=True)
|
||||
dialog._apply_settings_internal.assert_called_once_with(
|
||||
settings, raise_errors=True, apply_generation=settings_dialog._apply_settings_generation
|
||||
)
|
||||
dialog.speech_engine.reconfigure.assert_not_called()
|
||||
assert dialog._applying_settings is True
|
||||
_run_finish_idle(dialog, dialog_class, idle_calls)
|
||||
@@ -304,7 +306,11 @@ def test_second_pick_during_apply_resyncs_ui_when_the_worker_finishes(
|
||||
assert len(workers) == 1
|
||||
|
||||
workers[0].target()
|
||||
dialog._apply_settings_internal.assert_called_once_with(first, raise_errors=True)
|
||||
dialog._apply_settings_internal.assert_called_once_with(
|
||||
first,
|
||||
raise_errors=True,
|
||||
apply_generation=settings_dialog._apply_settings_generation,
|
||||
)
|
||||
assert dialog._applying_settings is True
|
||||
|
||||
_run_finish_idle(dialog, dialog_class, idle_calls)
|
||||
@@ -1385,13 +1391,73 @@ def test_deferred_text_edits_dropped_when_a_newer_apply_ran(
|
||||
"""A newer apply owns the engine and config; the stashed snapshot stays stale."""
|
||||
dialog = _dialog_stub()
|
||||
dialog._pending_text_edits = {"whispercpp_language_candidates": "en,es"}
|
||||
dialog._pending_text_edit_baseline = {"whispercpp_language_candidates": "en"}
|
||||
dialog._pending_apply_generation = settings_dialog._apply_settings_generation
|
||||
# A newer apply began after the edits were stashed.
|
||||
# A newer apply began after the edits were stashed and rewrote the key.
|
||||
settings_dialog._apply_settings_generation += 1
|
||||
settings_dialog._apply_settings_written["whispercpp_language_candidates"] = (
|
||||
settings_dialog._apply_settings_generation
|
||||
)
|
||||
dialog.config_manager.get.side_effect = lambda section, key, default=None: "en"
|
||||
|
||||
try:
|
||||
with patch.object(settings_dialog.threading, "Thread", _InlineThread):
|
||||
dialog_class._persist_pending_text_edits(dialog)
|
||||
finally:
|
||||
settings_dialog._apply_settings_written.pop("whispercpp_language_candidates", None)
|
||||
|
||||
dialog.speech_engine.reconfigure.assert_not_called()
|
||||
dialog.config_manager.set.assert_not_called()
|
||||
dialog.config_manager.save_settings.assert_not_called()
|
||||
|
||||
|
||||
def test_deferred_text_edits_dropped_when_newer_apply_wrote_baseline_value(
|
||||
settings_dialog: Any, dialog_class: type[Any]
|
||||
) -> None:
|
||||
"""A newer choice equal to the baseline is still a choice the edit must lose to."""
|
||||
dialog = _dialog_stub()
|
||||
dialog._pending_text_edits = {"whispercpp_language_candidates": "en,es"}
|
||||
dialog._pending_text_edit_baseline = {"whispercpp_language_candidates": "en"}
|
||||
dialog._pending_apply_generation = settings_dialog._apply_settings_generation
|
||||
# The reopened dialog applied the key back to its baseline value. The
|
||||
# config is indistinguishable from untouched, yet the write counts as
|
||||
# touched — the closed dialog's older edit must not overwrite it.
|
||||
settings_dialog._apply_settings_generation += 1
|
||||
settings_dialog._apply_settings_written["whispercpp_language_candidates"] = (
|
||||
settings_dialog._apply_settings_generation
|
||||
)
|
||||
dialog.config_manager.get.side_effect = lambda section, key, default=None: "en"
|
||||
|
||||
try:
|
||||
with patch.object(settings_dialog.threading, "Thread", _InlineThread):
|
||||
dialog_class._persist_pending_text_edits(dialog)
|
||||
finally:
|
||||
settings_dialog._apply_settings_written.pop("whispercpp_language_candidates", None)
|
||||
|
||||
dialog.speech_engine.reconfigure.assert_not_called()
|
||||
dialog.config_manager.set.assert_not_called()
|
||||
dialog.config_manager.save_settings.assert_not_called()
|
||||
|
||||
|
||||
def test_superseded_apply_skips_the_stale_snapshot(
|
||||
settings_dialog: Any, dialog_class: type[Any]
|
||||
) -> None:
|
||||
"""A worker whose snapshot a newer apply predates must not write at all."""
|
||||
dialog = _dialog_stub()
|
||||
dialog.speech_engine.state = RecognitionState.LISTENING
|
||||
stale_generation = settings_dialog._apply_settings_generation
|
||||
# A newer apply began after this snapshot was collected.
|
||||
settings_dialog._apply_settings_generation += 1
|
||||
|
||||
with patch.object(settings_dialog.threading, "Thread", _InlineThread):
|
||||
dialog_class._persist_pending_text_edits(dialog)
|
||||
result = dialog_class._apply_settings_internal(
|
||||
dialog,
|
||||
{"whispercpp_language_candidates": "en,es"},
|
||||
apply_generation=stale_generation,
|
||||
)
|
||||
|
||||
assert result is True
|
||||
# Dictation must keep running: the staleness check precedes the stop.
|
||||
dialog.speech_engine.stop_recognition.assert_not_called()
|
||||
dialog.speech_engine.reconfigure.assert_not_called()
|
||||
dialog.config_manager.set.assert_not_called()
|
||||
dialog.config_manager.save_settings.assert_not_called()
|
||||
|
||||
@@ -111,6 +111,50 @@ class TestTranscriptionHistory(unittest.TestCase):
|
||||
history.clear()
|
||||
self.assertFalse(history.extend_latest("x"))
|
||||
|
||||
def test_add_returns_snippet_id(self) -> None:
|
||||
history = TranscriptionHistory()
|
||||
first = history.add("one")
|
||||
second = history.add("two")
|
||||
self.assertIsInstance(first, int)
|
||||
self.assertIsInstance(second, int)
|
||||
self.assertNotEqual(first, second)
|
||||
|
||||
def test_add_returns_none_when_refused(self) -> None:
|
||||
history = TranscriptionHistory(enabled=False)
|
||||
self.assertIsNone(history.add("ignored"))
|
||||
|
||||
def test_extend_entry_targets_the_named_snippet(self) -> None:
|
||||
"""A straggler extends its own session even after newer commits."""
|
||||
history = TranscriptionHistory()
|
||||
first = history.add("one")
|
||||
assert first is not None
|
||||
history.add("two")
|
||||
self.assertTrue(history.extend_entry(first, "tail"))
|
||||
self.assertEqual(history.get_all(), ["two", "one tail"])
|
||||
|
||||
def test_extend_entry_unknown_id_returns_false(self) -> None:
|
||||
history = TranscriptionHistory()
|
||||
history.add("one")
|
||||
self.assertFalse(history.extend_entry(999, "tail"))
|
||||
self.assertEqual(history.get_all(), ["one"])
|
||||
|
||||
def test_extend_entry_after_clear_returns_false(self) -> None:
|
||||
history = TranscriptionHistory()
|
||||
first = history.add("one")
|
||||
assert first is not None
|
||||
history.clear()
|
||||
self.assertFalse(history.extend_entry(first, "tail"))
|
||||
|
||||
def test_extend_entry_refused_from_stale_epoch(self) -> None:
|
||||
history = TranscriptionHistory()
|
||||
first = history.add("one")
|
||||
assert first is not None
|
||||
epoch = history.epoch
|
||||
history.clear()
|
||||
history.add("two")
|
||||
self.assertFalse(history.extend_entry(first, "tail", expected_epoch=epoch))
|
||||
self.assertEqual(history.get_all(), ["two"])
|
||||
|
||||
def test_extend_latest_fires_change_callback(self) -> None:
|
||||
history = TranscriptionHistory()
|
||||
history.add("a")
|
||||
|
||||
Reference in New Issue
Block a user