Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion app/blitztext_linux.py
Original file line number Diff line number Diff line change
Expand Up @@ -92,7 +92,7 @@ def _infer_wayland_display() -> Optional[str]:
def _is_hotkey_device_access_error(err_msg: str) -> bool:
"""Return True for hotkey startup errors that can fall back to GUI/tray."""
text = err_msg.casefold()
return any(marker in text for marker in ("tastatur", "input", "evdev"))
return any(marker in text for marker in ("tastatur", "input", "evdev", "flatpak", "/dev/input", "/dev/uinput"))


def _require_display_environment() -> None:
Expand Down
3 changes: 3 additions & 0 deletions app/hotkey_service.py
Original file line number Diff line number Diff line change
Expand Up @@ -426,6 +426,9 @@ def _group_names() -> Set[str]:


def _build_missing_keyboard_message(context: str = "") -> str:
if os.path.exists("/.flatpak-info"):
return "Globale Hotkeys sind im Flatpak nicht verfügbar"

suffix = f" {context}" if context else ""
hints = []
if not os.path.exists("/dev/uinput"):
Expand Down
78 changes: 76 additions & 2 deletions app/paste_service.py
Original file line number Diff line number Diff line change
Expand Up @@ -136,6 +136,10 @@ def paste(self, text: str, force_autopaste: Optional[bool] = None) -> None:
return

do_autopaste = self.autopaste if force_autopaste is None else bool(force_autopaste)
if do_autopaste and shutil.which("ydotool") is None:
logger.info("ydotool fehlt, Auto-Paste wird deaktiviert.")
do_autopaste = False
Comment thread
TimInTech marked this conversation as resolved.

logger.info("Paste request: %d chars, autopaste=%s", len(text), do_autopaste)
previous_clipboard = self._read_clipboard() if do_autopaste else None
self._copy_to_clipboard(text)
Expand Down Expand Up @@ -163,6 +167,9 @@ def _copy_to_clipboard(self, text: str) -> None:
if _has_x11_clipboard():
self._xclip_copy(text)
return
if _has_qt_clipboard():
self._qt_copy(text)
return
Comment thread
TimInTech marked this conversation as resolved.
raise PasteServiceError(
"Kein nutzbares Clipboard-Backend gefunden. Installieren: sudo apt install wl-clipboard xclip"
)
Expand All @@ -174,6 +181,8 @@ def _read_clipboard(self) -> Optional[str]:
elif _has_x11_clipboard():
command = ["xclip", "-selection", "clipboard", "-o"]
timeout = _XCLIP_PASTE_TIMEOUT
elif _has_qt_clipboard():
return self._qt_read()
else:
return None
try:
Expand Down Expand Up @@ -213,6 +222,9 @@ def _restore_clipboard(self, previous: Optional[str]) -> None:
if _has_x11_clipboard():
self._xclip_copy(previous)
return
if _has_qt_clipboard():
self._qt_copy(previous)
return
raise PasteServiceError("Kein Clipboard-Backend fuer Restore verfuegbar.")
except PasteServiceError:
logger.warning("Originalzwischenablage konnte nicht wiederhergestellt werden")
Expand Down Expand Up @@ -341,16 +353,70 @@ def _ydotool_paste(self) -> bool:
logger.info("Auto-paste key injection completed with %s.", shortcut)
return not wayland_unknown

def _qt_copy(self, text: str) -> None:
try:
from PyQt6.QtWidgets import QApplication
from PyQt6.QtCore import QMetaObject, Qt, Q_ARG, QThread, QObject, pyqtSlot

app = QApplication.instance()
if not app:
raise PasteServiceError("Qt QApplication nicht verfuegbar.")
cb = app.clipboard()
if not cb:
raise PasteServiceError("Qt Clipboard nicht verfuegbar.")

if QThread.currentThread() == app.thread():
cb.setText(text)
else:
class QtClipboardHelper(QObject):
@pyqtSlot(str)
def set_text(self, t: str):
c = QApplication.instance().clipboard()
if c:
c.setText(t)

helper = QtClipboardHelper()
helper.moveToThread(app.thread())
QMetaObject.invokeMethod(
helper,
"set_text",
Qt.ConnectionType.BlockingQueuedConnection,
Q_ARG(str, text)
)
logger.info("Clipboard write completed via Qt (%d chars).", len(text))
except ImportError as exc:
raise PasteServiceError("PyQt6 Import fehlgeschlagen.") from exc

def _qt_read(self) -> Optional[str]:
try:
from PyQt6.QtWidgets import QApplication
from PyQt6.QtCore import QThread
app = QApplication.instance()
if not app:
return None
cb = app.clipboard()
if not cb:
return None

if QThread.currentThread() == app.thread():
return cb.text()
else:
return None
Comment on lines +401 to +404

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Marshal Qt clipboard reads before restoring

