diff --git a/src/vocalinux/text_injection/text_injector.py b/src/vocalinux/text_injection/text_injector.py index c579026..4889d90 100644 --- a/src/vocalinux/text_injection/text_injector.py +++ b/src/vocalinux/text_injection/text_injector.py @@ -895,6 +895,58 @@ class TextInjector: ) return False + def _clear_clipboard(self) -> bool: + """ + Clear the clipboard using the first available tool. + + Each backend has its own clear command: + - wl-copy: ``--clear`` flag removes the selection entirely + - xsel: ``--clear`` flag + - xclip: pipe empty input (creates an empty text offer) + + Returns True if the clipboard was cleared successfully. + """ + for tool in self._get_clipboard_tools(): + if self._clipboard_tool_health.get(tool) is False: + continue + try: + if tool == "wl-copy": + subprocess.run( + ["wl-copy", "--clear"], + check=True, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + timeout=self._clipboard_timeout, + ) + return True + if tool == "xsel": + subprocess.run( + ["xsel", "--clipboard", "--clear"], + check=True, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + timeout=self._clipboard_timeout, + ) + return True + if tool == "xclip": + subprocess.run( + ["xclip", "-selection", "clipboard"], + input="", + check=True, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + text=True, + timeout=self._clipboard_timeout, + ) + return True + except ( + subprocess.CalledProcessError, + subprocess.TimeoutExpired, + FileNotFoundError, + ): + continue + return False + def _should_copy_to_clipboard(self) -> bool: """Check if copy-to-clipboard setting is enabled.""" try: @@ -1204,6 +1256,56 @@ class TextInjector: except UnicodeEncodeError: return True + def _read_clipboard(self) -> Optional[str]: + """ + Read the current clipboard content. + + Returns the clipboard text, "" if the clipboard is verifiably empty, + or None if it could not be read (non-text data or no tool available). + """ + host_is_wayland = ( + self._session_environment == DesktopEnvironment.WAYLAND + or os.environ.get("XDG_SESSION_TYPE", "").lower() == "wayland" + or bool(os.environ.get("WAYLAND_DISPLAY")) + ) + + candidates: list[list[str]] = [] + if host_is_wayland and shutil.which("wl-paste"): + candidates.append(["wl-paste", "--no-newline"]) + if shutil.which("xclip"): + candidates.append(["xclip", "-selection", "clipboard", "-o"]) + if shutil.which("xsel"): + candidates.append(["xsel", "--clipboard", "--output"]) + if not host_is_wayland and shutil.which("wl-paste"): + candidates.append(["wl-paste", "--no-newline"]) + + for cmd in candidates: + try: + result = subprocess.run( + cmd, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + timeout=1.0, + text=True, + encoding="utf-8", + errors="replace", + ) + if result.returncode == 0: + return result.stdout + # wl-paste and xclip signal an empty clipboard with a non-zero + # exit code rather than empty stdout. Detect those messages so + # callers can distinguish "empty" (restore to empty) from + # "unreadable" (skip restore). + stderr_lower = result.stderr.lower() + if "nothing is copied" in stderr_lower or ( + "target" in stderr_lower and "not available" in stderr_lower + ): + return "" + except (subprocess.TimeoutExpired, OSError, UnicodeDecodeError): + continue + + return None + def _inject_via_clipboard_paste(self, text: str) -> bool: """ Inject text by copying to clipboard and simulating Ctrl+V with ydotool. @@ -1212,18 +1314,20 @@ class TextInjector: characters (accented letters, CJK, etc.) because ydotool simulates evdev key events which only cover US ASCII keycodes. See issue #362. - Note: this temporarily overwrites the user's clipboard. There is no - attempt to restore it afterward, as there is no safe race-free way to - do so on Wayland. + The previous clipboard content is saved before injection and restored + in a background thread after Ctrl+V completes, so the user's clipboard + is not permanently overwritten. Returns: True if successful, False otherwise """ logger.debug( "Using clipboard-paste injection for non-ASCII text " - "(user clipboard will be temporarily overwritten)" + "(saving clipboard to restore after paste)" ) + previous_clipboard = self._read_clipboard() + if not self._copy_to_clipboard(text): logger.warning("Could not copy text to clipboard for paste injection") return False @@ -1243,11 +1347,41 @@ class TextInjector: timeout=3, ) logger.info(f"Text injected via clipboard paste: '{text[:20]}...' ({len(text)} chars)") - return True except (subprocess.CalledProcessError, subprocess.TimeoutExpired) as e: logger.warning(f"Paste simulation failed: {e}") + if previous_clipboard is not None and not self._should_copy_to_clipboard(): + if previous_clipboard == "": + self._clear_clipboard() + else: + self._copy_to_clipboard(previous_clipboard) return False + # Restore the previous clipboard content after a short delay so the + # Ctrl+V paste has time to land before the clipboard changes. + # Skip restore when the user has opted in to keeping dictated text in + # the clipboard (copy_to_clipboard setting) — restoring would undo that. + if previous_clipboard is not None and not self._should_copy_to_clipboard(): + + def _restore() -> None: + time.sleep(0.3) + # If the user copied something else during the delay, don't + # overwrite their new clipboard content with stale data. + if self._read_clipboard() != text: + logger.debug("Clipboard changed during restore delay; skipping restore") + return + if previous_clipboard == "": + success = self._clear_clipboard() + else: + success = self._copy_to_clipboard(previous_clipboard) + if success: + logger.debug("Clipboard restored to previous content") + else: + logger.debug("Could not restore previous clipboard content") + + threading.Thread(target=_restore, daemon=True).start() + + return True + # ydotool 1.x (Flatpak pins v1.0.4): KEY_LEFTCTRL=29, KEY_V=47 press/release. _YDOTOOL_V1_CTRL_V = ["ydotool", "key", "29:1", "47:1", "47:0", "29:0"] # ydotool 0.1.x (common distro packages): named key sequences. diff --git a/tests/test_clipboard_restore.py b/tests/test_clipboard_restore.py new file mode 100644 index 0000000..9800502 --- /dev/null +++ b/tests/test_clipboard_restore.py @@ -0,0 +1,715 @@ +"""Tests for _read_clipboard() and clipboard restore in _inject_via_clipboard_paste. + +Covers the behaviour introduced in PR #588: before injecting text via the +ydotool clipboard-paste path, the existing clipboard content is saved and +restored ~300 ms after the Ctrl+V keystroke is sent. +""" + +import os +import subprocess +import threading +import time +import unittest +from typing import Any, cast +from unittest.mock import MagicMock, patch + + +def _make_injector() -> Any: + from vocalinux.text_injection.text_injector import DesktopEnvironment, TextInjector + + obj = cast(Any, TextInjector.__new__(TextInjector)) + obj._ibus_injector = None + obj.environment = DesktopEnvironment.WAYLAND + obj._session_environment = DesktopEnvironment.WAYLAND + obj._ibus_ready = False + obj._ibus_init_failed = False + obj._ibus_init_thread = None + obj._state_lock = threading.Lock() + obj._clipboard_tool_health = {} + obj._clipboard_timeout = 0.35 + return obj + + +class TestReadClipboard(unittest.TestCase): + """Unit tests for the _read_clipboard() helper. + + subprocess.run is called with text=True so stdout is always a str. + """ + + def test_wl_paste_used_first_on_wayland(self): + """On Wayland, wl-paste is the first candidate and its output is returned.""" + obj = _make_injector() + with patch.dict(os.environ, {"WAYLAND_DISPLAY": "wayland-0"}): + with patch("vocalinux.text_injection.text_injector.shutil.which") as mock_which: + mock_which.side_effect = lambda cmd: ( + "/usr/bin/wl-paste" if cmd == "wl-paste" else None + ) + with patch("vocalinux.text_injection.text_injector.subprocess.run") as mock_run: + mock_run.return_value = MagicMock(returncode=0, stdout="copied text") + result = obj._read_clipboard() + self.assertEqual(result, "copied text") + + def test_xclip_fallback_when_wl_paste_unavailable(self): + """Falls back to xclip when wl-paste is not installed.""" + obj = _make_injector() + with patch("vocalinux.text_injection.text_injector.shutil.which") as mock_which: + mock_which.side_effect = lambda cmd: ("/usr/bin/xclip" if cmd == "xclip" else None) + with patch("vocalinux.text_injection.text_injector.subprocess.run") as mock_run: + mock_run.return_value = MagicMock(returncode=0, stdout="xclip text") + result = obj._read_clipboard() + self.assertEqual(result, "xclip text") + + def test_xsel_fallback_when_wl_paste_and_xclip_unavailable(self): + """Falls back to xsel when wl-paste and xclip are both not installed.""" + obj = _make_injector() + with patch("vocalinux.text_injection.text_injector.shutil.which") as mock_which: + mock_which.side_effect = lambda cmd: ("/usr/bin/xsel" if cmd == "xsel" else None) + with patch("vocalinux.text_injection.text_injector.subprocess.run") as mock_run: + mock_run.return_value = MagicMock(returncode=0, stdout="xsel text") + result = obj._read_clipboard() + self.assertEqual(result, "xsel text") + + def test_returns_none_when_no_tool_installed(self): + """Returns None when no clipboard read tool is available.""" + obj = _make_injector() + with patch("vocalinux.text_injection.text_injector.shutil.which", return_value=None): + result = obj._read_clipboard() + self.assertIsNone(result) + + def test_returns_none_when_tool_exits_nonzero_with_unknown_stderr(self): + """Returns None when exit code is non-zero and stderr has no recognised empty-clipboard message.""" + obj = _make_injector() + with patch.dict(os.environ, {"WAYLAND_DISPLAY": "wayland-0"}): + with patch( + "vocalinux.text_injection.text_injector.shutil.which", + return_value="/usr/bin/wl-paste", + ): + with patch("vocalinux.text_injection.text_injector.subprocess.run") as mock_run: + mock_run.return_value = MagicMock(returncode=1, stdout="", stderr="some error") + result = obj._read_clipboard() + self.assertIsNone(result) + + def test_returns_none_and_does_not_raise_on_timeout(self): + """Returns None without raising when the clipboard tool times out.""" + obj = _make_injector() + with patch.dict(os.environ, {"WAYLAND_DISPLAY": "wayland-0"}): + with patch( + "vocalinux.text_injection.text_injector.shutil.which", + return_value="/usr/bin/wl-paste", + ): + with patch( + "vocalinux.text_injection.text_injector.subprocess.run", + side_effect=subprocess.TimeoutExpired("wl-paste", 1.0), + ): + result = obj._read_clipboard() + self.assertIsNone(result) + + def test_returns_none_and_does_not_raise_on_oserror(self): + """Returns None without raising when subprocess raises OSError.""" + obj = _make_injector() + with patch.dict(os.environ, {"WAYLAND_DISPLAY": "wayland-0"}): + with patch( + "vocalinux.text_injection.text_injector.shutil.which", + return_value="/usr/bin/wl-paste", + ): + with patch( + "vocalinux.text_injection.text_injector.subprocess.run", + side_effect=OSError("not found"), + ): + result = obj._read_clipboard() + self.assertIsNone(result) + + def test_tries_next_tool_after_failure(self): + """When the first tool fails, the next candidate is tried.""" + obj = _make_injector() + with patch("vocalinux.text_injection.text_injector.shutil.which") as mock_which: + mock_which.side_effect = lambda cmd: ( + "/usr/bin/" + cmd if cmd in ("wl-paste", "xclip") else None + ) + with patch("vocalinux.text_injection.text_injector.subprocess.run") as mock_run: + mock_run.side_effect = [ + MagicMock(returncode=1, stdout=""), # wl-paste fails + MagicMock(returncode=0, stdout="via xclip"), # xclip succeeds + ] + result = obj._read_clipboard() + self.assertEqual(result, "via xclip") + + def test_returns_arabic_text(self): + """Correctly returns multi-byte UTF-8 content such as Arabic text.""" + obj = _make_injector() + arabic = "مرحبا بالعالم" + with patch.dict(os.environ, {"WAYLAND_DISPLAY": "wayland-0"}): + with patch( + "vocalinux.text_injection.text_injector.shutil.which", + return_value="/usr/bin/wl-paste", + ): + with patch("vocalinux.text_injection.text_injector.subprocess.run") as mock_run: + mock_run.return_value = MagicMock(returncode=0, stdout=arabic) + result = obj._read_clipboard() + self.assertEqual(result, arabic) + + def test_wl_paste_used_as_fallback_on_x11(self): + """On X11, wl-paste is added as a last-resort candidate (XWayland environments).""" + from vocalinux.text_injection.text_injector import DesktopEnvironment + + obj = _make_injector() + obj._session_environment = DesktopEnvironment.X11 + with patch.dict(os.environ, {"XDG_SESSION_TYPE": "x11", "WAYLAND_DISPLAY": ""}): + with patch( + "vocalinux.text_injection.text_injector.shutil.which", + return_value="/usr/bin/wl-paste", + ): + with patch("vocalinux.text_injection.text_injector.subprocess.run") as mock_run: + mock_run.return_value = MagicMock(returncode=0, stdout="x11 clipboard") + result = obj._read_clipboard() + self.assertEqual(result, "x11 clipboard") + + +class TestClipboardRestoreAfterInjection(unittest.TestCase): + """Behavioural tests for the save/restore logic in _inject_via_clipboard_paste.""" + + def test_clipboard_restored_to_previous_content(self): + """Previous clipboard content is written back ~300ms after injection.""" + obj = _make_injector() + obj.wayland_tool = "ydotool" + copy_calls: list[str] = [] + + # First read: saves previous content. Second read (inside _restore): clipboard + # still holds injected text, confirming the user hasn't copied anything else. + with patch.object( + obj, "_read_clipboard", side_effect=["original clipboard", "injected text"] + ): + with patch.object( + obj, "_copy_to_clipboard", side_effect=lambda t: copy_calls.append(t) or True + ): + with patch.object( + obj, + "_ydotool_ctrl_v_command", + return_value=["ydotool", "key", "ctrl+v"], + ): + with patch.object(obj, "_should_copy_to_clipboard", return_value=False): + with patch( + "vocalinux.text_injection.text_injector.subprocess.run", + return_value=MagicMock(returncode=0), + ): + result = obj._inject_via_clipboard_paste("injected text") + time.sleep(0.5) # wait for the 300ms restore thread + + self.assertTrue(result) + self.assertEqual(copy_calls[0], "injected text") + self.assertEqual(copy_calls[-1], "original clipboard") + + def test_read_clipboard_is_called_before_overwriting(self): + """_read_clipboard must be called first, before _copy_to_clipboard.""" + obj = _make_injector() + obj.wayland_tool = "ydotool" + call_order: list[str] = [] + + with patch.object( + obj, + "_read_clipboard", + side_effect=lambda: call_order.append("read") or "prev", + ): + with patch.object( + obj, + "_copy_to_clipboard", + side_effect=lambda t: call_order.append(f"copy:{t}") or True, + ): + with patch.object( + obj, + "_ydotool_ctrl_v_command", + return_value=["ydotool", "key", "ctrl+v"], + ): + with patch.object(obj, "_should_copy_to_clipboard", return_value=False): + with patch( + "vocalinux.text_injection.text_injector.subprocess.run", + return_value=MagicMock(returncode=0), + ): + obj._inject_via_clipboard_paste("text") + time.sleep(0.5) + + self.assertEqual(call_order[0], "read", "read must happen before any copy") + self.assertIn("copy:text", call_order) + + def test_no_restore_when_clipboard_was_empty(self): + """No restore is attempted when _read_clipboard() returns None (clipboard empty).""" + obj = _make_injector() + obj.wayland_tool = "ydotool" + copy_calls: list[str] = [] + + with patch.object(obj, "_read_clipboard", return_value=None): + with patch.object( + obj, + "_copy_to_clipboard", + side_effect=lambda t: copy_calls.append(t) or True, + ): + with patch.object( + obj, + "_ydotool_ctrl_v_command", + return_value=["ydotool", "key", "ctrl+v"], + ): + with patch.object(obj, "_should_copy_to_clipboard", return_value=False): + with patch( + "vocalinux.text_injection.text_injector.subprocess.run", + return_value=MagicMock(returncode=0), + ): + result = obj._inject_via_clipboard_paste("text") + time.sleep(0.5) + + self.assertTrue(result) + # Only one call: the injection. No second call for restore. + self.assertEqual(copy_calls, ["text"]) + + def test_no_restore_when_copy_to_clipboard_setting_enabled(self): + """No restore when the user has opted-in to keeping dictated text in clipboard. + + When copy_to_clipboard=true, the user explicitly wants the dictated text + to stay in the clipboard after injection. Restoring the old content would + silently undo that preference. + """ + obj = _make_injector() + obj.wayland_tool = "ydotool" + copy_calls: list[str] = [] + + with patch.object(obj, "_read_clipboard", return_value="old clipboard"): + with patch.object( + obj, + "_copy_to_clipboard", + side_effect=lambda t: copy_calls.append(t) or True, + ): + with patch.object( + obj, + "_ydotool_ctrl_v_command", + return_value=["ydotool", "key", "ctrl+v"], + ): + with patch.object(obj, "_should_copy_to_clipboard", return_value=True): + with patch( + "vocalinux.text_injection.text_injector.subprocess.run", + return_value=MagicMock(returncode=0), + ): + result = obj._inject_via_clipboard_paste("dictated text") + time.sleep(0.5) + + self.assertTrue(result) + # Only one copy call — the injection. The old content must NOT be restored. + self.assertEqual(copy_calls, ["dictated text"]) + + def test_clipboard_restored_when_ctrl_v_fails(self): + """Clipboard is restored immediately when Ctrl+V fails, not silently lost.""" + obj = _make_injector() + obj.wayland_tool = "ydotool" + copy_calls: list[str] = [] + + with patch.object(obj, "_read_clipboard", return_value="prev"): + with patch.object( + obj, + "_copy_to_clipboard", + side_effect=lambda t: copy_calls.append(t) or True, + ): + with patch.object( + obj, + "_ydotool_ctrl_v_command", + return_value=["ydotool", "key", "ctrl+v"], + ): + with patch.object(obj, "_should_copy_to_clipboard", return_value=False): + with patch( + "vocalinux.text_injection.text_injector.subprocess.run", + side_effect=subprocess.CalledProcessError(1, ["ydotool", "key"]), + ): + result = obj._inject_via_clipboard_paste("text") + time.sleep(0.5) + + self.assertFalse(result) + # First copy: injection; second copy: immediate restore after paste failure + self.assertEqual(copy_calls, ["text", "prev"]) + + def test_restore_thread_is_a_daemon(self): + """The restore thread is a daemon so it never blocks interpreter shutdown.""" + obj = _make_injector() + obj.wayland_tool = "ydotool" + started: list[threading.Thread] = [] + original_start = threading.Thread.start + + def capture_start(self_t: threading.Thread) -> None: + started.append(self_t) + original_start(self_t) + + with patch.object(obj, "_read_clipboard", return_value="prev"): + with patch.object(obj, "_copy_to_clipboard", return_value=True): + with patch.object( + obj, + "_ydotool_ctrl_v_command", + return_value=["ydotool", "key", "ctrl+v"], + ): + with patch.object(obj, "_should_copy_to_clipboard", return_value=False): + with patch( + "vocalinux.text_injection.text_injector.subprocess.run", + return_value=MagicMock(returncode=0), + ): + with patch.object(threading.Thread, "start", capture_start): + obj._inject_via_clipboard_paste("text") + + self.assertEqual(len(started), 1, "exactly one restore thread should start") + self.assertTrue(started[0].daemon, "restore thread must be a daemon") + + def test_arabic_content_is_preserved(self): + """Clipboard content containing Arabic text is correctly restored.""" + obj = _make_injector() + obj.wayland_tool = "ydotool" + copy_calls: list[str] = [] + arabic_previous = "نص عربي سابق" + + with patch.object(obj, "_read_clipboard", side_effect=[arabic_previous, "مرحبا"]): + with patch.object( + obj, + "_copy_to_clipboard", + side_effect=lambda t: copy_calls.append(t) or True, + ): + with patch.object( + obj, + "_ydotool_ctrl_v_command", + return_value=["ydotool", "key", "ctrl+v"], + ): + with patch.object(obj, "_should_copy_to_clipboard", return_value=False): + with patch( + "vocalinux.text_injection.text_injector.subprocess.run", + return_value=MagicMock(returncode=0), + ): + obj._inject_via_clipboard_paste("مرحبا") + time.sleep(0.5) + + self.assertEqual(copy_calls[-1], arabic_previous) + + def test_restore_failure_is_handled_gracefully(self): + """No exception when _copy_to_clipboard returns False during restore.""" + obj = _make_injector() + obj.wayland_tool = "ydotool" + copy_calls: list[str] = [] + + def fake_copy(t: str) -> bool: + copy_calls.append(t) + # injection copy succeeds; restore copy fails + return t != "previous_content" + + with patch.object( + obj, "_read_clipboard", side_effect=["previous_content", "injected text"] + ): + with patch.object(obj, "_copy_to_clipboard", side_effect=fake_copy): + with patch.object( + obj, + "_ydotool_ctrl_v_command", + return_value=["ydotool", "key", "ctrl+v"], + ): + with patch.object(obj, "_should_copy_to_clipboard", return_value=False): + with patch( + "vocalinux.text_injection.text_injector.subprocess.run", + return_value=MagicMock(returncode=0), + ): + result = obj._inject_via_clipboard_paste("injected text") + time.sleep(0.5) + + self.assertTrue(result) + # The restore attempt was made but failed — no exception raised + self.assertIn("previous_content", copy_calls) + + def test_clipboard_restored_immediately_when_paste_fails(self): + """Original clipboard is restored right away when Ctrl+V fails, not after 300ms.""" + obj = _make_injector() + obj.wayland_tool = "ydotool" + copy_calls: list[str] = [] + + with patch.object(obj, "_read_clipboard", return_value="original content"): + with patch.object( + obj, + "_copy_to_clipboard", + side_effect=lambda t: copy_calls.append(t) or True, + ): + with patch.object( + obj, + "_ydotool_ctrl_v_command", + return_value=["ydotool", "key", "ctrl+v"], + ): + with patch.object(obj, "_should_copy_to_clipboard", return_value=False): + with patch( + "vocalinux.text_injection.text_injector.subprocess.run", + side_effect=subprocess.CalledProcessError(1, ["ydotool", "key"]), + ): + result = obj._inject_via_clipboard_paste("injected text") + + self.assertFalse(result) + # First copy: injected text; second copy: immediate restore after failure + self.assertEqual(copy_calls, ["injected text", "original content"]) + + def test_no_restore_when_paste_fails_and_clipboard_was_unreadable(self): + """No restore attempted when Ctrl+V fails and clipboard could not be read.""" + obj = _make_injector() + obj.wayland_tool = "ydotool" + copy_calls: list[str] = [] + + with patch.object(obj, "_read_clipboard", return_value=None): + with patch.object( + obj, + "_copy_to_clipboard", + side_effect=lambda t: copy_calls.append(t) or True, + ): + with patch.object( + obj, + "_ydotool_ctrl_v_command", + return_value=["ydotool", "key", "ctrl+v"], + ): + with patch.object(obj, "_should_copy_to_clipboard", return_value=False): + with patch( + "vocalinux.text_injection.text_injector.subprocess.run", + side_effect=subprocess.CalledProcessError(1, ["ydotool", "key"]), + ): + result = obj._inject_via_clipboard_paste("text") + + self.assertFalse(result) + # Only injection copy — no restore since clipboard was unreadable + self.assertEqual(copy_calls, ["text"]) + + def test_empty_clipboard_is_cleared_after_injection(self): + """When clipboard was empty, _clear_clipboard() is called instead of copy("").""" + obj = _make_injector() + obj.wayland_tool = "ydotool" + clear_called: list[bool] = [] + + with patch.object(obj, "_read_clipboard", side_effect=["", "new text"]): + with patch.object(obj, "_copy_to_clipboard", return_value=True): + with patch.object( + obj, "_clear_clipboard", side_effect=lambda: clear_called.append(True) or True + ): + with patch.object( + obj, + "_ydotool_ctrl_v_command", + return_value=["ydotool", "key", "ctrl+v"], + ): + with patch.object(obj, "_should_copy_to_clipboard", return_value=False): + with patch( + "vocalinux.text_injection.text_injector.subprocess.run", + return_value=MagicMock(returncode=0), + ): + result = obj._inject_via_clipboard_paste("new text") + time.sleep(0.5) + + self.assertTrue(result) + self.assertTrue( + clear_called, "_clear_clipboard() must be called for empty previous content" + ) + + def test_restore_skipped_when_clipboard_changed_during_delay(self): + """Restore is skipped if the user copied something else during the 300ms window.""" + obj = _make_injector() + obj.wayland_tool = "ydotool" + copy_calls: list[str] = [] + + # First read: previous content. Second read (inside _restore): user copied "new thing" + with patch.object(obj, "_read_clipboard", side_effect=["original", "new thing"]): + with patch.object( + obj, + "_copy_to_clipboard", + side_effect=lambda t: copy_calls.append(t) or True, + ): + with patch.object( + obj, + "_ydotool_ctrl_v_command", + return_value=["ydotool", "key", "ctrl+v"], + ): + with patch.object(obj, "_should_copy_to_clipboard", return_value=False): + with patch( + "vocalinux.text_injection.text_injector.subprocess.run", + return_value=MagicMock(returncode=0), + ): + result = obj._inject_via_clipboard_paste("injected text") + time.sleep(0.5) + + self.assertTrue(result) + # Only the injection copy — restore was skipped because clipboard changed + self.assertEqual(copy_calls, ["injected text"]) + + def test_restore_proceeds_when_clipboard_unchanged(self): + """Restore fires when clipboard still holds the injected text after 300ms.""" + obj = _make_injector() + obj.wayland_tool = "ydotool" + copy_calls: list[str] = [] + + # First read: previous content. Second read (inside _restore): still the injected text + with patch.object(obj, "_read_clipboard", side_effect=["original", "injected text"]): + with patch.object( + obj, + "_copy_to_clipboard", + side_effect=lambda t: copy_calls.append(t) or True, + ): + with patch.object( + obj, + "_ydotool_ctrl_v_command", + return_value=["ydotool", "key", "ctrl+v"], + ): + with patch.object(obj, "_should_copy_to_clipboard", return_value=False): + with patch( + "vocalinux.text_injection.text_injector.subprocess.run", + return_value=MagicMock(returncode=0), + ): + result = obj._inject_via_clipboard_paste("injected text") + time.sleep(0.5) + + self.assertTrue(result) + self.assertEqual(copy_calls, ["injected text", "original"]) + + +class TestClearClipboard(unittest.TestCase): + """Unit tests for the _clear_clipboard() helper.""" + + def test_wl_copy_clear_flag_used(self): + """Uses 'wl-copy --clear' on Wayland.""" + obj = _make_injector() + with patch( + "vocalinux.text_injection.text_injector.shutil.which", return_value="/usr/bin/wl-copy" + ): + with patch("vocalinux.text_injection.text_injector.subprocess.run") as mock_run: + mock_run.return_value = MagicMock(returncode=0) + result = obj._clear_clipboard() + self.assertTrue(result) + cmd = mock_run.call_args[0][0] + self.assertEqual(cmd, ["wl-copy", "--clear"]) + + def test_xsel_clear_flag_used(self): + """Uses 'xsel --clipboard --clear' when xsel is available.""" + obj = _make_injector() + obj._session_environment = None + with patch( + "vocalinux.text_injection.text_injector.shutil.which", + side_effect=lambda cmd: "/usr/bin/xsel" if cmd == "xsel" else None, + ): + with patch("vocalinux.text_injection.text_injector.subprocess.run") as mock_run: + mock_run.return_value = MagicMock(returncode=0) + with patch.dict("os.environ", {"XDG_SESSION_TYPE": "x11", "WAYLAND_DISPLAY": ""}): + result = obj._clear_clipboard() + self.assertTrue(result) + cmd = mock_run.call_args[0][0] + self.assertEqual(cmd, ["xsel", "--clipboard", "--clear"]) + + def test_xclip_empty_input_used(self): + """Uses xclip with empty input when xclip is the only available tool.""" + obj = _make_injector() + obj._session_environment = None + with patch( + "vocalinux.text_injection.text_injector.shutil.which", + side_effect=lambda cmd: "/usr/bin/xclip" if cmd == "xclip" else None, + ): + with patch("vocalinux.text_injection.text_injector.subprocess.run") as mock_run: + mock_run.return_value = MagicMock(returncode=0) + with patch.dict("os.environ", {"XDG_SESSION_TYPE": "x11", "WAYLAND_DISPLAY": ""}): + result = obj._clear_clipboard() + self.assertTrue(result) + call_kwargs = mock_run.call_args + self.assertEqual(call_kwargs[0][0], ["xclip", "-selection", "clipboard"]) + + def test_skips_unhealthy_tool_and_tries_next(self): + """Skips a tool marked unhealthy; returns False when it was the only tool.""" + obj = _make_injector() + obj._clipboard_tool_health["wl-copy"] = False + with patch( + "vocalinux.text_injection.text_injector.shutil.which", + side_effect=lambda cmd: "/usr/bin/wl-copy" if cmd == "wl-copy" else None, + ): + with patch("vocalinux.text_injection.text_injector.subprocess.run") as mock_run: + result = obj._clear_clipboard() + self.assertFalse(result) + mock_run.assert_not_called() + + def test_exception_falls_through_to_next_tool(self): + """When one tool throws, the next candidate is tried.""" + obj = _make_injector() + obj._session_environment = None + with patch( + "vocalinux.text_injection.text_injector.shutil.which", + side_effect=lambda cmd: "/usr/bin/" + cmd if cmd in ("xclip", "xsel") else None, + ): + with patch("vocalinux.text_injection.text_injector.subprocess.run") as mock_run: + mock_run.side_effect = [ + subprocess.CalledProcessError(1, ["xclip"]), # xclip fails + MagicMock(returncode=0), # xsel succeeds + ] + with patch.dict("os.environ", {"XDG_SESSION_TYPE": "x11", "WAYLAND_DISPLAY": ""}): + result = obj._clear_clipboard() + self.assertTrue(result) + + def test_clear_called_when_paste_fails_and_clipboard_was_empty(self): + """_clear_clipboard() is called immediately when paste fails and clipboard was empty.""" + obj = _make_injector() + obj.wayland_tool = "ydotool" + clear_called: list[bool] = [] + + with patch.object(obj, "_read_clipboard", return_value=""): + with patch.object(obj, "_copy_to_clipboard", return_value=True): + with patch.object( + obj, "_clear_clipboard", side_effect=lambda: clear_called.append(True) or True + ): + with patch.object( + obj, + "_ydotool_ctrl_v_command", + return_value=["ydotool", "key", "ctrl+v"], + ): + with patch.object(obj, "_should_copy_to_clipboard", return_value=False): + with patch( + "vocalinux.text_injection.text_injector.subprocess.run", + side_effect=subprocess.CalledProcessError(1, ["ydotool", "key"]), + ): + result = obj._inject_via_clipboard_paste("text") + + self.assertFalse(result) + self.assertTrue( + clear_called, + "_clear_clipboard() must be called when paste fails with empty previous clipboard", + ) + + def test_returns_false_when_no_tool_available(self): + """Returns False when no clipboard tool is installed.""" + obj = _make_injector() + with patch("vocalinux.text_injection.text_injector.shutil.which", return_value=None): + result = obj._clear_clipboard() + self.assertFalse(result) + + +class TestReadClipboardEmptyDetection(unittest.TestCase): + """Tests for _read_clipboard() detecting empty clipboard via stderr patterns.""" + + def test_wl_paste_nothing_is_copied_returns_empty_string(self): + """Returns '' when wl-paste stderr says 'Nothing is copied'.""" + obj = _make_injector() + with patch("shutil.which", return_value="/usr/bin/wl-paste"): + with patch( + "vocalinux.text_injection.text_injector.subprocess.run", + return_value=MagicMock(returncode=1, stdout="", stderr="Nothing is copied"), + ): + result = obj._read_clipboard() + self.assertEqual(result, "") + + def test_xclip_target_not_available_returns_empty_string(self): + """Returns '' when xclip stderr says target STRING not available.""" + obj = _make_injector() + obj._session_environment = None # non-Wayland so xclip is tried + with patch( + "shutil.which", side_effect=lambda cmd: "/usr/bin/xclip" if cmd == "xclip" else None + ): + with patch( + "vocalinux.text_injection.text_injector.subprocess.run", + return_value=MagicMock( + returncode=1, stdout="", stderr="Error: target STRING not available" + ), + ): + with patch.dict("os.environ", {"XDG_SESSION_TYPE": "x11", "WAYLAND_DISPLAY": ""}): + result = obj._read_clipboard() + self.assertEqual(result, "") + + def test_non_empty_clipboard_error_still_returns_none(self): + """A non-zero exit code with unknown stderr does not trigger empty detection.""" + obj = _make_injector() + with patch("shutil.which", return_value=None): + result = obj._read_clipboard() + self.assertIsNone(result) + + +if __name__ == "__main__": + unittest.main()