diff --git a/README.md b/README.md index 8169076..1d36007 100644 --- a/README.md +++ b/README.md @@ -29,8 +29,15 @@ systemctl --user enable --now ydotool # needed for auto-paste dikte # the settings window opens on first run ``` -`install.sh` adds the `dikte` command, a menu entry, an autostart entry and the -KDE shortcut. +On Ubuntu/GNOME X11, recording uses PulseAudio and clipboard/paste use the X11 +tools instead: + +```sh +sudo apt install pulseaudio-utils xclip xdotool ffmpeg +``` + +`install.sh` adds the `dikte` command, a menu entry and an autostart entry. The +settings window installs a GNOME or KDE global shortcut. Two keys go in the settings window: **OpenAI** and **OpenRouter**. Speech to text runs on either one (`gpt-4o-transcribe` by default), cleanup always on diff --git a/README.tr.md b/README.tr.md index de751b0..b6d3917 100644 --- a/README.tr.md +++ b/README.tr.md @@ -29,8 +29,15 @@ systemctl --user enable --now ydotool # otomatik yapıştırma için dikte # ilk açılışta ayarlar penceresi gelir ``` -`install.sh` `dikte` komutunu, menü girdisini, oturum açılışında otomatik -başlatmayı ve KDE kısayolunu kurar. +Ubuntu/GNOME X11 için kayıt PulseAudio üzerinden, pano ve yapıştırma ise X11 +araçlarıyla çalışır: + +```sh +sudo apt install pulseaudio-utils xclip xdotool ffmpeg +``` + +`install.sh` `dikte` komutunu, menü girdisini ve oturum açılışında otomatik +başlatmayı kurar. Ayarlar penceresi GNOME veya KDE global kısayolunu kurar. Ayarlar penceresinde iki anahtar istenir: **OpenAI** ve **OpenRouter**. Sesi yazıya çevirme ikisinden birinde çalışır (varsayılan `gpt-4o-transcribe`), diff --git a/audio.py b/audio.py index a9f47ed..0ef5038 100644 --- a/audio.py +++ b/audio.py @@ -27,11 +27,12 @@ CHANNELS = 1 SAMPLE_WIDTH = 2 # s16 CHUNK_FRAMES = 1024 CHUNK_BYTES = CHUNK_FRAMES * SAMPLE_WIDTH * CHANNELS +CHUNK_LATENCY_MS = round(CHUNK_FRAMES / RATE * 1000) MIN_FRAMES = int(RATE * 0.25) class Recorder(QObject): - """Runs pw-record as a child process and reads raw PCM from its stdout.""" + """Runs the available sound-server recorder and reads raw PCM from stdout.""" level = pyqtSignal(float) # 0.0 - 1.0, for the waveform stopped = pyqtSignal(str, float, object) # wav path, duration (s), per-chunk RMS @@ -44,6 +45,7 @@ class Recorder(QObject): self._buffer = bytearray() self._rms = [] self._cancelled = False + self._stopping = False self._lock = threading.Lock() @property @@ -53,21 +55,13 @@ class Recorder(QObject): def start(self, target="", max_seconds=300): if self.active: return - if not shutil.which("pw-record"): - self.failed.emit(t("pw-record not found. Is pipewire-audio installed?")) + cmd = recording_command(target) + if not cmd: + self.failed.emit(t( + "No audio recorder found. Install pulseaudio-utils or pipewire-audio." + )) return - cmd = [ - "pw-record", - "--raw", - f"--rate={RATE}", - f"--channels={CHANNELS}", - "--format=s16", - ] - if target: - cmd.append(f"--target={target}") - cmd.append("-") - try: self._proc = subprocess.Popen( cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, bufsize=0 @@ -79,12 +73,14 @@ class Recorder(QObject): self._buffer = bytearray() self._rms = [] self._cancelled = False + self._stopping = False self._max_bytes = int(max_seconds * RATE * SAMPLE_WIDTH * CHANNELS) self._thread = threading.Thread(target=self._pump, daemon=True) self._thread.start() def _pump(self): - stdout = self._proc.stdout + proc = self._proc + stdout = proc.stdout try: while True: chunk = stdout.read(CHUNK_BYTES) @@ -101,8 +97,25 @@ class Recorder(QObject): break except (OSError, ValueError): pass + # Nobody asked it to end and it captured nothing: the recorder is not + # installed properly, or the device was refused. Said out loud here, + # because stop() would otherwise report it as a recording that was too + # short, which sends the user looking in the wrong place. + with self._lock: + captured = bool(self._buffer) + if self._stopping or self._cancelled or captured: + return + try: + detail = proc.stderr.read().decode("utf-8", "replace").strip() + except (AttributeError, OSError): + detail = "" + self.failed.emit(t( + "Audio recorder stopped before receiving sound: {error}", + error=detail or f"exit code {proc.returncode}", + )) def _terminate(self): + self._stopping = True proc = self._proc if proc and proc.poll() is None: try: @@ -161,6 +174,39 @@ def write_wav(pcm, rate=RATE, channels=CHANNELS, width=SAMPLE_WIDTH): return path +def recording_command(target=""): + """Return a raw-s16 capture command for the sound server on this desktop. + + parec works with both PulseAudio and PipeWire's PulseAudio compatibility + service, and its source names are the same ones shown by list_sources(). + Keep pw-record as the fallback for minimal native-PipeWire installations. + """ + if shutil.which("parec"): + cmd = [ + "parec", "--record", "--raw", f"--rate={RATE}", + f"--channels={CHANNELS}", "--format=s16le", + # Left alone, parec holds about two seconds before handing anything + # over, and then hands over all of it at once: the level meter sits + # still and jumps, and the tail of a recording can be lost on the + # way out. A chunk of the meter is the unit the rest of this file + # is measured in, so ask for that. + f"--latency-msec={CHUNK_LATENCY_MS}", + ] + if target: + cmd.append(f"--device={target}") + return cmd + if shutil.which("pw-record"): + cmd = [ + "pw-record", "--raw", f"--rate={RATE}", + f"--channels={CHANNELS}", "--format=s16", + ] + if target: + cmd.append(f"--target={target}") + cmd.append("-") + return cmd + return [] + + class MeetingRecorder(QObject): """Microphone and speaker output into one stereo file: left is you, right is everyone else. diff --git a/cli.py b/cli.py index b584458..facb667 100644 --- a/cli.py +++ b/cli.py @@ -702,7 +702,7 @@ def cmd_shortcut(opts): if opts.shortcut == "status": rows = {} for name, (desktop_id, _label, key) in SHORTCUTS.items(): - rows[name] = {"registered": hotkey.kde_shortcut_status(desktop_id), + rows[name] = {"registered": hotkey.shortcut_status(desktop_id), "configured": conf[key]} lines = [f"{name:8} {row['registered'] or '(not installed)':16} " f"setting: {row['configured'] or '(none)'}" @@ -713,7 +713,7 @@ def cmd_shortcut(opts): desktop_id, label, key = SHORTCUTS[opts.which] if opts.shortcut == "remove": - hotkey.remove_kde_shortcut(desktop_id) + hotkey.remove_shortcut(desktop_id) return out(opts, {"ok": True, "removed": opts.which}, f"Removed the {opts.which} shortcut.") @@ -727,7 +727,7 @@ def cmd_shortcut(opts): return fail(opts, f"{combo} is also used by: {', '.join(clashes[:6])}. " "Pass --force to install it anyway.", 1, conflicts=clashes) - ok, message = hotkey.install_kde_shortcut( + ok, message = hotkey.install_shortcut( combo, ipc.command_for(opts.which), name=label, desktop_id=desktop_id, ) if not ok: diff --git a/hotkey.py b/hotkey.py index 77f1163..b6b3ed1 100644 --- a/hotkey.py +++ b/hotkey.py @@ -1,10 +1,12 @@ -"""Global shortcut: KDE custom-shortcut installation plus a built-in evdev listener.""" +"""GNOME/KDE global-shortcut installation plus a built-in evdev listener.""" +import ast import glob import os import pathlib import re import select +import shutil import struct import subprocess import threading @@ -19,6 +21,8 @@ ASK_DESKTOP_ID = "dikte-ask.desktop" APPLICATIONS_DIR = pathlib.Path.home() / ".local/share/applications" DESKTOP_FILE = APPLICATIONS_DIR / DESKTOP_ID SHORTCUTS_FILE = pathlib.Path.home() / ".config/kglobalshortcutsrc" +GNOME_MEDIA_SCHEMA = "org.gnome.settings-daemon.plugins.media-keys" +GNOME_BINDING_SCHEMA = "org.gnome.settings-daemon.plugins.media-keys.custom-keybinding" # --- evdev key codes (linux/input-event-codes.h) -------------------------- @@ -172,7 +176,154 @@ class EvdevHotkey(QObject): return True -# --- KDE custom shortcut -------------------------------------------------- +# --- the desktop's own shortcut ------------------------------------------- + +def _gnome(): + desktop = os.environ.get("XDG_CURRENT_DESKTOP", "").lower() + return "gnome" in desktop and shutil.which("gsettings") is not None + + +def _gnome_path(desktop_id): + name = re.sub(r"[^a-zA-Z0-9_-]+", "-", desktop_id.removesuffix(".desktop")) + return f"/org/gnome/settings-daemon/plugins/media-keys/custom-keybindings/{name}/" + + +def gnome_accelerator(shortcut): + """Translate Qt-style Ctrl+Alt+A into GNOME's a syntax.""" + parts = [part.strip() for part in str(shortcut).split("+") if part.strip()] + modifiers = [] + key = "" + names = { + "ctrl": "", "control": "", + "alt": "", "shift": "", + "super": "", "meta": "", + } + for part in parts: + modifier = names.get(part.lower()) + if modifier: + if modifier not in modifiers: + modifiers.append(modifier) + else: + key = part.lower() if len(part) == 1 else part + return "".join(modifiers) + key if key else "" + + +def display_accelerator(accelerator): + """Translate a GNOME accelerator back to the form shown in Dikte.""" + text = str(accelerator) + parts = [] + for token, label in (("", "Ctrl"), ("", "Ctrl"), + ("", "Alt"), ("", "Shift"), + ("", "Super")): + if token.lower() in text.lower(): + parts.append(label) + text = re.sub(re.escape(token), "", text, flags=re.IGNORECASE) + key = text.strip() + if len(key) == 1: + key = key.upper() + if key: + parts.append(key) + return "+".join(parts) + + +def _gsettings(*args, check=True): + return subprocess.run( + ["gsettings", *args], capture_output=True, text=True, timeout=10, check=check, + ) + + +def _gsettings_array(value): + """Parse a gsettings string-array, including the empty `@as []` form.""" + text = str(value).strip() + if text.startswith("@as "): + text = text[4:].strip() + parsed = ast.literal_eval(text) if text else [] + if not isinstance(parsed, (list, tuple)): + raise ValueError(f"not a string array: {value}") + return list(parsed) + + +def install_gnome_shortcut(shortcut, exec_command, + name="Dikte: start/stop recording", + desktop_id=DESKTOP_ID): + path = _gnome_path(desktop_id) + try: + current = _gsettings( + "get", GNOME_MEDIA_SCHEMA, "custom-keybindings" + ).stdout.strip() + paths = _gsettings_array(current) + if path not in paths: + paths.append(path) + _gsettings("set", GNOME_MEDIA_SCHEMA, "custom-keybindings", repr(paths)) + schema = f"{GNOME_BINDING_SCHEMA}:{path}" + _gsettings("set", schema, "name", repr(name)) + _gsettings("set", schema, "command", repr(exec_command)) + accelerator = gnome_accelerator(shortcut) + if not accelerator: + raise ValueError(t("Could not parse the shortcut: {shortcut}", + shortcut=shortcut)) + _gsettings("set", schema, "binding", repr(accelerator)) + except (ValueError, SyntaxError, subprocess.SubprocessError, OSError) as exc: + return False, t("Could not register the GNOME shortcut: {error}", error=exc) + return True, t("Shortcut saved: {shortcut}", shortcut=shortcut) + + +def remove_gnome_shortcut(desktop_id=DESKTOP_ID): + path = _gnome_path(desktop_id) + try: + current = _gsettings( + "get", GNOME_MEDIA_SCHEMA, "custom-keybindings" + ).stdout.strip() + paths = _gsettings_array(current) + if path in paths: + paths.remove(path) + _gsettings("set", GNOME_MEDIA_SCHEMA, "custom-keybindings", repr(paths)) + except (ValueError, SyntaxError, subprocess.SubprocessError, OSError): + pass + + +def gnome_shortcut_status(desktop_id=DESKTOP_ID): + path = _gnome_path(desktop_id) + try: + current = _gsettings( + "get", GNOME_MEDIA_SCHEMA, "custom-keybindings" + ).stdout.strip() + paths = _gsettings_array(current) + if path not in paths: + return None + value = _gsettings( + "get", f"{GNOME_BINDING_SCHEMA}:{path}", "binding" + ).stdout.strip() + accelerator = ast.literal_eval(value) + return display_accelerator(accelerator) if accelerator else None + except (ValueError, SyntaxError, subprocess.SubprocessError, OSError): + return None + + +def install_shortcut(shortcut, exec_command, name="Dikte: start/stop recording", + desktop_id=DESKTOP_ID): + if _gnome(): + return install_gnome_shortcut(shortcut, exec_command, name, desktop_id) + return install_kde_shortcut(shortcut, exec_command, name, desktop_id) + + +def remove_shortcut(desktop_id=DESKTOP_ID): + if _gnome(): + remove_gnome_shortcut(desktop_id) + else: + remove_kde_shortcut(desktop_id) + + +def shortcut_status(desktop_id=DESKTOP_ID): + return (gnome_shortcut_status(desktop_id) if _gnome() + else kde_shortcut_status(desktop_id)) + + +def desktop_name(): + return "GNOME" if _gnome() else "KDE" + + +# --- KDE ------------------------------------------------------------------ def install_kde_shortcut(shortcut, exec_command, name="Dikte: start/stop recording", desktop_id=DESKTOP_ID): diff --git a/i18n.py b/i18n.py index f2b1a70..dd70837 100644 --- a/i18n.py +++ b/i18n.py @@ -101,19 +101,22 @@ TR = { "Unexpected error: {error}": "Beklenmeyen hata: {error}", # --- audio / paste errors ----------------------------------------- - "pw-record not found. Is pipewire-audio installed?": - "pw-record bulunamadı. pipewire-audio kurulu mu?", "Could not start recording: {error}": "Kayıt başlatılamadı: {error}", - "wl-copy not found. Install wl-clipboard.": - "wl-copy bulunamadı. wl-clipboard paketini kur.", + "No audio recorder found. Install pulseaudio-utils or pipewire-audio.": + "Ses kayıt aracı bulunamadı. pulseaudio-utils ya da pipewire-audio kur.", + "Audio recorder stopped before receiving sound: {error}": + "Ses kayıt aracı veri alamadan kapandı: {error}", "Could not copy to clipboard: {error}": "Panoya kopyalanamadı: {error}", - "wl-copy exited with code {code}.": "wl-copy {code} koduyla çıktı.", - "ydotool not found, cannot paste automatically.": - "ydotool bulunamadı, otomatik yapıştırma yapılamıyor.", + "{tool} not found. Install {packages}.": + "{tool} bulunamadı. {packages} paketlerini kur.", + "{tool} exited with code {code}.": "{tool} {code} koduyla çıktı.", + "{tool} not found, cannot paste automatically.": + "{tool} bulunamadı, otomatik yapıştırma yapılamıyor.", "Unknown key: {key}": "Bilinmeyen tuş: {key}", - "Could not run ydotool: {error}": "ydotool çalıştırılamadı: {error}", - "ydotool failed: {error}\nIs ydotoold running? (systemctl --user status ydotool)": - "ydotool hatası: {error}\nydotoold çalışıyor mu? (systemctl --user status ydotool)", + "Could not run {tool}: {error}": "{tool} çalıştırılamadı: {error}", + "{tool} failed: {error}": "{tool} hatası: {error}", + "Is ydotoold running? (systemctl --user status ydotool)": + "ydotoold çalışıyor mu? (systemctl --user status ydotool)", # --- api errors ---------------------------------------------------- "{service} API key is empty. Add it in Settings.": @@ -268,9 +271,19 @@ TR = { # --- settings: shortcut ------------------------------------------------ "Install as a KDE shortcut": "KDE kısayolu olarak kur", + "Install as a global shortcut": "Global kısayol olarak kur", "Remove": "Kaldır", "Registered in KDE: {shortcut}": "KDE'de kayıtlı: {shortcut}", "No KDE shortcut installed.": "KDE kısayolu kurulu değil.", + "Registered in {desktop}: {shortcut}": "{desktop}'da kayıtlı: {shortcut}", + "No global shortcut installed.": "Global kısayol kurulu değil.", + "No global shortcut installed. The tray menu starts a meeting too.": + "Global kısayol kurulu değil. Toplantı tepsi menüsünden de başlatılabilir.", + "No global shortcut installed. The tray menu asks it too.": + "Global kısayol kurulu değil. Tepsi menüsünden de soru sorulabilir.", + "Shortcut saved: {shortcut}": "Kısayol kaydedildi: {shortcut}", + "Could not register the GNOME shortcut: {error}": + "GNOME kısayolu kaydedilemedi: {error}", "Use the built-in listener (/dev/input), for when the KDE shortcut is not active yet": "Yerleşik dinleyici kullan (/dev/input), KDE kısayolu henüz etkin değilken", "Works immediately, no session restart. The only difference: the key " diff --git a/install.sh b/install.sh index 8aef023..2cbaaee 100755 --- a/install.sh +++ b/install.sh @@ -19,21 +19,33 @@ echo "────────────────" # 1. Dependencies ---------------------------------------------------------- missing=() -for cmd in pw-record wl-copy wl-paste ydotool ffmpeg; do +audio_cmds=(ffmpeg) +if command -v parec >/dev/null || command -v pw-record >/dev/null; then + : +else + missing+=("pulseaudio-utils-or-pipewire-audio") +fi +if [[ "${XDG_SESSION_TYPE:-}" == "x11" ]]; then + desktop_cmds=(xclip xdotool) +else + desktop_cmds=(wl-copy wl-paste ydotool) +fi +for cmd in "${audio_cmds[@]}" "${desktop_cmds[@]}"; do command -v "$cmd" >/dev/null || missing+=("$cmd") done python3 -c 'import PyQt6.QtWidgets' 2>/dev/null || missing+=("python-pyqt6") if ((${#missing[@]})); then warn "Missing: ${missing[*]}" - say "Arch/CachyOS: sudo pacman -S --needed pipewire-audio wl-clipboard ydotool ffmpeg python-pyqt6" + say "Ubuntu X11: sudo apt install pulseaudio-utils xclip xdotool ffmpeg" + say "Arch Wayland: sudo pacman -S --needed pipewire-audio wl-clipboard ydotool ffmpeg python-pyqt6" echo else ok "All dependencies present" fi # 2. ydotoold -------------------------------------------------------------- -if command -v ydotool >/dev/null; then +if [[ "${XDG_SESSION_TYPE:-}" != "x11" ]] && command -v ydotool >/dev/null; then if systemctl --user is-active --quiet ydotool 2>/dev/null \ || systemctl --user is-active --quiet ydotoold 2>/dev/null; then ok "ydotoold is running (auto-paste ready)" @@ -87,7 +99,10 @@ Type=Application X-KDE-GlobalAccel-CommandShortcut=true EOF -if command -v kwriteconfig6 >/dev/null; then +if [[ "${XDG_CURRENT_DESKTOP:-}" == *GNOME* || "${XDG_CURRENT_DESKTOP:-}" == *gnome* ]]; then + ok "GNOME detected" + say "Open Dikte Settings > Shortcut to install the global shortcut." +elif command -v kwriteconfig6 >/dev/null; then kwriteconfig6 --notify --file kglobalshortcutsrc \ --group services --group dikte-toggle.desktop \ --key _launch "$SHORTCUT" @@ -96,10 +111,10 @@ if command -v kwriteconfig6 >/dev/null; then say "next login. Until then open Settings → Shortcut and turn on the" say "built-in listener to use it right away." else - warn "kwriteconfig6 not found. Add the shortcut via System Settings > Shortcuts" + warn "No supported shortcut manager found. Add the shortcut in desktop settings." fi echo ok "Done. Start it with: dikte" -say "The settings window opens on first run; add your OpenAI and OpenRouter keys." +say "The settings window opens on first run; add an OpenAI, Groq or OpenRouter key." echo diff --git a/paste.py b/paste.py index 2065d56..29b3a5b 100644 --- a/paste.py +++ b/paste.py @@ -1,37 +1,119 @@ -"""Clipboard (wl-clipboard) and key injection (ydotool).""" +"""Clipboard and key injection, through whichever pair of programs is here. +A Wayland session has wl-clipboard and ydotool, an X11 one has xclip and +xdotool, and a session is one or the other. Which it is gets decided in one +place, and each desktop is a small group of functions below it: another desktop, +or another operating system, adds a group and a line to the chooser rather than +a branch inside every function here. +""" + +import collections +import os import shutil import subprocess import time from i18n import t -# Linux input event codes (linux/input-event-codes.h) +# Linux input event codes (linux/input-event-codes.h), which is what ydotool +# takes. They are also the list of keys a paste shortcut may be built from, so +# xdotool is held to the same table rather than being handed the text as typed. KEYCODES = { "ctrl": 29, "control": 29, "shift": 42, "alt": 56, "super": 125, "meta": 125, "v": 47, "insert": 110, "enter": 28, "return": 28, } +# xdotool speaks X keysyms, which spell some of those differently. +KEYSYMS = {"control": "ctrl", "meta": "super", "insert": "Insert", + "enter": "Return", "return": "Return"} + class PasteError(Exception): pass +def _keys(shortcut): + """'Ctrl+V' -> ['ctrl', 'v'], every one of them a key we know.""" + parts = [key.strip().lower() for key in str(shortcut).split("+") if key.strip()] + for key in parts: + if key not in KEYCODES: + raise PasteError(t("Unknown key: {key}", key=key)) + return parts + + +def _ydotool_command(shortcut): + """ydotool wants a press event per key, then a release in reverse.""" + codes = [KEYCODES[key] for key in _keys(shortcut)] + return ["ydotool", "key", *[f"{code}:1" for code in codes], + *[f"{code}:0" for code in reversed(codes)]] + + +def _xdotool_command(shortcut): + """xdotool takes the whole combination as one argument.""" + keys = [KEYSYMS.get(key, key) for key in _keys(shortcut)] + return ["xdotool", "key", "--clearmodifiers", "+".join(keys)] + + +Desktop = collections.namedtuple( + "Desktop", + # The two programs, the packages to install them from, how to build the key + # press, and what else to say when the key press fails. + "clipboard keyboard packages read_command copy_command key_command key_hint", +) + +WAYLAND = Desktop( + clipboard="wl-copy", + keyboard="ydotool", + packages="wl-clipboard and ydotool", + read_command=["wl-paste", "--no-newline"], + copy_command=["wl-copy"], + key_command=_ydotool_command, + key_hint="Is ydotoold running? (systemctl --user status ydotool)", +) + +X11 = Desktop( + clipboard="xclip", + keyboard="xdotool", + packages="xclip and xdotool", + read_command=["xclip", "-selection", "clipboard", "-out"], + copy_command=["xclip", "-selection", "clipboard", "-in"], + key_command=_xdotool_command, + key_hint="", +) + + +def desktop(): + """The pair of programs this session's clipboard and keyboard go through. + + Read every time rather than settled at import: a session started before the + display server was up would otherwise be stuck with the wrong answer, and a + test would have nowhere to say which one it means. + """ + if os.environ.get("XDG_SESSION_TYPE") == "x11": + return X11 + if os.environ.get("DISPLAY") and not os.environ.get("WAYLAND_DISPLAY"): + return X11 + return WAYLAND + + +# --- the clipboard --------------------------------------------------------- + def read_clipboard(): - if not shutil.which("wl-paste"): + here = desktop() + if not shutil.which(here.read_command[0]): return None try: - res = subprocess.run(["wl-paste", "--no-newline"], capture_output=True, timeout=5) + res = subprocess.run(here.read_command, capture_output=True, timeout=5) except (subprocess.SubprocessError, OSError): return None return res.stdout if res.returncode == 0 else None -def _run_wl_copy(payload): - """wl-copy forks to keep owning the selection; leaving its pipes open makes - subprocess.run wait for EOF forever, hence DEVNULL.""" +def _run_copy(payload): + """The clipboard owner forks to keep holding the selection; leaving its + pipes open makes subprocess.run wait for EOF forever, hence DEVNULL.""" return subprocess.run( - ["wl-copy"], + desktop().copy_command, input=payload, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, @@ -40,51 +122,50 @@ def _run_wl_copy(payload): def copy(text): - if not shutil.which("wl-copy"): - raise PasteError(t("wl-copy not found. Install wl-clipboard.")) + here = desktop() + if not shutil.which(here.clipboard): + raise PasteError(t("{tool} not found. Install {packages}.", + tool=here.clipboard, packages=here.packages)) try: - res = _run_wl_copy(text.encode("utf-8")) + res = _run_copy(text.encode("utf-8")) except (subprocess.SubprocessError, OSError) as exc: raise PasteError(t("Could not copy to clipboard: {error}", error=exc)) from exc if res.returncode != 0: - raise PasteError(t("wl-copy exited with code {code}.", code=res.returncode)) + raise PasteError(t("{tool} exited with code {code}.", + tool=here.clipboard, code=res.returncode)) def copy_bytes(data): - if data is None or not shutil.which("wl-copy"): + if data is None or not shutil.which(desktop().clipboard): return try: - _run_wl_copy(data) + _run_copy(data) except (subprocess.SubprocessError, OSError): pass -def ydotool_ready(): - return shutil.which("ydotool") is not None +# --- the key press --------------------------------------------------------- + +def paste_ready(): + return shutil.which(desktop().keyboard) is not None def press(shortcut="ctrl+v", delay=0.12): - """Press a key combination through ydotool, e.g. 'ctrl+v'.""" - if not ydotool_ready(): - raise PasteError(t("ydotool not found, cannot paste automatically.")) + """Press a key combination, e.g. 'ctrl+v'.""" + here = desktop() + if not paste_ready(): + raise PasteError(t("{tool} not found, cannot paste automatically.", + tool=here.keyboard)) - codes = [] - for key in (k.strip().lower() for k in shortcut.split("+") if k.strip()): - code = KEYCODES.get(key) - if code is None: - raise PasteError(t("Unknown key: {key}", key=key)) - codes.append(code) - - seq = [f"{c}:1" for c in codes] + [f"{c}:0" for c in reversed(codes)] + command = here.key_command(shortcut) time.sleep(delay) # let the selection settle and focus come back try: - res = subprocess.run(["ydotool", "key", *seq], capture_output=True, - text=True, timeout=10) + res = subprocess.run(command, capture_output=True, text=True, timeout=10) except (subprocess.SubprocessError, OSError) as exc: - raise PasteError(t("Could not run ydotool: {error}", error=exc)) from exc + raise PasteError(t("Could not run {tool}: {error}", + tool=here.keyboard, error=exc)) from exc if res.returncode != 0: - raise PasteError(t( - "ydotool failed: {error}\nIs ydotoold running? " - "(systemctl --user status ydotool)", - error=res.stderr.strip() or "unknown error", - )) + message = t("{tool} failed: {error}", tool=here.keyboard, + error=res.stderr.strip() or "unknown error") + raise PasteError(f"{message}\n{t(here.key_hint)}" if here.key_hint + else message) diff --git a/settings_ui.py b/settings_ui.py index 9638ae8..1690a6b 100644 --- a/settings_ui.py +++ b/settings_ui.py @@ -1289,7 +1289,7 @@ class SettingsWindow(QDialog): ) if answer != QMessageBox.StandardButton.Yes: return - ok, message = hotkey.install_kde_shortcut(combo, self.launch_command) + ok, message = hotkey.install_shortcut(combo, self.launch_command) QMessageBox.information(self, t("Shortcut"), message) if ok: self.conf["shortcut"] = combo @@ -1297,14 +1297,15 @@ class SettingsWindow(QDialog): self._refresh_shortcut_status() def _remove_shortcut(self): - hotkey.remove_kde_shortcut() + hotkey.remove_shortcut() self._refresh_shortcut_status() def _refresh_shortcut_status(self): - current = hotkey.kde_shortcut_status() + current = hotkey.shortcut_status() self.shortcut_status.setText( - t("Registered in KDE: {shortcut}", shortcut=current) if current - else t("No KDE shortcut installed.") + t("Registered in {desktop}: {shortcut}", + desktop=hotkey.desktop_name(), shortcut=current) if current + else t("No global shortcut installed.") ) def _install_meeting_shortcut(self): @@ -1322,7 +1323,7 @@ class SettingsWindow(QDialog): ) if answer != QMessageBox.StandardButton.Yes: return - ok, message = hotkey.install_kde_shortcut( + ok, message = hotkey.install_shortcut( combo, self.meeting_command, name="Dikte: start/end a meeting recording", desktop_id=hotkey.MEETING_DESKTOP_ID, ) @@ -1333,14 +1334,15 @@ class SettingsWindow(QDialog): self._refresh_meeting_shortcut_status() def _remove_meeting_shortcut(self): - hotkey.remove_kde_shortcut(hotkey.MEETING_DESKTOP_ID) + hotkey.remove_shortcut(hotkey.MEETING_DESKTOP_ID) self._refresh_meeting_shortcut_status() def _refresh_meeting_shortcut_status(self): - current = hotkey.kde_shortcut_status(hotkey.MEETING_DESKTOP_ID) + current = hotkey.shortcut_status(hotkey.MEETING_DESKTOP_ID) self.meeting_shortcut_status.setText( - t("Registered in KDE: {shortcut}", shortcut=current) if current - else t("No KDE shortcut installed. The tray menu starts a meeting too.") + t("Registered in {desktop}: {shortcut}", + desktop=hotkey.desktop_name(), shortcut=current) if current + else t("No global shortcut installed. The tray menu starts a meeting too.") ) # ---- Claude ---------------------------------------------------------- @@ -1360,7 +1362,7 @@ class SettingsWindow(QDialog): ) if answer != QMessageBox.StandardButton.Yes: return - ok, message = hotkey.install_kde_shortcut( + ok, message = hotkey.install_shortcut( combo, self.ask_command, name="Dikte: ask Claude Code", desktop_id=hotkey.ASK_DESKTOP_ID, ) @@ -1371,14 +1373,15 @@ class SettingsWindow(QDialog): self._refresh_ask_shortcut_status() def _remove_ask_shortcut(self): - hotkey.remove_kde_shortcut(hotkey.ASK_DESKTOP_ID) + hotkey.remove_shortcut(hotkey.ASK_DESKTOP_ID) self._refresh_ask_shortcut_status() def _refresh_ask_shortcut_status(self): - current = hotkey.kde_shortcut_status(hotkey.ASK_DESKTOP_ID) + current = hotkey.shortcut_status(hotkey.ASK_DESKTOP_ID) self.assistant_shortcut_status.setText( - t("Registered in KDE: {shortcut}", shortcut=current) if current - else t("No KDE shortcut installed. The tray menu asks it too.") + t("Registered in {desktop}: {shortcut}", + desktop=hotkey.desktop_name(), shortcut=current) if current + else t("No global shortcut installed. The tray menu asks it too.") ) def _assistant_provider_changed(self): diff --git a/tests/test_audio.py b/tests/test_audio.py index 9a8dbaa..7595d1b 100644 --- a/tests/test_audio.py +++ b/tests/test_audio.py @@ -7,6 +7,7 @@ speakers, and neither list may go missing when pactl is absent. import array import contextlib +import io import json import os import subprocess @@ -194,10 +195,10 @@ class FakeProcess: """A pw-record that hands over a fixed buffer and then ends.""" def __init__(self, data): - import io self.stdout = io.BytesIO(data) self.stderr = io.BytesIO(b"") self.signals = [] + self.returncode = 0 self._alive = True def poll(self): @@ -215,6 +216,60 @@ class FakeProcess: self._alive = False +@linux_only +class RecordingCommand(DikteTest): + """Which program captures the microphone, and how it is asked to.""" + + def test_parec_is_preferred(self): + """It speaks to PulseAudio and to PipeWire's compatibility service, so + it is the one that works on both desktops.""" + with only_these_tools("parec", "pw-record"): + self.assertEqual(audio.recording_command()[0], "parec") + + def test_pw_record_is_the_fallback(self): + with only_these_tools("pw-record"): + self.assertEqual(audio.recording_command()[0], "pw-record") + + def test_neither_is_installed(self): + with only_these_tools(): + self.assertEqual(audio.recording_command(), []) + + def test_both_capture_the_format_the_rest_of_the_code_expects(self): + for tool in ("parec", "pw-record"): + with self.subTest(tool=tool), only_these_tools(tool): + cmd = audio.recording_command() + joined = " ".join(cmd) + self.assertIn(str(audio.RATE), joined) + self.assertIn(str(audio.CHANNELS), joined) + self.assertIn("s16", joined) + + def test_parec_is_asked_for_the_level_meter_s_own_chunk(self): + """Left alone it buffers about two seconds, which the waveform shows as + a still bar that jumps once a second, and which can cost the tail of a + recording when the process is asked to stop.""" + with only_these_tools("parec"): + self.assertIn(f"--latency-msec={audio.CHUNK_LATENCY_MS}", + audio.recording_command()) + + def test_the_latency_asked_for_is_the_chunk_the_meter_reads(self): + self.assertEqual(audio.CHUNK_LATENCY_MS, + round(audio.CHUNK_FRAMES / audio.RATE * 1000)) + + def test_a_chosen_microphone_reaches_either_one(self): + with only_these_tools("parec"): + self.assertIn("--device=alsa_input.usb", audio.recording_command( + "alsa_input.usb")) + with only_these_tools("pw-record"): + self.assertIn("--target=alsa_input.usb", audio.recording_command( + "alsa_input.usb")) + + def test_no_microphone_named_means_no_device_flag(self): + for tool, flag in (("parec", "--device="), ("pw-record", "--target=")): + with self.subTest(tool=tool), only_these_tools(tool): + self.assertFalse([arg for arg in audio.recording_command() + if arg.startswith(flag)]) + + @linux_only class RecorderChain(DikteTest): """Start to WAV, with pw-record faked out.""" @@ -233,15 +288,6 @@ class RecorderChain(DikteTest): recorder.stop() return recorder, results, failures, popen - def test_pw_record_is_not_installed(self): - recorder = audio.Recorder() - failures = [] - recorder.failed.connect(failures.append) - with only_these_tools(): - recorder.start() - self.assertEqual(len(failures), 1) - self.assertIn("pipewire", failures[0]) - def test_the_capture_format_is_what_the_rest_of_the_code_expects(self): _, _, _, popen = self.record(silence(1.0)) cmd = popen.call_args.args[0] @@ -295,6 +341,58 @@ class RecorderChain(DikteTest): self.addCleanup(os.unlink, path) self.assertLessEqual(duration, 1.1) + def test_a_recorder_that_is_not_installed_at_all(self): + recorder = audio.Recorder() + failures = [] + recorder.failed.connect(failures.append) + with only_these_tools(): + recorder.start() + self.assertEqual(len(failures), 1) + self.assertIn("pulseaudio-utils", failures[0]) + + def pump(self, data=b"", stderr=b"", stopping=False, cancelled=False): + """Run the pump in this thread, where a queued signal would need an + event loop nobody is running here.""" + recorder = audio.Recorder() + failures = [] + recorder.failed.connect(failures.append) + proc = FakeProcess(data) + proc.stderr = io.BytesIO(stderr) + proc._alive = False + recorder._proc = proc + recorder._max_bytes = 10 ** 9 + recorder._stopping = stopping + recorder._cancelled = cancelled + recorder._pump() + return failures + + def test_a_recorder_that_died_on_its_own_says_so(self): + """parec refused the device, or the sound server went away.""" + failures = self.pump(stderr=b"connection refused\n") + self.assertEqual(len(failures), 1) + self.assertIn("connection refused", failures[0]) + + def test_a_death_with_nothing_on_stderr_still_names_the_exit_code(self): + failures = self.pump() + self.assertIn("exit code", failures[0]) + + def test_a_recording_we_ended_ourselves_is_not_a_death(self): + """Otherwise a stray keypress produces two errors, and the first one + sends the user looking for a broken sound server.""" + self.assertEqual(self.pump(stopping=True), []) + + def test_a_cancelled_recording_is_not_a_death(self): + self.assertEqual(self.pump(cancelled=True), []) + + def test_a_recorder_that_captured_something_first_is_not_a_death(self): + self.assertEqual(self.pump(data=silence(0.5)), []) + + def test_a_short_recording_reports_only_that(self): + _, results, failures, _ = self.record(silence(0.1)) + self.assertEqual(results, []) + self.assertEqual(len(failures), 1) + self.assertIn("0.3", failures[0]) + def test_a_recorder_that_could_not_start(self): recorder = audio.Recorder() failures = [] diff --git a/tests/test_hotkey.py b/tests/test_hotkey.py index b3e0d70..4297724 100644 --- a/tests/test_hotkey.py +++ b/tests/test_hotkey.py @@ -1,5 +1,7 @@ -"""Parsing a shortcut, and the KDE entry it is installed as.""" +"""Parsing a shortcut, and the entry the desktop is asked to register it as.""" +import contextlib +import os import subprocess import unittest from unittest import mock @@ -121,6 +123,193 @@ class Bindings(DikteTest): self.assertEqual(len(listener._bindings[57]), 2) +@linux_only +class Chooser(DikteTest): + """Which desktop is asked to register the shortcut.""" + + @contextlib.contextmanager + def under(self, desktop, has_gsettings=True): + """A session that says it is this desktop, with or without gsettings.""" + with mock.patch.dict(os.environ, {"XDG_CURRENT_DESKTOP": desktop}), \ + mock.patch.object(hotkey.shutil, "which", + return_value="/usr/bin/gsettings" + if has_gsettings else None): + yield + + def test_gnome_when_the_session_says_so_and_gsettings_is_there(self): + with self.under("GNOME"): + self.assertEqual(hotkey.desktop_name(), "GNOME") + + def test_kde_otherwise(self): + with self.under("KDE"): + self.assertEqual(hotkey.desktop_name(), "KDE") + + def test_a_gnome_session_with_no_gsettings_falls_back(self): + """Nothing to write the binding with, so KDE's file is the only try.""" + with self.under("GNOME", has_gsettings=False): + self.assertEqual(hotkey.desktop_name(), "KDE") + + def test_the_desktop_is_matched_loosely(self): + for desktop in ("GNOME", "ubuntu:GNOME", "gnome"): + with self.subTest(desktop=desktop), self.under(desktop): + self.assertEqual(hotkey.desktop_name(), "GNOME") + + def test_installing_goes_to_whichever_it_is(self): + with self.under("GNOME"), \ + mock.patch.object(hotkey, "install_gnome_shortcut", + return_value=(True, "ok")) as gnome: + hotkey.install_shortcut("Ctrl+Space", "dikte toggle") + gnome.assert_called_once() + + with self.under("KDE"), \ + mock.patch.object(hotkey, "install_kde_shortcut", + return_value=(True, "ok")) as kde: + hotkey.install_shortcut("Ctrl+Space", "dikte toggle") + kde.assert_called_once() + + def test_removing_and_reading_back_go_to_the_same_one(self): + with self.under("GNOME"), \ + mock.patch.object(hotkey, "remove_gnome_shortcut") as remove, \ + mock.patch.object(hotkey, "gnome_shortcut_status", + return_value="Ctrl+Space") as status: + hotkey.remove_shortcut() + self.assertEqual(hotkey.shortcut_status(), "Ctrl+Space") + remove.assert_called_once() + status.assert_called_once() + + +@linux_only +class GnomeAccelerator(DikteTest): + """Qt spells a combination one way, GNOME another.""" + + def test_the_default_shortcut(self): + self.assertEqual(hotkey.gnome_accelerator("Ctrl+Space"), "Space") + + def test_several_modifiers_keep_their_order(self): + self.assertEqual(hotkey.gnome_accelerator("Ctrl+Alt+A"), "a") + + def test_the_synonyms(self): + self.assertEqual(hotkey.gnome_accelerator("Meta+A"), + hotkey.gnome_accelerator("Super+A")) + self.assertEqual(hotkey.gnome_accelerator("Control+A"), + hotkey.gnome_accelerator("Ctrl+A")) + + def test_a_modifier_repeated_is_written_once(self): + self.assertEqual(hotkey.gnome_accelerator("Ctrl+Control+A"), "a") + + def test_modifiers_with_no_key_are_not_a_shortcut(self): + self.assertEqual(hotkey.gnome_accelerator("Ctrl+Alt"), "") + self.assertEqual(hotkey.gnome_accelerator(""), "") + + def test_what_goes_out_comes_back_the_way_dikte_writes_it(self): + for shortcut in ("Ctrl+Space", "Ctrl+Alt+A", "Shift+F9", "Super+M"): + with self.subTest(shortcut=shortcut): + accelerator = hotkey.gnome_accelerator(shortcut) + self.assertEqual(hotkey.display_accelerator(accelerator), shortcut) + + def test_the_control_spelling_gnome_also_uses(self): + self.assertEqual(hotkey.display_accelerator("a"), "Ctrl+A") + + def test_an_empty_binding(self): + self.assertEqual(hotkey.display_accelerator(""), "") + + +@linux_only +class GsettingsArray(DikteTest): + def test_a_list_of_paths(self): + self.assertEqual( + hotkey._gsettings_array("['/org/gnome/one/', '/org/gnome/two/']"), + ["/org/gnome/one/", "/org/gnome/two/"]) + + def test_the_empty_form_gsettings_prints(self): + self.assertEqual(hotkey._gsettings_array("@as []"), []) + + def test_nothing_at_all(self): + self.assertEqual(hotkey._gsettings_array(""), []) + + def test_something_that_is_not_an_array(self): + with self.assertRaises(ValueError): + hotkey._gsettings_array("'just a string'") + + +@linux_only +class GnomeShortcut(DikteTest): + """The gsettings calls, without a session bus to make them against.""" + + def setUp(self): + super().setUp() + self.enterContext(mock.patch.dict(os.environ, + {"XDG_CURRENT_DESKTOP": "GNOME"})) + self.enterContext(mock.patch.object(hotkey.shutil, "which", + return_value="/usr/bin/gsettings")) + + def gsettings(self, listed="@as []", binding="'Space'"): + def run(cmd, **kwargs): + if cmd[1] == "get" and cmd[3] == "custom-keybindings": + return FakeCompleted(stdout=listed) + return FakeCompleted(stdout=binding) + return mock.patch.object(subprocess, "run", side_effect=run) + + def written(self, run, key): + """The value the last `gsettings set ... ` was given.""" + for call in reversed(run.mock_calls): + cmd = call.args[0] if call.args else [] + if len(cmd) > 4 and cmd[1] == "set" and cmd[3] == key: + return cmd[4] + return None + + def test_installing_registers_the_path_the_name_and_the_binding(self): + with self.gsettings() as run: + ok, message = hotkey.install_shortcut("Ctrl+Space", "dikte toggle") + self.assertTrue(ok) + self.assertIn("Ctrl+Space", message) + self.assertIn(hotkey.DESKTOP_ID.removesuffix(".desktop"), + self.written(run, "custom-keybindings")) + self.assertEqual(self.written(run, "command"), repr("dikte toggle")) + self.assertEqual(self.written(run, "binding"), repr("Space")) + + def test_installing_twice_does_not_list_the_path_twice(self): + path = hotkey._gnome_path(hotkey.DESKTOP_ID) + with self.gsettings(listed=repr([path])) as run: + hotkey.install_shortcut("Ctrl+Space", "dikte toggle") + self.assertIsNone(self.written(run, "custom-keybindings")) + + def test_each_verb_gets_its_own_path(self): + self.assertNotEqual(hotkey._gnome_path(hotkey.DESKTOP_ID), + hotkey._gnome_path(hotkey.ASK_DESKTOP_ID)) + + def test_a_shortcut_gnome_cannot_express(self): + with self.gsettings(): + ok, message = hotkey.install_shortcut("Ctrl+Alt", "dikte toggle") + self.assertFalse(ok) + self.assertIn("Ctrl+Alt", message) + + def test_no_session_bus_to_talk_to(self): + with mock.patch.object(subprocess, "run", side_effect=OSError("no bus")): + ok, _ = hotkey.install_shortcut("Ctrl+Space", "dikte toggle") + self.assertFalse(ok) + + def test_reading_back_a_shortcut_that_is_registered(self): + path = hotkey._gnome_path(hotkey.DESKTOP_ID) + with self.gsettings(listed=repr([path])): + self.assertEqual(hotkey.shortcut_status(), "Ctrl+Space") + + def test_reading_back_one_that_is_not(self): + with self.gsettings(listed="@as []"): + self.assertIsNone(hotkey.shortcut_status()) + + def test_removing_takes_the_path_off_the_list(self): + path = hotkey._gnome_path(hotkey.DESKTOP_ID) + with self.gsettings(listed=repr([path, "/org/gnome/other/"])) as run: + hotkey.remove_shortcut() + self.assertEqual(self.written(run, "custom-keybindings"), + repr(["/org/gnome/other/"])) + + def test_removing_one_that_was_never_installed(self): + with self.gsettings(listed="@as []"): + hotkey.remove_shortcut() # must not raise + + @linux_only class KdeShortcut(DikteTest): def setUp(self): diff --git a/tests/test_paste.py b/tests/test_paste.py index 96c258d..1d6806b 100644 --- a/tests/test_paste.py +++ b/tests/test_paste.py @@ -1,12 +1,18 @@ """The clipboard and the key press, which is where a dictation actually lands. Everything here shells out, so the tools are faked. What the tests hold onto is -the command line: a paste that presses the wrong codes, or in the wrong order, +the command line: a paste that presses the wrong keys, or in the wrong order, types nothing and looks like a hang. + +Both desktops owe the same promises, so those are written once and run against +each of them. A third one added to paste.py inherits the same list rather than +needing its own copy of it. """ +import os import subprocess import unittest +from typing import ClassVar from unittest import mock import paste @@ -14,47 +20,86 @@ from tests.support import DikteTest, FakeCompleted, linux_only, only_these_tools @linux_only -class ReadClipboard(DikteTest): - def test_no_wl_paste_installed(self): +class Chooser(DikteTest): + """Which pair of programs this session's clipboard goes through.""" + + def under(self, **env): + with mock.patch.dict(os.environ, env, clear=True): + return paste.desktop() + + def test_a_wayland_session(self): + self.assertIs(self.under(XDG_SESSION_TYPE="wayland", + WAYLAND_DISPLAY="wayland-0"), paste.WAYLAND) + + def test_an_x11_session(self): + self.assertIs(self.under(XDG_SESSION_TYPE="x11", DISPLAY=":0"), paste.X11) + + def test_a_display_with_no_wayland_beside_it(self): + self.assertIs(self.under(DISPLAY=":0"), paste.X11) + + def test_an_x11_display_under_wayland_is_still_wayland(self): + """XWayland sets DISPLAY too; the session type is the one to believe.""" + self.assertIs(self.under(XDG_SESSION_TYPE="wayland", + DISPLAY=":0", WAYLAND_DISPLAY="wayland-0"), + paste.WAYLAND) + + def test_nothing_set_at_all(self): + self.assertIs(self.under(), paste.WAYLAND) + + +class DesktopContract: + """What both desktops owe. Each of them subclasses this once, below.""" + + env: ClassVar[dict] = {} + here = None + + def setUp(self): + super().setUp() + self.enterContext(mock.patch.dict(os.environ, self.env, clear=True)) + + # ---- reading the clipboard ------------------------------------------- + + def test_no_reader_installed(self): with only_these_tools(): self.assertIsNone(paste.read_clipboard()) def test_what_is_on_the_clipboard_comes_back_as_bytes(self): - with only_these_tools("wl-paste"), \ + with only_these_tools(self.here.read_command[0]), \ mock.patch.object(subprocess, "run", return_value=FakeCompleted(stdout=b"hello")) as run: self.assertEqual(paste.read_clipboard(), b"hello") - self.assertEqual(run.call_args.args[0], ["wl-paste", "--no-newline"]) + self.assertEqual(run.call_args.args[0], self.here.read_command) def test_an_empty_clipboard_is_not_an_error(self): - with only_these_tools("wl-paste"), \ + with only_these_tools(self.here.read_command[0]), \ mock.patch.object(subprocess, "run", return_value=FakeCompleted(returncode=1)): self.assertIsNone(paste.read_clipboard()) - def test_a_tool_that_will_not_run(self): - with only_these_tools("wl-paste"), \ + def test_a_reader_that_will_not_run(self): + with only_these_tools(self.here.read_command[0]), \ mock.patch.object(subprocess, "run", side_effect=OSError("nope")): self.assertIsNone(paste.read_clipboard()) + # ---- copying ---------------------------------------------------------- -@linux_only -class Copy(DikteTest): - def test_no_wl_copy_installed(self): + def test_no_clipboard_tool_installed_says_what_to_install(self): with only_these_tools(), self.assertRaises(paste.PasteError) as caught: paste.copy("hello") - self.assertIn("wl-clipboard", str(caught.exception)) + self.assertIn(self.here.clipboard, str(caught.exception)) + self.assertIn(self.here.packages.split(" and ")[0], str(caught.exception)) def test_the_text_goes_in_as_utf8(self): - with only_these_tools("wl-copy"), \ + with only_these_tools(self.here.clipboard), \ mock.patch.object(subprocess, "run", return_value=FakeCompleted()) as run: paste.copy("günaydın") + self.assertEqual(run.call_args.args[0], self.here.copy_command) self.assertEqual(run.call_args.kwargs["input"], "günaydın".encode()) def test_the_pipes_are_closed_so_the_call_can_return(self): - """wl-copy forks and holds the selection; a pipe nobody drains hangs.""" - with only_these_tools("wl-copy"), \ + """The clipboard owner forks; a pipe nobody drains hangs the caller.""" + with only_these_tools(self.here.clipboard), \ mock.patch.object(subprocess, "run", return_value=FakeCompleted()) as run: paste.copy("hello") @@ -62,95 +107,130 @@ class Copy(DikteTest): self.assertEqual(run.call_args.kwargs["stderr"], subprocess.DEVNULL) def test_a_non_zero_exit_is_reported(self): - with only_these_tools("wl-copy"), \ + with only_these_tools(self.here.clipboard), \ mock.patch.object(subprocess, "run", return_value=FakeCompleted(returncode=1)), \ self.assertRaises(paste.PasteError): paste.copy("hello") - def test_a_tool_that_will_not_run(self): - with only_these_tools("wl-copy"), \ + def test_a_clipboard_tool_that_will_not_run(self): + with only_these_tools(self.here.clipboard), \ mock.patch.object(subprocess, "run", side_effect=OSError("nope")), \ self.assertRaises(paste.PasteError): paste.copy("hello") - -@linux_only -class CopyBytes(DikteTest): - def test_nothing_to_restore(self): - with only_these_tools("wl-copy"), \ + def test_there_is_nothing_to_restore(self): + with only_these_tools(self.here.clipboard), \ mock.patch.object(subprocess, "run") as run: paste.copy_bytes(None) run.assert_not_called() def test_restoring_never_raises(self): """It runs after the paste went in; failing here must not undo that.""" - with only_these_tools("wl-copy"), \ + with only_these_tools(self.here.clipboard), \ mock.patch.object(subprocess, "run", side_effect=OSError("nope")): paste.copy_bytes(b"whatever was there before") def test_the_bytes_go_back_untouched(self): - with only_these_tools("wl-copy"), \ + with only_these_tools(self.here.clipboard), \ mock.patch.object(subprocess, "run", return_value=FakeCompleted()) as run: paste.copy_bytes(b"\x89PNG\r\n") self.assertEqual(run.call_args.kwargs["input"], b"\x89PNG\r\n") + # ---- pressing the key ------------------------------------------------- -@linux_only -class Press(DikteTest): - def setUp(self): - super().setUp() - # The settle delay is real time nobody needs to spend in a test. - self.patch_attr(paste.time, "sleep", lambda seconds: None) - - def run_press(self, shortcut, result=None): - with only_these_tools("ydotool"), \ + def press(self, shortcut, result=None): + with only_these_tools(self.here.keyboard), \ + mock.patch.object(paste.time, "sleep", lambda seconds: None), \ mock.patch.object(subprocess, "run", return_value=result or FakeCompleted()) as run: paste.press(shortcut) return run.call_args.args[0] - def test_no_ydotool_installed(self): + def test_no_keyboard_tool_installed(self): with only_these_tools(): - self.assertFalse(paste.ydotool_ready()) - with self.assertRaises(paste.PasteError): + self.assertFalse(paste.paste_ready()) + with self.assertRaises(paste.PasteError) as caught: paste.press() - - def test_ctrl_v_presses_down_then_lets_go_in_reverse(self): - self.assertEqual(self.run_press("ctrl+v"), - ["ydotool", "key", "29:1", "47:1", "47:0", "29:0"]) - - def test_three_keys(self): - self.assertEqual(self.run_press("ctrl+shift+v"), - ["ydotool", "key", "29:1", "42:1", "47:1", - "47:0", "42:0", "29:0"]) + self.assertIn(self.here.keyboard, str(caught.exception)) def test_case_and_spacing_do_not_matter(self): - self.assertEqual(self.run_press(" Ctrl + V "), self.run_press("ctrl+v")) + self.assertEqual(self.press(" Ctrl + V "), self.press("ctrl+v")) - def test_the_synonyms_land_on_the_same_codes(self): - self.assertEqual(self.run_press("control+insert"), - ["ydotool", "key", "29:1", "110:1", "110:0", "29:0"]) - self.assertEqual(self.run_press("super+enter"), self.run_press("meta+return")) - - def test_a_key_nobody_mapped(self): - with only_these_tools("ydotool"), mock.patch.object(subprocess, "run"), \ + def test_a_key_nobody_mapped_is_refused_before_the_tool_runs(self): + """Whichever desktop it is, the shortcut is held to one table.""" + with only_these_tools(self.here.keyboard), \ + mock.patch.object(subprocess, "run") as run, \ self.assertRaises(paste.PasteError) as caught: paste.press("ctrl+f13") self.assertIn("f13", str(caught.exception)) - - def test_ydotoold_not_running_says_so(self): - with self.assertRaises(paste.PasteError) as caught: - self.run_press("ctrl+v", FakeCompleted(returncode=1, stderr="no socket")) - self.assertIn("ydotoold", str(caught.exception)) + run.assert_not_called() def test_a_tool_that_will_not_run(self): - with only_these_tools("ydotool"), \ + with only_these_tools(self.here.keyboard), \ + mock.patch.object(paste.time, "sleep", lambda seconds: None), \ mock.patch.object(subprocess, "run", side_effect=OSError("nope")), \ self.assertRaises(paste.PasteError): paste.press("ctrl+v") + def test_a_failed_key_press_names_the_tool_and_what_it_said(self): + with self.assertRaises(paste.PasteError) as caught: + self.press("ctrl+v", FakeCompleted(returncode=1, stderr="no socket")) + self.assertIn(self.here.keyboard, str(caught.exception)) + self.assertIn("no socket", str(caught.exception)) + + +@linux_only +class Wayland(DesktopContract, DikteTest): + env: ClassVar[dict] = {"XDG_SESSION_TYPE": "wayland", + "WAYLAND_DISPLAY": "wayland-0"} + here = paste.WAYLAND + + def test_ydotool_presses_down_then_lets_go_in_reverse(self): + self.assertEqual(self.press("ctrl+v"), + ["ydotool", "key", "29:1", "47:1", "47:0", "29:0"]) + + def test_three_keys(self): + self.assertEqual(self.press("ctrl+shift+v"), + ["ydotool", "key", "29:1", "42:1", "47:1", + "47:0", "42:0", "29:0"]) + + def test_the_synonyms_land_on_the_same_codes(self): + self.assertEqual(self.press("control+insert"), + ["ydotool", "key", "29:1", "110:1", "110:0", "29:0"]) + self.assertEqual(self.press("super+enter"), self.press("meta+return")) + + def test_a_failure_asks_after_the_daemon(self): + """ydotool needs ydotoold, and says nothing useful when it is not up.""" + with self.assertRaises(paste.PasteError) as caught: + self.press("ctrl+v", FakeCompleted(returncode=1, stderr="no socket")) + self.assertIn("ydotoold", str(caught.exception)) + + +@linux_only +class X11(DesktopContract, DikteTest): + env: ClassVar[dict] = {"XDG_SESSION_TYPE": "x11", "DISPLAY": ":0"} + here = paste.X11 + + def test_xdotool_takes_the_combination_as_one_argument(self): + self.assertEqual(self.press("ctrl+v"), + ["xdotool", "key", "--clearmodifiers", "ctrl+v"]) + + def test_three_keys(self): + self.assertEqual(self.press("ctrl+shift+v"), + ["xdotool", "key", "--clearmodifiers", "ctrl+shift+v"]) + + def test_the_keys_x_spells_differently(self): + """xdotool wants keysyms, not the names the code table is keyed by.""" + self.assertEqual(self.press("control+insert")[-1], "ctrl+Insert") + self.assertEqual(self.press("meta+enter")[-1], "super+Return") + + def test_no_daemon_to_ask_after(self): + with self.assertRaises(paste.PasteError) as caught: + self.press("ctrl+v", FakeCompleted(returncode=1, stderr="bad keysym")) + self.assertNotIn("ydotoold", str(caught.exception)) + if __name__ == "__main__": unittest.main()