When the Qt fallback is the selected clipboard backend and paste is invoked from the transcription worker (see app/blitztext_linux.py where QThreadPool runs _TranscribeWorker.run), this branch returns None instead of reading the current clipboard. If ydotool is installed and auto-paste succeeds in an environment without wl-copy/xclip, previous_clipboard stays None, so _restore_clipboard(previous_clipboard) becomes a no-op and the user's original clipboard is overwritten. The existing fix only marshals Qt writes; Qt reads need the same treatment for worker-thread paste calls.

Useful? React with 👍 / 👎.

Copy link
Copy Markdown
Owner Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Bewusst als akzeptierter Narrow Edge-Case / Folgepunkt behandelt — Thread absichtlich NICHT resolved.

Der Fix in 13e20bb marshallt Qt-Clipboard-Writes auf den GUI-Thread; Qt-Reads aus dem Worker-Thread bleiben bewusst konservativ: _qt_read() gibt außerhalb des GUI-Threads None zurück (app/paste_service.py:401-404). Betroffen ist ausschließlich die seltene Kombination: Qt-Fallback als einziges Backend (weder wl-copy noch xclip vorhanden) UND ydotool installiert UND Paste aus dem Transkriptions-Worker (QThreadPool). Nur dann bleibt previous_clipboard=None, _restore_clipboard wird zum No-op und der transkribierte Text bleibt im Clipboard stehen.

Auf Standard-Wayland/X11-Hosts sowie bei GUI-Thread-Paste (ComposeWindow._on_paste_clicked) greift der Restore unverändert. Kein Datenverlust über den Clipboard-Inhalt hinaus. Wird als bekannter Folgepunkt für ein späteres Worker-Thread-Read-Marshalling dokumentiert.

except Exception as exc:
logger.debug("Qt clipboard read failed: %s", exc)
return None


def check_dependencies() -> list[str]:
"""Gibt eine Liste fehlender System-Abhaengigkeiten zurueck.

Verwendet von install.sh-Verifikation und Einstellungs-Dialog.
"""
missing = []
if shutil.which("wl-copy") is None and shutil.which("xclip") is None:
is_flatpak = os.path.exists("/.flatpak-info")
if shutil.which("wl-copy") is None and shutil.which("xclip") is None and not is_flatpak:
missing.append("wl-clipboard oder xclip")
if shutil.which("ydotool") is None:
if shutil.which("ydotool") is None and not is_flatpak:
missing.append("ydotool")
return missing

Expand All @@ -369,6 +435,14 @@ def _has_x11_clipboard() -> bool:
return bool(os.environ.get("DISPLAY") and shutil.which("xclip") is not None)


def _has_qt_clipboard() -> bool:
try:
from PyQt6.QtWidgets import QApplication
return QApplication.instance() is not None
except ImportError:
return False


def _looks_like_missing_ydotoold(stderr: str) -> bool:
lowered = stderr.lower()
return any(marker in lowered for marker in _YDOTOOL_MISSING_DAEMON_MARKERS)
132 changes: 132 additions & 0 deletions tests/test_flatpak_fallback.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,132 @@
import pytest
from unittest.mock import patch, MagicMock

from app.paste_service import PasteService, check_dependencies
from app.hotkey_service import _build_missing_keyboard_message


# 1. PasteService.check_dependencies()
def test_check_dependencies_flatpak():
def mock_exists(path):
return path == "/.flatpak-info"

def mock_which(cmd):
return None # wl-copy, xclip, ydotool als fehlend

with patch("app.paste_service.os.path.exists", side_effect=mock_exists), \
patch("app.paste_service.shutil.which", side_effect=mock_which):
missing = check_dependencies()
assert missing == [] # Keine harte fehlende Abhängigkeit im Flatpak


def test_check_dependencies_no_flatpak():
def mock_exists(path):
return False

def mock_which(cmd):
return None

with patch("app.paste_service.os.path.exists", side_effect=mock_exists), \
patch("app.paste_service.shutil.which", side_effect=mock_which):
missing = check_dependencies()
assert "wl-clipboard oder xclip" in missing
assert "ydotool" in missing


# 2. HotkeyService Flatpak-Hinweis
def test_hotkey_service_flatpak_hint():
def mock_exists(path):
return path == "/.flatpak-info"

with patch("app.hotkey_service.os.path.exists", side_effect=mock_exists):
msg = _build_missing_keyboard_message()
assert msg == "Globale Hotkeys sind im Flatpak nicht verfügbar"


# 3. PasteService Auto-Paste ohne ydotool
def test_paste_service_autopaste_without_ydotool():
def mock_which(cmd):
if cmd == "ydotool":
return None
if cmd == "wl-copy":
return "/usr/bin/wl-copy"
return None

