From c9bbb3d09da9c83a87b47d946e67e303361eaf4f Mon Sep 17 00:00:00 2001 From: gummiflip Date: Thu, 2 Jul 2026 00:00:12 +0200 Subject: [PATCH 1/5] feat(paste): detect terminal windows and use Ctrl+Shift+V xdotool-based active-window detection lets PasteService send Ctrl+Shift+V in known terminal emulators instead of Ctrl+V, which is usually a no-op or copy there. X11-only, falls back cleanly on Wayland/missing xdotool. First step of the PasteService-Hardening roadmap item (clipboard-restore and CopyQ-cleanup follow separately). --- app/paste_service.py | 65 ++++++++++++++++++- tests/test_paste_service.py | 124 ++++++++++++++++++++++++++++++++++++ tests/test_state_machine.py | 2 + 3 files changed, 190 insertions(+), 1 deletion(-) create mode 100644 tests/test_paste_service.py diff --git a/app/paste_service.py b/app/paste_service.py index 0b85126..ea0ffad 100644 --- a/app/paste_service.py +++ b/app/paste_service.py @@ -31,11 +31,42 @@ # KEY_LEFTCTRL=29, KEY_V=47 (siehe /usr/include/linux/input-event-codes.h). # Sequenz: Strg down, V down, V up, Strg up. _CTRL_V_KEYCODES = ["29:1", "47:1", "47:0", "29:0"] +# Strg+Shift+V fuer Terminals (dort ist Strg+V meist "nichts tun" oder Copy). +# KEY_LEFTSHIFT=42 zusaetzlich zu KEY_LEFTCTRL=29, KEY_V=47. +_CTRL_SHIFT_V_KEYCODES = ["29:1", "42:1", "47:1", "47:0", "42:0", "29:0"] +# Bekannte Terminal-Emulator-Fensterklassen (lowercase-Vergleich). X11-only -- +# unter Wayland gibt es ohne Compositor-spezifische Erweiterung keine +# generische "aktives Fenster"-Abfrage wie xdotool sie fuer X11 bietet. +_KNOWN_TERMINAL_WINDOW_CLASSES = frozenset( + { + "gnome-terminal-server", + "xterm", + "konsole", + "kitty", + "alacritty", + "kgx", + "tilix", + "xfce4-terminal", + "terminator", + "mate-terminal", + "org.wezfurlong.wezterm", + "foot", + "footclient", + "lxterminal", + "ghostty", + "org.gnome.terminal", + "com.github.alacritty.alacritty", + } +) # Subprocess-Timeouts: verhindern, dass ein haengendes wl-copy/ydotool den # Transkriptions-Worker dauerhaft blockiert (sonst bleibt der App-State auf # TRANSCRIBING/LLM_REWRITING haengen und kein neuer Hotkey-Toggle ist moeglich). _WL_COPY_TIMEOUT = 5.0 _YDOTOOL_TIMEOUT = 5.0 +_WL_PASTE_TIMEOUT = 5.0 +_XCLIP_PASTE_TIMEOUT = 5.0 +_XDOTOOL_TIMEOUT = 2.0 +_COPYQ_TIMEOUT = 2.0 _YDOTOOL_MISSING_DAEMON_MARKERS = ( "failed to connect", "no such file or directory", @@ -44,6 +75,33 @@ ) +def _detect_active_window_class() -> Optional[str]: + # X11-only: xdotool benoetigt DISPLAY und kann unter Wayland ohne + # Compositor-spezifische Erweiterungen das aktive Fenster nicht generisch abfragen. + if not os.environ.get("DISPLAY"): + return None + if shutil.which("xdotool") is None: + return None + try: + result = subprocess.run( + ["xdotool", "getactivewindow", "getwindowclassname"], + check=True, + stdout=subprocess.PIPE, + stderr=subprocess.DEVNULL, + text=True, + timeout=_XDOTOOL_TIMEOUT, + ) + except (subprocess.TimeoutExpired, subprocess.CalledProcessError, OSError): + return None + window_class = result.stdout.strip().lower() + return window_class or None + + +def _is_terminal_active() -> bool: + window_class = _detect_active_window_class() + return bool(window_class and window_class in _KNOWN_TERMINAL_WINDOW_CLASSES) + + class PasteServiceError(Exception): """Raised when clipboard write or key injection fails hard.""" @@ -156,9 +214,14 @@ def _ydotool_paste(self) -> None: return # Kurze Pause damit Clipboard-Inhalt sicher verfuegbar ist time.sleep(_PASTE_DELAY) + keycodes = _CTRL_SHIFT_V_KEYCODES if _is_terminal_active() else _CTRL_V_KEYCODES + if keycodes is _CTRL_SHIFT_V_KEYCODES: + logger.debug("Aktives Fenster ist ein Terminal -- sende Ctrl+Shift+V via ydotool.") + else: + logger.debug("Aktives Fenster ist kein Terminal -- sende Ctrl+V via ydotool.") try: result = subprocess.run( - ["ydotool", "key", "--key-delay", str(self.key_delay_ms), *_CTRL_V_KEYCODES], + ["ydotool", "key", "--key-delay", str(self.key_delay_ms), *keycodes], check=False, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, diff --git a/tests/test_paste_service.py b/tests/test_paste_service.py new file mode 100644 index 0000000..a068f1d --- /dev/null +++ b/tests/test_paste_service.py @@ -0,0 +1,124 @@ +from __future__ import annotations + +import os +import subprocess +from unittest.mock import MagicMock, patch + +from app.paste_service import ( + _CTRL_SHIFT_V_KEYCODES, + _CTRL_V_KEYCODES, + _XDOTOOL_TIMEOUT, + PasteService, + _detect_active_window_class, + _is_terminal_active, +) + + +class TestDetectActiveWindowClass: + def test_returns_lowercase_class_on_success(self): + result = MagicMock(stdout="Konsole\n", returncode=0) + with patch.dict(os.environ, {"DISPLAY": ":0"}, clear=False): + with patch("app.paste_service.shutil.which", return_value="/usr/bin/xdotool"): + with patch("app.paste_service.subprocess.run", return_value=result) as run_mock: + assert _detect_active_window_class() == "konsole" + run_mock.assert_called_once_with( + ["xdotool", "getactivewindow", "getwindowclassname"], + check=True, + stdout=subprocess.PIPE, + stderr=subprocess.DEVNULL, + text=True, + timeout=_XDOTOOL_TIMEOUT, + ) + + def test_returns_none_if_xdotool_missing(self): + with patch.dict(os.environ, {"DISPLAY": ":0"}, clear=False): + with patch("app.paste_service.shutil.which", return_value=None): + with patch("app.paste_service.subprocess.run") as run_mock: + assert _detect_active_window_class() is None + run_mock.assert_not_called() + + def test_returns_none_on_timeout(self): + with patch.dict(os.environ, {"DISPLAY": ":0"}, clear=False): + with patch("app.paste_service.shutil.which", return_value="/usr/bin/xdotool"): + with patch( + "app.paste_service.subprocess.run", + side_effect=subprocess.TimeoutExpired(cmd="xdotool", timeout=_XDOTOOL_TIMEOUT), + ): + assert _detect_active_window_class() is None + + def test_returns_none_on_called_process_error(self): + with patch.dict(os.environ, {"DISPLAY": ":0"}, clear=False): + with patch("app.paste_service.shutil.which", return_value="/usr/bin/xdotool"): + with patch( + "app.paste_service.subprocess.run", + side_effect=subprocess.CalledProcessError( + returncode=1, + cmd=["xdotool", "getactivewindow", "getwindowclassname"], + ), + ): + assert _detect_active_window_class() is None + + def test_returns_none_on_oserror(self): + with patch.dict(os.environ, {"DISPLAY": ":0"}, clear=False): + with patch("app.paste_service.shutil.which", return_value="/usr/bin/xdotool"): + with patch( + "app.paste_service.subprocess.run", + side_effect=OSError("boom"), + ): + assert _detect_active_window_class() is None + + def test_returns_none_if_display_not_set(self): + with patch.dict(os.environ, {}, clear=True): + with patch("app.paste_service.shutil.which") as which_mock: + with patch("app.paste_service.subprocess.run") as run_mock: + assert _detect_active_window_class() is None + which_mock.assert_not_called() + run_mock.assert_not_called() + + +class TestIsTerminalActive: + def test_returns_true_for_terminal_window(self): + with patch("app.paste_service._detect_active_window_class", return_value="konsole"): + assert _is_terminal_active() is True + + def test_returns_false_for_non_terminal_window(self): + with patch("app.paste_service._detect_active_window_class", return_value="firefox"): + assert _is_terminal_active() is False + + def test_returns_false_when_detection_is_none(self): + with patch("app.paste_service._detect_active_window_class", return_value=None): + assert _is_terminal_active() is False + + +class TestYdotoolPaste: + def test_sends_ctrl_shift_v_when_terminal_active(self): + service = PasteService(autopaste=True, key_delay_ms=135) + result = MagicMock(returncode=0, stderr=b"") + with patch("app.paste_service.shutil.which", return_value="/usr/bin/ydotool"): + with patch("app.paste_service._is_terminal_active", return_value=True): + with patch("app.paste_service.time.sleep"): + with patch("app.paste_service.subprocess.run", return_value=result) as run_mock: + service._ydotool_paste() + run_mock.assert_called_once_with( + ["ydotool", "key", "--key-delay", "135", *_CTRL_SHIFT_V_KEYCODES], + check=False, + stdout=subprocess.DEVNULL, + stderr=subprocess.PIPE, + timeout=5.0, + ) + + def test_sends_ctrl_v_when_non_terminal_active(self): + service = PasteService(autopaste=True, key_delay_ms=246) + result = MagicMock(returncode=0, stderr=b"") + with patch("app.paste_service.shutil.which", return_value="/usr/bin/ydotool"): + with patch("app.paste_service._is_terminal_active", return_value=False): + with patch("app.paste_service.time.sleep"): + with patch("app.paste_service.subprocess.run", return_value=result) as run_mock: + service._ydotool_paste() + run_mock.assert_called_once_with( + ["ydotool", "key", "--key-delay", "246", *_CTRL_V_KEYCODES], + check=False, + stdout=subprocess.DEVNULL, + stderr=subprocess.PIPE, + timeout=5.0, + ) diff --git a/tests/test_state_machine.py b/tests/test_state_machine.py index d662db6..a9e7d7b 100644 --- a/tests/test_state_machine.py +++ b/tests/test_state_machine.py @@ -63,6 +63,7 @@ def test_paste_passes_timeout_to_subprocess(self): def test_paste_uses_configured_key_delay_for_ydotool(self): svc = PasteService(autopaste=True, key_delay_ms=135) with patch("app.paste_service.shutil.which", return_value="/usr/bin/tool"), \ + patch("app.paste_service._is_terminal_active", return_value=False), \ patch("app.paste_service.time.sleep"), \ patch("app.paste_service.subprocess.run") as run_mock: def side_effect(cmd, *args, **kwargs): @@ -100,6 +101,7 @@ def test_force_autopaste_override_enables_ydotool(self): """force_autopaste=True überschreibt autopaste=False: Clipboard + ydotool.""" svc = PasteService(autopaste=False) with patch("app.paste_service.shutil.which", return_value="/usr/bin/tool"), \ + patch("app.paste_service._is_terminal_active", return_value=False), \ patch("app.paste_service.time.sleep"), \ patch("app.paste_service.subprocess.run") as run_mock: def side_effect(cmd, *args, **kwargs): From e4703f8d0141093e4fe685c9aaff708c23d12ff9 Mon Sep 17 00:00:00 2001 From: gummiflip Date: Thu, 2 Jul 2026 00:07:03 +0200 Subject: [PATCH 2/5] feat(paste): restore clipboard content after autopaste MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Reads the previous clipboard content before writing the transcribed text, then restores it once ydotool has delivered the paste keystrokes. Wayland (wl-paste) and X11 (xclip -o) are supported; restore is best-effort and never blocks the main paste path on failure. Only applies to the autopaste flow — clipboard_only() leaves the transcribed text in the clipboard for manual paste, as intended. Second step of the PasteService-Hardening roadmap item (CopyQ cleanup follows separately). --- app/paste_service.py | 70 ++++++++++++++++-- tests/test_paste_service.py | 144 +++++++++++++++++++++++++++++++++++- tests/test_state_machine.py | 2 +- 3 files changed, 206 insertions(+), 10 deletions(-) diff --git a/app/paste_service.py b/app/paste_service.py index ea0ffad..8e5cf5f 100644 --- a/app/paste_service.py +++ b/app/paste_service.py @@ -137,11 +137,13 @@ def paste(self, text: str, force_autopaste: Optional[bool] = None) -> None: logger.debug("paste() mit leerem Text aufgerufen, uebersprungen.") return + do_autopaste = self.autopaste if force_autopaste is None else bool(force_autopaste) + previous_clipboard = self._read_clipboard() if do_autopaste else None self._copy_to_clipboard(text) - do_autopaste = self.autopaste if force_autopaste is None else bool(force_autopaste) - if do_autopaste: - self._ydotool_paste() + if do_autopaste and self._ydotool_paste(): + time.sleep(_PASTE_DELAY) + self._restore_clipboard(previous_clipboard) def clipboard_only(self, text: str) -> None: """Nur Clipboard, kein ydotool -- fuer Faelle wo Auto-Paste unterwuenscht.""" @@ -164,6 +166,58 @@ def _copy_to_clipboard(self, text: str) -> None: "Kein nutzbares Clipboard-Backend gefunden. Installieren: sudo apt install wl-clipboard xclip" ) + def _read_clipboard(self) -> Optional[str]: + if _has_wayland_clipboard(): + command = ["wl-paste", "--no-newline"] + timeout = _WL_PASTE_TIMEOUT + elif _has_x11_clipboard(): + command = ["xclip", "-selection", "clipboard", "-o"] + timeout = _XCLIP_PASTE_TIMEOUT + else: + return None + try: + process = subprocess.Popen( + command, + stdout=subprocess.PIPE, + stderr=subprocess.DEVNULL, + ) + except subprocess.TimeoutExpired: + return None + except subprocess.CalledProcessError: + return None + except (OSError, ValueError) as exc: + logger.debug("Clipboard-Inhalt konnte nicht gelesen werden: %s", exc) + return None + try: + stdout, _ = process.communicate(timeout=timeout) + except subprocess.TimeoutExpired: + process.kill() + process.communicate() + return None + if process.returncode != 0: + return None + if isinstance(stdout, bytes): + try: + return stdout.decode("utf-8") + except UnicodeDecodeError as exc: + logger.debug("Clipboard-Inhalt konnte nicht decodiert werden: %s", exc) + return None + return stdout + + def _restore_clipboard(self, previous: Optional[str]) -> None: + if previous is None: + return + try: + if _has_wayland_clipboard(): + self._wl_copy(previous) + return + if _has_x11_clipboard(): + self._xclip_copy(previous) + return + raise PasteServiceError("Kein Clipboard-Backend fuer Restore verfuegbar.") + except PasteServiceError: + logger.warning("Originalzwischenablage konnte nicht wiederhergestellt werden") + def _wl_copy(self, text: str) -> None: # WICHTIG: wl-copy forkt einen Hintergrund-Daemon, der die Auswahl # "besitzt". Dieser Kindprozess erbt offene Pipes -- mit stderr=PIPE @@ -205,13 +259,13 @@ def _xclip_copy(self, text: str) -> None: except subprocess.CalledProcessError as exc: raise PasteServiceError(f"xclip fehlgeschlagen (rc={exc.returncode})") from exc - def _ydotool_paste(self) -> None: + def _ydotool_paste(self) -> bool: if shutil.which("ydotool") is None: logger.warning( "ydotool nicht gefunden -- Auto-Paste uebersprungen. " "Installieren: sudo apt install ydotool" ) - return + return False # Kurze Pause damit Clipboard-Inhalt sicher verfuegbar ist time.sleep(_PASTE_DELAY) keycodes = _CTRL_SHIFT_V_KEYCODES if _is_terminal_active() else _CTRL_V_KEYCODES @@ -235,7 +289,7 @@ def _ydotool_paste(self) -> None: "(Text liegt bereits im Clipboard).", _YDOTOOL_TIMEOUT, ) - return + return False if result.returncode != 0: stderr = result.stderr.decode(errors="replace").strip() if result.stderr else "" # Nicht fatal -- Clipboard-Inhalt ist bereits gesetzt @@ -244,8 +298,10 @@ def _ydotool_paste(self) -> None: "ydotoold nicht verfügbar -- Auto-Paste uebersprungen " "(Text liegt bereits im Clipboard)." ) - return + return False logger.warning("ydotool Ctrl+V fehlgeschlagen (rc=%d): %s", result.returncode, stderr) + return False + return True def check_dependencies() -> list[str]: diff --git a/tests/test_paste_service.py b/tests/test_paste_service.py index a068f1d..fb1405b 100644 --- a/tests/test_paste_service.py +++ b/tests/test_paste_service.py @@ -8,7 +8,10 @@ _CTRL_SHIFT_V_KEYCODES, _CTRL_V_KEYCODES, _XDOTOOL_TIMEOUT, + _WL_PASTE_TIMEOUT, + _XCLIP_PASTE_TIMEOUT, PasteService, + PasteServiceError, _detect_active_window_class, _is_terminal_active, ) @@ -98,7 +101,7 @@ def test_sends_ctrl_shift_v_when_terminal_active(self): with patch("app.paste_service._is_terminal_active", return_value=True): with patch("app.paste_service.time.sleep"): with patch("app.paste_service.subprocess.run", return_value=result) as run_mock: - service._ydotool_paste() + assert service._ydotool_paste() is True run_mock.assert_called_once_with( ["ydotool", "key", "--key-delay", "135", *_CTRL_SHIFT_V_KEYCODES], check=False, @@ -114,7 +117,7 @@ def test_sends_ctrl_v_when_non_terminal_active(self): with patch("app.paste_service._is_terminal_active", return_value=False): with patch("app.paste_service.time.sleep"): with patch("app.paste_service.subprocess.run", return_value=result) as run_mock: - service._ydotool_paste() + assert service._ydotool_paste() is True run_mock.assert_called_once_with( ["ydotool", "key", "--key-delay", "246", *_CTRL_V_KEYCODES], check=False, @@ -122,3 +125,140 @@ def test_sends_ctrl_v_when_non_terminal_active(self): stderr=subprocess.PIPE, timeout=5.0, ) + + +class TestReadClipboard: + def test_reads_wayland_clipboard_with_wl_paste(self): + service = PasteService() + process = MagicMock(returncode=0) + process.communicate.return_value = (b"vorher", b"") + with patch("app.paste_service._has_wayland_clipboard", return_value=True): + with patch("app.paste_service.subprocess.Popen", return_value=process) as popen_mock: + assert service._read_clipboard() == "vorher" + popen_mock.assert_called_once_with( + ["wl-paste", "--no-newline"], + stdout=subprocess.PIPE, + stderr=subprocess.DEVNULL, + ) + process.communicate.assert_called_once_with(timeout=_WL_PASTE_TIMEOUT) + + def test_reads_x11_clipboard_with_xclip(self): + service = PasteService() + process = MagicMock(returncode=0) + process.communicate.return_value = (b"vorher-x11", b"") + with patch("app.paste_service._has_wayland_clipboard", return_value=False): + with patch("app.paste_service._has_x11_clipboard", return_value=True): + with patch("app.paste_service.subprocess.Popen", return_value=process) as popen_mock: + assert service._read_clipboard() == "vorher-x11" + popen_mock.assert_called_once_with( + ["xclip", "-selection", "clipboard", "-o"], + stdout=subprocess.PIPE, + stderr=subprocess.DEVNULL, + ) + process.communicate.assert_called_once_with(timeout=_XCLIP_PASTE_TIMEOUT) + + def test_returns_none_on_timeout(self): + service = PasteService() + process = MagicMock() + process.communicate.side_effect = [ + subprocess.TimeoutExpired(cmd="wl-paste", timeout=_WL_PASTE_TIMEOUT), + (b"", b""), + ] + with patch("app.paste_service._has_wayland_clipboard", return_value=True): + with patch("app.paste_service.subprocess.Popen", return_value=process): + assert service._read_clipboard() is None + process.kill.assert_called_once_with() + + def test_returns_none_on_error(self): + service = PasteService() + with patch("app.paste_service._has_wayland_clipboard", return_value=True): + with patch("app.paste_service.subprocess.Popen", side_effect=OSError("boom")): + assert service._read_clipboard() is None + + +class TestRestoreClipboard: + def test_none_is_noop(self): + service = PasteService() + with patch.object(service, "_wl_copy") as wl_copy_mock: + with patch.object(service, "_xclip_copy") as xclip_copy_mock: + service._restore_clipboard(None) + wl_copy_mock.assert_not_called() + xclip_copy_mock.assert_not_called() + + def test_restores_wayland_clipboard_with_saved_text(self): + service = PasteService() + with patch("app.paste_service._has_wayland_clipboard", return_value=True): + with patch.object(service, "_wl_copy") as wl_copy_mock: + service._restore_clipboard("wiederherstellen") + wl_copy_mock.assert_called_once_with("wiederherstellen") + + def test_restores_x11_clipboard_with_empty_string(self): + service = PasteService() + with patch("app.paste_service._has_wayland_clipboard", return_value=False): + with patch("app.paste_service._has_x11_clipboard", return_value=True): + with patch.object(service, "_xclip_copy") as xclip_copy_mock: + service._restore_clipboard("") + xclip_copy_mock.assert_called_once_with("") + + def test_logs_warning_on_restore_failure(self): + service = PasteService() + with patch("app.paste_service._has_wayland_clipboard", return_value=True): + with patch.object(service, "_wl_copy", side_effect=PasteServiceError("boom")): + with patch("app.paste_service.logger.warning") as warning_mock: + service._restore_clipboard("vorher") + warning_mock.assert_called_once_with("Originalzwischenablage konnte nicht wiederhergestellt werden") + + +class TestPasteClipboardRestore: + def test_paste_restores_clipboard_after_successful_autopaste(self): + service = PasteService(autopaste=True) + calls: list[str] = [] + + with patch.object(service, "_read_clipboard", side_effect=lambda: calls.append("read") or "vorher"): + with patch.object(service, "_copy_to_clipboard", side_effect=lambda text: calls.append(f"copy:{text}")): + with patch.object(service, "_ydotool_paste", side_effect=lambda: calls.append("ydotool") or True): + with patch("app.paste_service.time.sleep", side_effect=lambda _: calls.append("sleep")): + with patch.object( + service, + "_restore_clipboard", + side_effect=lambda value: calls.append(f"restore:{value}"), + ): + service.paste("neu", force_autopaste=True) + + assert calls == ["read", "copy:neu", "ydotool", "sleep", "restore:vorher"] + + def test_paste_does_not_restore_when_autopaste_disabled(self): + service = PasteService(autopaste=True) + with patch.object(service, "_read_clipboard") as read_mock: + with patch.object(service, "_copy_to_clipboard") as copy_mock: + with patch.object(service, "_ydotool_paste") as paste_mock: + with patch.object(service, "_restore_clipboard") as restore_mock: + service.paste("neu", force_autopaste=False) + read_mock.assert_not_called() + copy_mock.assert_called_once_with("neu") + paste_mock.assert_not_called() + restore_mock.assert_not_called() + + def test_paste_does_not_restore_when_ydotool_did_not_paste(self): + service = PasteService(autopaste=True) + with patch.object(service, "_read_clipboard", return_value="vorher") as read_mock: + with patch.object(service, "_copy_to_clipboard") as copy_mock: + with patch.object(service, "_ydotool_paste", return_value=False) as paste_mock: + with patch("app.paste_service.time.sleep") as sleep_mock: + with patch.object(service, "_restore_clipboard") as restore_mock: + service.paste("neu", force_autopaste=True) + read_mock.assert_called_once_with() + copy_mock.assert_called_once_with("neu") + paste_mock.assert_called_once_with() + sleep_mock.assert_not_called() + restore_mock.assert_not_called() + + def test_clipboard_only_does_not_restore(self): + service = PasteService(autopaste=True) + with patch.object(service, "_read_clipboard") as read_mock: + with patch.object(service, "_copy_to_clipboard") as copy_mock: + with patch.object(service, "_restore_clipboard") as restore_mock: + service.clipboard_only("neu") + read_mock.assert_not_called() + copy_mock.assert_called_once_with("neu") + restore_mock.assert_not_called() diff --git a/tests/test_state_machine.py b/tests/test_state_machine.py index a9e7d7b..eae942d 100644 --- a/tests/test_state_machine.py +++ b/tests/test_state_machine.py @@ -108,7 +108,7 @@ def side_effect(cmd, *args, **kwargs): return subprocess.CompletedProcess(cmd, 0, b"", b"") run_mock.side_effect = side_effect svc.paste("hallo welt", force_autopaste=True) - assert run_mock.call_count == 2 + assert run_mock.call_count == 3 cmd_names = [call.args[0][0] for call in run_mock.call_args_list] assert any(name in ("wl-copy", "xclip") for name in cmd_names) assert "ydotool" in cmd_names From cefc03929fe529d2ad3b5bb584e0f9ae95591dc5 Mon Sep 17 00:00:00 2001 From: gummiflip Date: Thu, 2 Jul 2026 00:11:56 +0200 Subject: [PATCH 3/5] chore(paste): clarify restore-delay comments, drop unreachable except Popen() without check=True never raises CalledProcessError, so that branch in _read_clipboard() was dead. Also documents why paste() and _ydotool_paste() each sleep _PASTE_DELAY for a different reason, so a future reader doesn't mistake it for an accidental double wait. --- app/paste_service.py | 5 ++--- tests/test_paste_service.py | 16 ++++++++++++++++ 2 files changed, 18 insertions(+), 3 deletions(-) diff --git a/app/paste_service.py b/app/paste_service.py index 8e5cf5f..12452ee 100644 --- a/app/paste_service.py +++ b/app/paste_service.py @@ -142,6 +142,7 @@ def paste(self, text: str, force_autopaste: Optional[bool] = None) -> None: self._copy_to_clipboard(text) if do_autopaste and self._ydotool_paste(): + # Kurze Pause, damit die Ziel-App den eingefuegten Text uebernehmen kann time.sleep(_PASTE_DELAY) self._restore_clipboard(previous_clipboard) @@ -183,8 +184,6 @@ def _read_clipboard(self) -> Optional[str]: ) except subprocess.TimeoutExpired: return None - except subprocess.CalledProcessError: - return None except (OSError, ValueError) as exc: logger.debug("Clipboard-Inhalt konnte nicht gelesen werden: %s", exc) return None @@ -266,7 +265,7 @@ def _ydotool_paste(self) -> bool: "Installieren: sudo apt install ydotool" ) return False - # Kurze Pause damit Clipboard-Inhalt sicher verfuegbar ist + # Kurze Pause, damit der neue Clipboard-Inhalt vor Ctrl+V sicher anliegt time.sleep(_PASTE_DELAY) keycodes = _CTRL_SHIFT_V_KEYCODES if _is_terminal_active() else _CTRL_V_KEYCODES if keycodes is _CTRL_SHIFT_V_KEYCODES: diff --git a/tests/test_paste_service.py b/tests/test_paste_service.py index fb1405b..dbd6692 100644 --- a/tests/test_paste_service.py +++ b/tests/test_paste_service.py @@ -7,6 +7,7 @@ from app.paste_service import ( _CTRL_SHIFT_V_KEYCODES, _CTRL_V_KEYCODES, + _PASTE_DELAY, _XDOTOOL_TIMEOUT, _WL_PASTE_TIMEOUT, _XCLIP_PASTE_TIMEOUT, @@ -197,6 +198,7 @@ def test_restores_x11_clipboard_with_empty_string(self): with patch("app.paste_service._has_wayland_clipboard", return_value=False): with patch("app.paste_service._has_x11_clipboard", return_value=True): with patch.object(service, "_xclip_copy") as xclip_copy_mock: + # "" ist ein gueltiger alter Clipboard-Inhalt und darf nicht wie None behandelt werden. service._restore_clipboard("") xclip_copy_mock.assert_called_once_with("") @@ -253,6 +255,20 @@ def test_paste_does_not_restore_when_ydotool_did_not_paste(self): sleep_mock.assert_not_called() restore_mock.assert_not_called() + def test_paste_restores_empty_previous_clipboard_after_successful_autopaste(self): + service = PasteService(autopaste=True) + with patch.object(service, "_read_clipboard", return_value="") as read_mock: + with patch.object(service, "_copy_to_clipboard") as copy_mock: + with patch.object(service, "_ydotool_paste", return_value=True) as paste_mock: + with patch("app.paste_service.time.sleep") as sleep_mock: + with patch.object(service, "_restore_clipboard") as restore_mock: + service.paste("neu", force_autopaste=True) + read_mock.assert_called_once_with() + copy_mock.assert_called_once_with("neu") + paste_mock.assert_called_once_with() + sleep_mock.assert_called_once_with(_PASTE_DELAY) + restore_mock.assert_called_once_with("") + def test_clipboard_only_does_not_restore(self): service = PasteService(autopaste=True) with patch.object(service, "_read_clipboard") as read_mock: From f95e72eeb62b7e46f06d48e5272df61f4ead8005 Mon Sep 17 00:00:00 2001 From: gummiflip Date: Thu, 2 Jul 2026 00:21:23 +0200 Subject: [PATCH 4/5] feat(paste): opportunistic CopyQ history cleanup after autopaste After a successful autopaste-and-restore cycle, remove the transient pasted text from CopyQ's clipboard history if CopyQ is installed and its top entry matches exactly (trailing newline tolerant). No hard CopyQ dependency: missing binary, timeouts, and other errors are swallowed and never affect the main paste path. Third and final step of the PasteService-Hardening roadmap item. --- app/paste_service.py | 26 +++++++++ tests/test_paste_service.py | 113 ++++++++++++++++++++++++++++++++---- tests/test_state_machine.py | 3 + 3 files changed, 131 insertions(+), 11 deletions(-) diff --git a/app/paste_service.py b/app/paste_service.py index 12452ee..25e7d2a 100644 --- a/app/paste_service.py +++ b/app/paste_service.py @@ -145,6 +145,7 @@ def paste(self, text: str, force_autopaste: Optional[bool] = None) -> None: # Kurze Pause, damit die Ziel-App den eingefuegten Text uebernehmen kann time.sleep(_PASTE_DELAY) self._restore_clipboard(previous_clipboard) + self._cleanup_copyq(text) def clipboard_only(self, text: str) -> None: """Nur Clipboard, kein ydotool -- fuer Faelle wo Auto-Paste unterwuenscht.""" @@ -217,6 +218,31 @@ def _restore_clipboard(self, previous: Optional[str]) -> None: except PasteServiceError: logger.warning("Originalzwischenablage konnte nicht wiederhergestellt werden") + def _cleanup_copyq(self, text: str) -> None: + if shutil.which("copyq") is None: + return + try: + result = subprocess.run( + ["copyq", "read", "0"], + stdout=subprocess.PIPE, + stderr=subprocess.DEVNULL, + timeout=_COPYQ_TIMEOUT, + text=True, + ) + # "copyq read" haengt teils einen Zeilenumbruch an -- fuer den + # Vergleich unerheblich, das eigentliche Entfernen bezieht sich + # ohnehin auf den Index (Position 0), nicht auf den String. + if result.stdout.rstrip("\n") != text: + return + subprocess.run( + ["copyq", "remove", "0"], + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + timeout=_COPYQ_TIMEOUT, + ) + except (subprocess.TimeoutExpired, OSError, ValueError) as exc: + logger.debug("CopyQ-Cleanup uebersprungen: %s", exc) + def _wl_copy(self, text: str) -> None: # WICHTIG: wl-copy forkt einen Hintergrund-Daemon, der die Auswahl # "besitzt". Dieser Kindprozess erbt offene Pipes -- mit stderr=PIPE diff --git a/tests/test_paste_service.py b/tests/test_paste_service.py index dbd6692..518baaf 100644 --- a/tests/test_paste_service.py +++ b/tests/test_paste_service.py @@ -7,6 +7,7 @@ from app.paste_service import ( _CTRL_SHIFT_V_KEYCODES, _CTRL_V_KEYCODES, + _COPYQ_TIMEOUT, _PASTE_DELAY, _XDOTOOL_TIMEOUT, _WL_PASTE_TIMEOUT, @@ -211,8 +212,85 @@ def test_logs_warning_on_restore_failure(self): warning_mock.assert_called_once_with("Originalzwischenablage konnte nicht wiederhergestellt werden") +class TestCleanupCopyq: + def test_returns_early_when_copyq_missing(self): + service = PasteService() + with patch("app.paste_service.shutil.which", return_value=None): + with patch("app.paste_service.subprocess.run") as run_mock: + service._cleanup_copyq("neu") + run_mock.assert_not_called() + + def test_removes_top_history_entry_when_text_matches_exactly(self): + service = PasteService() + read_result = MagicMock(stdout="neu") + with patch("app.paste_service.shutil.which", return_value="/usr/bin/copyq"): + with patch("app.paste_service.subprocess.run", side_effect=[read_result, MagicMock()]) as run_mock: + service._cleanup_copyq("neu") + + assert run_mock.call_count == 2 + assert run_mock.call_args_list[0].args[0] == ["copyq", "read", "0"] + assert run_mock.call_args_list[0].kwargs == { + "stdout": subprocess.PIPE, + "stderr": subprocess.DEVNULL, + "timeout": _COPYQ_TIMEOUT, + "text": True, + } + assert run_mock.call_args_list[1].args[0] == ["copyq", "remove", "0"] + assert run_mock.call_args_list[1].kwargs == { + "stdout": subprocess.DEVNULL, + "stderr": subprocess.DEVNULL, + "timeout": _COPYQ_TIMEOUT, + } + + def test_removes_top_history_entry_when_text_matches_with_trailing_newline(self): + """copyq read haengt teils einen Zeilenumbruch an -- muss trotzdem als Match zaehlen.""" + service = PasteService() + read_result = MagicMock(stdout="neu\n") + with patch("app.paste_service.shutil.which", return_value="/usr/bin/copyq"): + with patch("app.paste_service.subprocess.run", side_effect=[read_result, MagicMock()]) as run_mock: + service._cleanup_copyq("neu") + assert run_mock.call_count == 2 + assert run_mock.call_args_list[1].args[0] == ["copyq", "remove", "0"] + + def test_does_not_remove_when_text_differs(self): + service = PasteService() + read_result = MagicMock(stdout="anderer text") + with patch("app.paste_service.shutil.which", return_value="/usr/bin/copyq"): + with patch("app.paste_service.subprocess.run", return_value=read_result) as run_mock: + service._cleanup_copyq("neu") + run_mock.assert_called_once_with( + ["copyq", "read", "0"], + stdout=subprocess.PIPE, + stderr=subprocess.DEVNULL, + timeout=_COPYQ_TIMEOUT, + text=True, + ) + + def test_swallows_errors_from_read_and_remove(self): + service = PasteService() + with patch("app.paste_service.shutil.which", return_value="/usr/bin/copyq"): + with patch( + "app.paste_service.subprocess.run", + side_effect=subprocess.TimeoutExpired(cmd="copyq", timeout=_COPYQ_TIMEOUT), + ): + service._cleanup_copyq("neu") + with patch( + "app.paste_service.subprocess.run", + side_effect=[ + MagicMock(stdout="neu"), + OSError("boom"), + ], + ): + service._cleanup_copyq("neu") + with patch( + "app.paste_service.subprocess.run", + side_effect=ValueError("bad args"), + ): + service._cleanup_copyq("neu") + + class TestPasteClipboardRestore: - def test_paste_restores_clipboard_after_successful_autopaste(self): + def test_paste_restores_clipboard_and_cleans_up_copyq_after_successful_autopaste(self): service = PasteService(autopaste=True) calls: list[str] = [] @@ -225,56 +303,69 @@ def test_paste_restores_clipboard_after_successful_autopaste(self): "_restore_clipboard", side_effect=lambda value: calls.append(f"restore:{value}"), ): - service.paste("neu", force_autopaste=True) + with patch.object( + service, + "_cleanup_copyq", + side_effect=lambda value: calls.append(f"cleanup:{value}"), + ): + service.paste("neu", force_autopaste=True) - assert calls == ["read", "copy:neu", "ydotool", "sleep", "restore:vorher"] + assert calls == ["read", "copy:neu", "ydotool", "sleep", "restore:vorher", "cleanup:neu"] - def test_paste_does_not_restore_when_autopaste_disabled(self): + def test_paste_does_not_restore_or_cleanup_when_autopaste_disabled(self): service = PasteService(autopaste=True) with patch.object(service, "_read_clipboard") as read_mock: with patch.object(service, "_copy_to_clipboard") as copy_mock: with patch.object(service, "_ydotool_paste") as paste_mock: with patch.object(service, "_restore_clipboard") as restore_mock: - service.paste("neu", force_autopaste=False) + with patch.object(service, "_cleanup_copyq") as cleanup_mock: + service.paste("neu", force_autopaste=False) read_mock.assert_not_called() copy_mock.assert_called_once_with("neu") paste_mock.assert_not_called() restore_mock.assert_not_called() + cleanup_mock.assert_not_called() - def test_paste_does_not_restore_when_ydotool_did_not_paste(self): + def test_paste_does_not_restore_or_cleanup_when_ydotool_did_not_paste(self): service = PasteService(autopaste=True) with patch.object(service, "_read_clipboard", return_value="vorher") as read_mock: with patch.object(service, "_copy_to_clipboard") as copy_mock: with patch.object(service, "_ydotool_paste", return_value=False) as paste_mock: with patch("app.paste_service.time.sleep") as sleep_mock: with patch.object(service, "_restore_clipboard") as restore_mock: - service.paste("neu", force_autopaste=True) + with patch.object(service, "_cleanup_copyq") as cleanup_mock: + service.paste("neu", force_autopaste=True) read_mock.assert_called_once_with() copy_mock.assert_called_once_with("neu") paste_mock.assert_called_once_with() sleep_mock.assert_not_called() restore_mock.assert_not_called() + cleanup_mock.assert_not_called() - def test_paste_restores_empty_previous_clipboard_after_successful_autopaste(self): + def test_paste_restores_empty_previous_clipboard_and_cleans_up_copyq(self): service = PasteService(autopaste=True) with patch.object(service, "_read_clipboard", return_value="") as read_mock: with patch.object(service, "_copy_to_clipboard") as copy_mock: with patch.object(service, "_ydotool_paste", return_value=True) as paste_mock: with patch("app.paste_service.time.sleep") as sleep_mock: with patch.object(service, "_restore_clipboard") as restore_mock: - service.paste("neu", force_autopaste=True) + with patch.object(service, "_cleanup_copyq") as cleanup_mock: + service.paste("neu", force_autopaste=True) read_mock.assert_called_once_with() copy_mock.assert_called_once_with("neu") paste_mock.assert_called_once_with() sleep_mock.assert_called_once_with(_PASTE_DELAY) restore_mock.assert_called_once_with("") + cleanup_mock.assert_called_once_with("neu") - def test_clipboard_only_does_not_restore(self): + def test_clipboard_only_does_not_restore_or_cleanup(self): service = PasteService(autopaste=True) with patch.object(service, "_read_clipboard") as read_mock: with patch.object(service, "_copy_to_clipboard") as copy_mock: with patch.object(service, "_restore_clipboard") as restore_mock: - service.clipboard_only("neu") + with patch.object(service, "_cleanup_copyq") as cleanup_mock: + service.clipboard_only("neu") read_mock.assert_not_called() copy_mock.assert_called_once_with("neu") restore_mock.assert_not_called() + cleanup_mock.assert_not_called() diff --git a/tests/test_state_machine.py b/tests/test_state_machine.py index eae942d..5737866 100644 --- a/tests/test_state_machine.py +++ b/tests/test_state_machine.py @@ -64,6 +64,7 @@ def test_paste_uses_configured_key_delay_for_ydotool(self): svc = PasteService(autopaste=True, key_delay_ms=135) with patch("app.paste_service.shutil.which", return_value="/usr/bin/tool"), \ patch("app.paste_service._is_terminal_active", return_value=False), \ + patch.object(PasteService, "_cleanup_copyq"), \ patch("app.paste_service.time.sleep"), \ patch("app.paste_service.subprocess.run") as run_mock: def side_effect(cmd, *args, **kwargs): @@ -102,6 +103,7 @@ def test_force_autopaste_override_enables_ydotool(self): svc = PasteService(autopaste=False) with patch("app.paste_service.shutil.which", return_value="/usr/bin/tool"), \ patch("app.paste_service._is_terminal_active", return_value=False), \ + patch.object(PasteService, "_cleanup_copyq"), \ patch("app.paste_service.time.sleep"), \ patch("app.paste_service.subprocess.run") as run_mock: def side_effect(cmd, *args, **kwargs): @@ -117,6 +119,7 @@ def test_autopaste_false_without_override_skips_ydotool(self): """autopaste=False ohne force_autopaste: nur Clipboard-Write, kein ydotool.""" svc = PasteService(autopaste=False) with patch("app.paste_service.shutil.which", return_value="/usr/bin/tool"), \ + patch.object(PasteService, "_cleanup_copyq"), \ patch("app.paste_service.subprocess.run") as run_mock: run_mock.return_value = subprocess.CompletedProcess([], 0, b"", b"") svc.paste("hallo welt") From 9bcb58a6d265bfb1c1ee740168a64750695d67ce Mon Sep 17 00:00:00 2001 From: gummiflip Date: Thu, 2 Jul 2026 00:41:56 +0200 Subject: [PATCH 5/5] fix(tests): isolate clipboard-restore test from real clipboard tools test_force_autopaste_override_enables_ydotool asserted call_count==3 but only mocked subprocess.run, not the subprocess.Popen used by _read_clipboard(). On a desktop with wl-copy/xclip actually installed this reads the real clipboard and the assertion holds; in CI, where those binaries are absent, _read_clipboard() returns None and the restore step never fires a subprocess.run call, dropping the count to 2 and failing the build. Mock _read_clipboard() directly so the test no longer depends on whichever clipboard tooling happens to be present on the runner. --- tests/test_state_machine.py | 1 + 1 file changed, 1 insertion(+) diff --git a/tests/test_state_machine.py b/tests/test_state_machine.py index 5737866..3a6ea34 100644 --- a/tests/test_state_machine.py +++ b/tests/test_state_machine.py @@ -104,6 +104,7 @@ def test_force_autopaste_override_enables_ydotool(self): with patch("app.paste_service.shutil.which", return_value="/usr/bin/tool"), \ patch("app.paste_service._is_terminal_active", return_value=False), \ patch.object(PasteService, "_cleanup_copyq"), \ + patch.object(PasteService, "_read_clipboard", return_value="alter text"), \ patch("app.paste_service.time.sleep"), \ patch("app.paste_service.subprocess.run") as run_mock: def side_effect(cmd, *args, **kwargs):