svc = PasteService(autopaste=True)

with patch("app.paste_service.shutil.which", side_effect=mock_which), \
patch("app.paste_service._has_wayland_clipboard", return_value=True), \
patch("app.paste_service.PasteService._wl_copy") as mock_wl_copy, \
patch("app.paste_service.PasteService._ydotool_paste") as mock_ydotool:

svc.paste("test text")

# Clipboard copy continues
mock_wl_copy.assert_called_once_with("test text")

# Auto-Paste is skipped, no ydotool called
mock_ydotool.assert_not_called()


# 4. PasteService Qt-Clipboard-Fallback
def test_paste_service_qt_clipboard_fallback():
svc = PasteService(autopaste=False)

with patch("app.paste_service._has_wayland_clipboard", return_value=False), \
patch("app.paste_service._has_x11_clipboard", return_value=False), \
patch("app.paste_service._has_qt_clipboard", return_value=True), \
patch("app.paste_service.PasteService._qt_copy") as mock_qt_copy:

svc.paste("fallback text")
mock_qt_copy.assert_called_once_with("fallback text")


def test_paste_service_qt_read_thread_safety():
svc = PasteService(autopaste=False)

app_mock = MagicMock()
app_mock.thread.return_value = "main_thread"

cb_mock = MagicMock()
cb_mock.text.return_value = "clipboard content"
app_mock.clipboard.return_value = cb_mock

def mock_current_thread_gui():
return "main_thread"

def mock_current_thread_other():
return "other_thread"

# Test 1: GUI Thread reads successfully
with patch("PyQt6.QtWidgets.QApplication.instance", return_value=app_mock), \
patch("PyQt6.QtCore.QThread.currentThread", side_effect=mock_current_thread_gui):

result = svc._qt_read()
assert result == "clipboard content"

# Test 2: Non-GUI Thread returns None
with patch("PyQt6.QtWidgets.QApplication.instance", return_value=app_mock), \
patch("PyQt6.QtCore.QThread.currentThread", side_effect=mock_current_thread_other):

result = svc._qt_read()
assert result is None


def test_paste_service_qt_copy_real_thread():
import threading
from PyQt6.QtWidgets import QApplication
app = QApplication.instance() or QApplication(["-platform", "offscreen"])

svc = PasteService(autopaste=False)

def background_task():
svc._qt_copy("threaded content test")

t = threading.Thread(target=background_task)
t.start()

while t.is_alive():
app.processEvents()
t.join(0.01)

assert app.clipboard().text() == "threaded content test"
77 changes: 45 additions & 32 deletions tests/test_paste_service.py
Original file line number Diff line number Diff line change
Expand Up @@ -219,6 +219,15 @@ def test_restores_x11_clipboard_with_empty_string(self):
service._restore_clipboard("")
xclip_copy_mock.assert_called_once_with("")

def test_restores_qt_clipboard_with_saved_text(self):
service = PasteService()
with patch("app.paste_service._has_wayland_clipboard", return_value=False):
with patch("app.paste_service._has_x11_clipboard", return_value=False):
with patch("app.paste_service._has_qt_clipboard", return_value=True):
with patch.object(service, "_qt_copy") as qt_copy_mock:
service._restore_clipboard("qt-restore")
qt_copy_mock.assert_called_once_with("qt-restore")

def test_logs_warning_on_restore_failure(self):
service = PasteService()
with patch("app.paste_service._has_wayland_clipboard", return_value=True):
Expand Down Expand Up @@ -310,32 +319,34 @@ def test_paste_restores_clipboard_and_cleans_up_copyq_after_successful_autopaste
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}"),
):
with patch("app.paste_service.shutil.which", return_value="/usr/bin/ydotool"):
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,
"_cleanup_copyq",
side_effect=lambda value: calls.append(f"cleanup:{value}"),
"_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", "cleanup:neu"]

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:
with patch.object(service, "_cleanup_copyq") as cleanup_mock:
service.paste("neu", force_autopaste=False)
with patch("app.paste_service.shutil.which", return_value="/usr/bin/ydotool"):
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:
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()
Expand All @@ -344,13 +355,14 @@ def test_paste_does_not_restore_or_cleanup_when_autopaste_disabled(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:
with patch.object(service, "_cleanup_copyq") as cleanup_mock:
service.paste("neu", force_autopaste=True)
with patch("app.paste_service.shutil.which", return_value="/usr/bin/ydotool"):
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:
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()
Expand All @@ -360,13 +372,14 @@ def test_paste_does_not_restore_or_cleanup_when_ydotool_did_not_paste(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:
with patch.object(service, "_cleanup_copyq") as cleanup_mock:
service.paste("neu", force_autopaste=True)
with patch("app.paste_service.shutil.which", return_value="/usr/bin/ydotool"):
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:
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()
Expand Down