diff --git a/DEBIAN/control b/DEBIAN/control index 485a90e..0f5c5c2 100644 --- a/DEBIAN/control +++ b/DEBIAN/control @@ -1,8 +1,9 @@ Package: input-remapper -Version: 2.2.1 +Version: 2.3.0 Architecture: all Maintainer: Sezanzeb -Depends: build-essential, libpython3-dev, libdbus-1-dev, python3, python3-evdev, python3-dasbus, python3-gi, gettext, python3-cairo, libgtk-3-0, libgtksourceview-4-dev, python3-pydantic, python3-packaging, python3-psutil +Depends: build-essential, libpython3-dev, libdbus-1-dev, python3, python3-evdev, python3-dasbus, python3-gi, gettext, python3-cairo, libgtk-3-0, libgtksourceview-4-dev, python3-pydantic, python3-packaging, python3-psutil, python3-xlib +Recommends: python3-pywayland Description: A tool to change the mapping of your input device buttons Replaces: python3-key-mapper, key-mapper, input-remapper-gtk, input-remapper-daemon, python3-inputremapper Conflicts: python3-key-mapper, key-mapper, input-remapper-gtk, input-remapper-daemon, python3-inputremapper diff --git a/DEBIAN/postinst b/DEBIAN/postinst index cfaa22c..857ba3b 100755 --- a/DEBIAN/postinst +++ b/DEBIAN/postinst @@ -14,4 +14,8 @@ if [ -d "/run/systemd/system/" ]; then pkill -f input-remapper-service # might have been started by the gui previously systemctl enable input-remapper systemctl start input-remapper + + # The focus-service is a per-user service. Enable it globally so it starts in + # each user's graphical session (takes effect on next login). + systemctl --global enable input-remapper-focus.service 2> /dev/null || true fi diff --git a/bin/input-remapper-focus-service b/bin/input-remapper-focus-service new file mode 100755 index 0000000..be3a475 --- /dev/null +++ b/bin/input-remapper-focus-service @@ -0,0 +1,28 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +# input-remapper - GUI for device specific keyboard mappings +# Copyright (C) 2025 sezanzeb +# +# This file is part of input-remapper. +# +# input-remapper is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# input-remapper is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with input-remapper. If not, see . + +"""Starts the user-level focus-service.""" + +from inputremapper.bin.input_remapper_focus_service import ( + InputRemapperFocusServiceBin, +) + +if __name__ == "__main__": + InputRemapperFocusServiceBin.main() diff --git a/data/input-remapper-focus.service b/data/input-remapper-focus.service new file mode 100644 index 0000000..71d1c35 --- /dev/null +++ b/data/input-remapper-focus.service @@ -0,0 +1,16 @@ +[Unit] +Description=Input Remapper focus-driven preset binding (per-user service) +Documentation=https://github.com/sezanzeb/input-remapper +# Runs in the graphical user session, talks to the root daemon over the system bus. +After=graphical-session.target +PartOf=graphical-session.target + +[Service] +Type=dbus +BusName=inputremapper.Focus +ExecStart=/usr/bin/input-remapper-focus-service +Restart=on-failure +RestartSec=2 + +[Install] +WantedBy=graphical-session.target diff --git a/data/input-remapper.glade b/data/input-remapper.glade index df11529..92fcef8 100644 --- a/data/input-remapper.glade +++ b/data/input-remapper.glade @@ -1209,6 +1209,132 @@ Only relevant when mapping relative inputs (e.g. mouse) to absolute outputs (e.g 2 + + + True + False + vertical + 18 + 18 + 18 + 18 + 12 + + + True + False + 12 + + + True + False + True + start + Enable application bindings + + + + + + True + True + 0 + + + + + True + True + center + Automatically apply presets based on the focused application + + + False + True + 1 + + + + + False + True + 0 + + + + + True + False + start + True + + 0.7 + + + False + True + 1 + + + + + Add application + True + True + True + start + Add a new application binding + + + False + True + 2 + + + + + True + False + + + False + True + 3 + + + + + True + True + + + True + False + none + + + True + False + start + none + + + + + + + True + True + 4 + + + + + Applications + Applications + 3 + + True diff --git a/inputremapper/bin/input_remapper_focus_service.py b/inputremapper/bin/input_remapper_focus_service.py new file mode 100644 index 0000000..501bbc3 --- /dev/null +++ b/inputremapper/bin/input_remapper_focus_service.py @@ -0,0 +1,100 @@ +# -*- coding: utf-8 -*- +# input-remapper - GUI for device specific keyboard mappings +# Copyright (C) 2025 sezanzeb +# +# This file is part of input-remapper. +# +# input-remapper is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# input-remapper is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with input-remapper. If not, see . + +"""Starts the user-level focus-service for focus-driven preset binding.""" + +import os +import sys +import time +from argparse import ArgumentParser +from typing import Optional + +from inputremapper.configs.global_config import GlobalConfig +from inputremapper.daemon import Daemon, DaemonProxy +from inputremapper.logging.logger import logger + + +class InputRemapperFocusServiceBin: + @staticmethod + def _connect_daemon(retries: int = 10, delay: float = 1.0) -> Optional[DaemonProxy]: + """Connect to the root daemon over the system bus, without pkexec. + + The focus-service runs as a background user service and must never + trigger a polkit prompt, so it connects with fallback=False and simply + retries until the daemon (started separately, e.g. via its systemd + unit) becomes available. + """ + for attempt in range(retries): + daemon = Daemon.connect(fallback=False) + if daemon is not None: + return daemon + logger.info( + "Daemon not available yet (attempt %d/%d), retrying in %.1fs", + attempt + 1, + retries, + delay, + ) + time.sleep(delay) + return None + + @staticmethod + def main() -> None: + parser = ArgumentParser() + parser.add_argument( + "-d", + "--debug", + action="store_true", + dest="debug", + help="Displays additional debug information", + default=False, + ) + parser.add_argument( + "--hide-info", + action="store_true", + dest="hide_info", + help="Don't display version information", + default=False, + ) + + options = parser.parse_args(sys.argv[1:]) + + logger.update_verbosity(options.debug) + + if not options.hide_info: + logger.log_info("input-remapper-focus-service") + + if os.getuid() == 0: + logger.warning( + "The focus-service is meant to run as the logged-in user, not root" + ) + + # Imported here so log verbosity is configured first. + from inputremapper.focus.focus_service import FocusService + from inputremapper.focus.focus_watcher import select_backend + + global_config = GlobalConfig() + backend = select_backend() + + daemon = InputRemapperFocusServiceBin._connect_daemon() + if daemon is None: + logger.error("Could not connect to the input-remapper daemon, exiting") + sys.exit(1) + + focus_service = FocusService(global_config, backend, daemon) + focus_service.run() diff --git a/inputremapper/bin/input_remapper_gtk.py b/inputremapper/bin/input_remapper_gtk.py index a602d0f..5b87346 100755 --- a/inputremapper/bin/input_remapper_gtk.py +++ b/inputremapper/bin/input_remapper_gtk.py @@ -87,6 +87,12 @@ class InputRemapperGtkBin: message_broker = MessageBroker() global_config = GlobalConfig() + global_config.load_config() + + # Start the user-level focus-service if focus-driven application binding is + # enabled, mirroring how the reader-service is launched below. + if global_config.get_app_binding_enabled(): + InputRemapperGtkBin.start_focus_service() # Create the ReaderClient before we start the reader-service, otherwise the # privileged service creates and owns those pipes, and then they cannot be accessed @@ -143,6 +149,14 @@ class InputRemapperGtkBin: logger.error(e) sys.exit(11) + @staticmethod + def start_focus_service(): + from inputremapper.gui.focus_service_client import ( + ensure_focus_service_running, + ) + + ensure_focus_service_running() + @staticmethod def stop(daemon, controller): if isinstance(daemon, Daemon): diff --git a/inputremapper/configs/app_binding.py b/inputremapper/configs/app_binding.py new file mode 100644 index 0000000..7b67702 --- /dev/null +++ b/inputremapper/configs/app_binding.py @@ -0,0 +1,108 @@ +# -*- coding: utf-8 -*- +# input-remapper - GUI for device specific keyboard mappings +# Copyright (C) 2025 sezanzeb +# +# This file is part of input-remapper. +# +# input-remapper is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# input-remapper is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with input-remapper. If not, see . + +"""Models describing which presets should be applied for which application. + +These models are persisted as part of the global config (config.json) and are +consumed by the input-remapper-focus-service to decide which presets to inject +depending on the currently focused window. +""" + +from __future__ import annotations + +import enum +import re +from typing import List, TYPE_CHECKING + +try: + from pydantic.v1 import BaseModel, Field, root_validator, validator +except ImportError: + from pydantic import ( # type: ignore[assignment, no-redef] + BaseModel, + Field, + root_validator, + validator, + ) + +from inputremapper.configs.validation_errors import ( + EmptyAppIdError, + InvalidTitleRegexError, +) + +if TYPE_CHECKING: + from inputremapper.focus.focus_watcher import FocusEvent + + +class MatchType(str, enum.Enum): + """How an AppBinding decides whether it applies to a focused window.""" + + wm_class = "wm_class" + title_regex = "title_regex" + + +def normalize_app_id(app_id: str) -> str: + """Normalize a window identifier for case-insensitive comparison.""" + return app_id.strip().lower() + + +class BoundPreset(BaseModel): + """A (device, preset) pair that should be injected for an application.""" + + group_key: str + preset: str + + @validator("group_key", "preset") + def _not_empty(cls, value: str) -> str: # noqa: N805 + if not value or not value.strip(): + raise ValueError("group_key and preset must not be empty") + return value + + +class AppBinding(BaseModel): + """Binds a focused application to a list of presets to inject.""" + + app_id: str + match: MatchType = MatchType.wm_class + presets: List[BoundPreset] = Field(default_factory=list) + + @validator("app_id") + def _app_id_not_empty(cls, value: str) -> str: # noqa: N805 + if not value or not value.strip(): + raise EmptyAppIdError() + return value + + @root_validator(skip_on_failure=True) + def _valid_regex(cls, values: dict) -> dict: # noqa: N805 + if values.get("match") == MatchType.title_regex: + app_id = values.get("app_id", "") + try: + re.compile(app_id) + except re.error as error: + raise InvalidTitleRegexError(app_id, str(error)) + return values + + def matches(self, event: "FocusEvent") -> bool: + """Whether this binding applies to the given focus event.""" + if self.match == MatchType.title_regex: + try: + return re.search(self.app_id, event.title) is not None + except re.error: + return False + + return normalize_app_id(self.app_id) == normalize_app_id(event.app_id) diff --git a/inputremapper/configs/global_config.py b/inputremapper/configs/global_config.py index 6964134..b6222b2 100644 --- a/inputremapper/configs/global_config.py +++ b/inputremapper/configs/global_config.py @@ -23,8 +23,9 @@ from __future__ import annotations import copy import json import os -from typing import Optional +from typing import List, Optional +from inputremapper.configs.app_binding import AppBinding from inputremapper.configs.paths import PathUtils from inputremapper.logging.logger import logger, VERSION from inputremapper.user import UserUtils @@ -37,6 +38,8 @@ NONE = "none" INITIAL_CONFIG = { "version": VERSION, "autoload": {}, + "app_binding_enabled": False, + "app_bindings": [], } @@ -87,6 +90,32 @@ class GlobalConfig: return self._config.get("autoload", {}).get(group_key) == preset + def get_app_binding_enabled(self) -> bool: + """Whether focus-driven preset binding is enabled.""" + return bool(self._config.get("app_binding_enabled", False)) + + def set_app_binding_enabled(self, enabled: bool) -> None: + """Enable or disable focus-driven preset binding.""" + self._config["app_binding_enabled"] = bool(enabled) + self._save_config() + + def get_app_bindings(self) -> List[AppBinding]: + """Get the configured application bindings as model objects.""" + bindings = [] + for raw in self._config.get("app_bindings", []): + try: + bindings.append(AppBinding(**raw)) + except (TypeError, ValueError) as error: + logger.error("Ignoring invalid app_binding %s: %s", raw, str(error)) + return bindings + + def set_app_bindings(self, bindings: List[AppBinding]) -> None: + """Persist the given application bindings.""" + self._config["app_bindings"] = [ + json.loads(binding.json()) for binding in bindings + ] + self._save_config() + def load_config(self, path: Optional[str] = None): """Load the config from the file system. Parameters diff --git a/inputremapper/configs/migrations.py b/inputremapper/configs/migrations.py index c533136..6285e8a 100644 --- a/inputremapper/configs/migrations.py +++ b/inputremapper/configs/migrations.py @@ -98,6 +98,9 @@ class Migrations: if v < version.parse("1.6.0-beta"): self._convert_to_individual_mappings() + if v < version.parse("2.3.0"): + self._add_app_binding_config() + # add new migrations here if v < version.parse(VERSION): @@ -212,6 +215,35 @@ class Migrations: logger.info('Updating version in config to "%s"', VERSION) json.dump(config, file, indent=4) + def _add_app_binding_config(self): + """Add the focus-driven app-binding keys to config.json if missing. + + This is an additive, backwards-compatible change: presets and autoload + configuration are left untouched. + """ + config_file = os.path.join(PathUtils.config_path(), "config.json") + if not os.path.exists(config_file): + return + + with open(config_file, "r") as file: + config = json.load(file) + + changed = False + if "app_binding_enabled" not in config: + config["app_binding_enabled"] = False + changed = True + if "app_bindings" not in config: + config["app_bindings"] = [] + changed = True + + if not changed: + return + + with open(config_file, "w") as file: + logger.info("Adding app-binding defaults to config") + json.dump(config, file, indent=4) + file.write("\n") + def _rename_to_input_remapper(self): """Rename .config/key-mapper to .config/input-remapper.""" old_config_path = os.path.join(UserUtils.home, ".config/key-mapper") diff --git a/inputremapper/configs/validation_errors.py b/inputremapper/configs/validation_errors.py index af9d213..c3f6604 100644 --- a/inputremapper/configs/validation_errors.py +++ b/inputremapper/configs/validation_errors.py @@ -116,6 +116,16 @@ class MissingOutputAxisError(ValueError): ) +class EmptyAppIdError(ValueError): + def __init__(self): + super().__init__("app_id must not be empty") + + +class InvalidTitleRegexError(ValueError): + def __init__(self, pattern: str, reason: str): + super().__init__(f'"{pattern}" is not a valid title regex: {reason}') + + class MacroError(ValueError): """Macro syntax errors.""" diff --git a/inputremapper/focus/__init__.py b/inputremapper/focus/__init__.py new file mode 100644 index 0000000..74436e8 --- /dev/null +++ b/inputremapper/focus/__init__.py @@ -0,0 +1,20 @@ +# -*- coding: utf-8 -*- +# input-remapper - GUI for device specific keyboard mappings +# Copyright (C) 2025 sezanzeb +# +# This file is part of input-remapper. +# +# input-remapper is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# input-remapper is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with input-remapper. If not, see . + +"""Focus detection and focus-driven preset binding (user-level).""" diff --git a/inputremapper/focus/backends/__init__.py b/inputremapper/focus/backends/__init__.py new file mode 100644 index 0000000..db3f37f --- /dev/null +++ b/inputremapper/focus/backends/__init__.py @@ -0,0 +1,20 @@ +# -*- coding: utf-8 -*- +# input-remapper - GUI for device specific keyboard mappings +# Copyright (C) 2025 sezanzeb +# +# This file is part of input-remapper. +# +# input-remapper is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# input-remapper is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with input-remapper. If not, see . + +"""Focus detection backends, one per compositor / display server.""" diff --git a/inputremapper/focus/backends/hyprland.py b/inputremapper/focus/backends/hyprland.py new file mode 100644 index 0000000..08a618d --- /dev/null +++ b/inputremapper/focus/backends/hyprland.py @@ -0,0 +1,174 @@ +# -*- coding: utf-8 -*- +# input-remapper - GUI for device specific keyboard mappings +# Copyright (C) 2025 sezanzeb +# +# This file is part of input-remapper. +# +# input-remapper is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# input-remapper is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with input-remapper. If not, see . + +"""Focus backend for Hyprland (socket2 event stream + socket1 queries).""" + +from __future__ import annotations + +import json +import os +import socket +from typing import Optional + +import gi + +from inputremapper.focus.focus_backend import FocusBackend, FocusEvent, OnFocus +from inputremapper.logging.logger import logger + +gi.require_version("GLib", "2.0") +from gi.repository import GLib # noqa: E402 + + +def _runtime_dirs(signature: str) -> list[str]: + """Candidate hypr runtime directories, newest layout first.""" + candidates = [] + xdg_runtime = os.environ.get("XDG_RUNTIME_DIR") + if xdg_runtime: + candidates.append(os.path.join(xdg_runtime, "hypr", signature)) + # legacy location for older Hyprland versions + candidates.append(os.path.join("/tmp", "hypr", signature)) + return candidates + + +class HyprlandFocusBackend(FocusBackend): + """Reads ``activewindow`` events from the Hyprland socket2.""" + + name = "hyprland" + + def __init__(self) -> None: + self._socket: Optional[socket.socket] = None + self._watch_id: Optional[int] = None + self._buffer = b"" + self._on_focus: Optional[OnFocus] = None + + @staticmethod + def _signature() -> Optional[str]: + return os.environ.get("HYPRLAND_INSTANCE_SIGNATURE") + + @classmethod + def _socket_path(cls, name: str) -> Optional[str]: + signature = cls._signature() + if not signature: + return None + for directory in _runtime_dirs(signature): + path = os.path.join(directory, name) + if os.path.exists(path): + return path + return None + + @staticmethod + def is_available() -> bool: + return HyprlandFocusBackend._socket_path(".socket2.sock") is not None + + def start(self, on_focus: OnFocus) -> None: + path = self._socket_path(".socket2.sock") + if not path: + raise RuntimeError("Hyprland socket2 not found") + + self._on_focus = on_focus + self._socket = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + self._socket.connect(path) + self._socket.setblocking(False) + + self._watch_id = GLib.io_add_watch( + self._socket.fileno(), + GLib.IO_IN | GLib.IO_HUP | GLib.IO_ERR, + self._on_readable, + ) + logger.debug("Hyprland focus backend listening on %s", path) + + def _on_readable(self, _fd: int, condition: int) -> bool: + if condition & (GLib.IO_HUP | GLib.IO_ERR): + logger.error("Hyprland socket2 closed") + return False + + assert self._socket is not None + try: + data = self._socket.recv(65536) + except BlockingIOError: + return True + except OSError as error: + logger.error("Hyprland socket2 read failed: %s", error) + return False + + if not data: + return False + + self._buffer += data + while b"\n" in self._buffer: + line, self._buffer = self._buffer.split(b"\n", 1) + self._handle_line(line.decode(errors="replace")) + return True + + def _handle_line(self, line: str) -> None: + # Lines look like "activewindow>>CLASS,TITLE" + name, separator, data = line.partition(">>") + if not separator or name != "activewindow": + return + + app_id, _, title = data.partition(",") + event = FocusEvent(app_id=app_id, title=title, backend=self.name) + if self._on_focus is not None: + self._on_focus(event) + + def stop(self) -> None: + if self._watch_id is not None: + GLib.source_remove(self._watch_id) + self._watch_id = None + if self._socket is not None: + self._socket.close() + self._socket = None + self._buffer = b"" + + def get_current(self) -> Optional[FocusEvent]: + path = self._socket_path(".socket.sock") + if not path: + return None + + try: + with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as sock: + sock.connect(path) + sock.sendall(b"j/activewindow") + chunks = [] + while True: + chunk = sock.recv(65536) + if not chunk: + break + chunks.append(chunk) + except OSError as error: + logger.error("Failed to query Hyprland active window: %s", error) + return None + + raw = b"".join(chunks).decode(errors="replace").strip() + if not raw: + return None + + try: + data = json.loads(raw) + except json.JSONDecodeError: + return None + + if not data: + return None + + return FocusEvent( + app_id=data.get("class", "") or "", + title=data.get("title", "") or "", + backend=self.name, + ) diff --git a/inputremapper/focus/backends/null.py b/inputremapper/focus/backends/null.py new file mode 100644 index 0000000..864c40f --- /dev/null +++ b/inputremapper/focus/backends/null.py @@ -0,0 +1,55 @@ +# -*- coding: utf-8 -*- +# input-remapper - GUI for device specific keyboard mappings +# Copyright (C) 2025 sezanzeb +# +# This file is part of input-remapper. +# +# input-remapper is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# input-remapper is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with input-remapper. If not, see . + +"""A fallback backend used when no focus detection is possible. + +This is selected for example on GNOME-Wayland, where the focused window cannot +be queried without a shell extension. The feature is effectively inactive, but +the focus-service still runs and reports a clear status. +""" + +from __future__ import annotations + +from typing import Optional + +from inputremapper.focus.focus_backend import FocusBackend, FocusEvent, OnFocus +from inputremapper.logging.logger import logger + + +class NullBackend(FocusBackend): + """Does nothing. Always available as the last-resort fallback.""" + + name = "null" + + @staticmethod + def is_available() -> bool: + return True + + def start(self, on_focus: OnFocus) -> None: + logger.warning( + "No supported focus detection is available for this session " + "(GNOME-Wayland is not supported). Focus-driven preset binding is " + "inactive." + ) + + def stop(self) -> None: + pass + + def get_current(self) -> Optional[FocusEvent]: + return None diff --git a/inputremapper/focus/backends/sway.py b/inputremapper/focus/backends/sway.py new file mode 100644 index 0000000..8e53d71 --- /dev/null +++ b/inputremapper/focus/backends/sway.py @@ -0,0 +1,215 @@ +# -*- coding: utf-8 -*- +# input-remapper - GUI for device specific keyboard mappings +# Copyright (C) 2025 sezanzeb +# +# This file is part of input-remapper. +# +# input-remapper is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# input-remapper is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with input-remapper. If not, see . + +"""Focus backend for Sway (and other i3-IPC compatible compositors).""" + +from __future__ import annotations + +import json +import os +import socket +import struct +from typing import Optional + +import gi + +from inputremapper.focus.focus_backend import FocusBackend, FocusEvent, OnFocus +from inputremapper.logging.logger import logger + +gi.require_version("GLib", "2.0") +from gi.repository import GLib # noqa: E402 + +# https://i3wm.org/docs/ipc.html +_MAGIC = b"i3-ipc" +_HEADER = struct.Struct("=6sII") # magic, length, type (native byte order) + +_MSG_SUBSCRIBE = 2 +_MSG_GET_TREE = 4 + +_EVENT_MASK = 0x80000000 + + +def _pack(message_type: int, payload: bytes = b"") -> bytes: + return _HEADER.pack(_MAGIC, len(payload), message_type) + payload + + +def _recv_exactly(sock: socket.socket, length: int) -> bytes: + chunks = [] + received = 0 + while received < length: + chunk = sock.recv(length - received) + if not chunk: + raise ConnectionError("Socket closed while reading") + chunks.append(chunk) + received += len(chunk) + return b"".join(chunks) + + +def _recv_message(sock: socket.socket) -> tuple[int, bytes]: + header = _recv_exactly(sock, _HEADER.size) + _, length, message_type = _HEADER.unpack(header) + payload = _recv_exactly(sock, length) if length else b"" + return message_type, payload + + +def _container_to_event(container: dict) -> Optional[FocusEvent]: + if not container: + return None + + app_id = container.get("app_id") + if not app_id: + # XWayland windows expose their class via window_properties instead. + window_properties = container.get("window_properties") or {} + app_id = window_properties.get("class") + + title = container.get("name") or "" + return FocusEvent(app_id=app_id or "", title=title, backend=SwayFocusBackend.name) + + +def _find_focused(node: dict) -> Optional[dict]: + if node.get("focused"): + return node + for child in node.get("nodes", []) + node.get("floating_nodes", []): + found = _find_focused(child) + if found is not None: + return found + return None + + +class SwayFocusBackend(FocusBackend): + """Subscribes to window events over the i3-IPC socket ($SWAYSOCK).""" + + name = "sway" + + def __init__(self) -> None: + self._socket: Optional[socket.socket] = None + self._watch_id: Optional[int] = None + self._buffer = b"" + self._on_focus: Optional[OnFocus] = None + + @staticmethod + def _socket_path() -> Optional[str]: + return os.environ.get("SWAYSOCK") or os.environ.get("I3SOCK") + + @staticmethod + def is_available() -> bool: + path = SwayFocusBackend._socket_path() + return path is not None and os.path.exists(path) + + def start(self, on_focus: OnFocus) -> None: + path = self._socket_path() + if not path: + raise RuntimeError("SWAYSOCK is not set") + + self._on_focus = on_focus + self._socket = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + self._socket.connect(path) + self._socket.sendall(_pack(_MSG_SUBSCRIBE, json.dumps(["window"]).encode())) + self._socket.setblocking(False) + + self._watch_id = GLib.io_add_watch( + self._socket.fileno(), + GLib.IO_IN | GLib.IO_HUP | GLib.IO_ERR, + self._on_readable, + ) + logger.debug("Sway focus backend listening on %s", path) + + def _on_readable(self, _fd: int, condition: int) -> bool: + if condition & (GLib.IO_HUP | GLib.IO_ERR): + logger.error("Sway IPC socket closed") + return False + + assert self._socket is not None + try: + data = self._socket.recv(65536) + except BlockingIOError: + return True + except OSError as error: + logger.error("Sway IPC read failed: %s", error) + return False + + if not data: + return False + + self._buffer += data + self._consume_buffer() + return True + + def _consume_buffer(self) -> None: + while len(self._buffer) >= _HEADER.size: + _, length, message_type = _HEADER.unpack(self._buffer[: _HEADER.size]) + total = _HEADER.size + length + if len(self._buffer) < total: + break + + payload = self._buffer[_HEADER.size : total] + self._buffer = self._buffer[total:] + + if message_type & _EVENT_MASK: + self._handle_event(payload) + + def _handle_event(self, payload: bytes) -> None: + try: + data = json.loads(payload.decode()) + except (json.JSONDecodeError, UnicodeDecodeError): + return + + change = data.get("change") + if change not in ("focus", "title", "new"): + return + + event = _container_to_event(data.get("container") or {}) + if event is not None and self._on_focus is not None: + self._on_focus(event) + + def stop(self) -> None: + if self._watch_id is not None: + GLib.source_remove(self._watch_id) + self._watch_id = None + if self._socket is not None: + self._socket.close() + self._socket = None + self._buffer = b"" + + def get_current(self) -> Optional[FocusEvent]: + path = self._socket_path() + if not path: + return None + + try: + with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as sock: + sock.connect(path) + sock.sendall(_pack(_MSG_GET_TREE)) + while True: + message_type, payload = _recv_message(sock) + if message_type == _MSG_GET_TREE: + break + except OSError as error: + logger.error("Failed to query Sway tree: %s", error) + return None + + try: + tree = json.loads(payload.decode()) + except (json.JSONDecodeError, UnicodeDecodeError): + return None + + focused = _find_focused(tree) + if focused is None: + return None + return _container_to_event(focused) diff --git a/inputremapper/focus/backends/wlr_foreign_toplevel.py b/inputremapper/focus/backends/wlr_foreign_toplevel.py new file mode 100644 index 0000000..391f0ec --- /dev/null +++ b/inputremapper/focus/backends/wlr_foreign_toplevel.py @@ -0,0 +1,210 @@ +# -*- coding: utf-8 -*- +# input-remapper - GUI for device specific keyboard mappings +# Copyright (C) 2025 sezanzeb +# +# This file is part of input-remapper. +# +# input-remapper is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# input-remapper is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with input-remapper. If not, see . + +"""Focus backend for wlroots based compositors and KDE/KWin (Wayland). + +Uses the ``zwlr_foreign_toplevel_manager_v1`` protocol to learn which toplevel +window is currently "activated" (focused). pywayland is an optional dependency: +if it (or the protocol bindings) are missing, the backend reports itself as +unavailable and the registry falls back to the next backend. +""" + +from __future__ import annotations + +import os +from typing import Any, Dict, Optional + +import gi + +from inputremapper.focus.focus_backend import FocusBackend, FocusEvent, OnFocus +from inputremapper.logging.logger import logger + +gi.require_version("GLib", "2.0") +from gi.repository import GLib # noqa: E402 + +# zwlr_foreign_toplevel_handle_v1 state value for "activated". +_STATE_ACTIVATED = 2 + +_MANAGER_INTERFACE = "zwlr_foreign_toplevel_manager_v1" + + +def _import_protocol() -> Optional[Any]: + """Import the zwlr_foreign_toplevel manager class, or None if unavailable.""" + try: + from pywayland.protocol.wlr_foreign_toplevel_management_unstable_v1 import ( # noqa: E501 + ZwlrForeignToplevelManagerV1, + ) + except (ImportError, ModuleNotFoundError): + return None + return ZwlrForeignToplevelManagerV1 + + +class WlrForeignToplevelBackend(FocusBackend): + """Tracks the activated toplevel via zwlr_foreign_toplevel_manager_v1.""" + + name = "wlr-foreign-toplevel" + + def __init__(self) -> None: + self._display: Any = None + self._registry: Any = None + self._manager: Any = None + self._watch_id: Optional[int] = None + self._on_focus: Optional[OnFocus] = None + # handle -> {"app_id": str, "title": str, "activated": bool} + self._toplevels: Dict[Any, Dict[str, Any]] = {} + self._current: Optional[FocusEvent] = None + + @staticmethod + def is_available() -> bool: + if not os.environ.get("WAYLAND_DISPLAY"): + return False + + if _import_protocol() is None: + logger.debug("pywayland or wlr-foreign-toplevel bindings not available") + return False + + # Verify the compositor actually exports the manager global. + try: + from pywayland.client import Display + except (ImportError, ModuleNotFoundError): + return False + + display = None + try: + display = Display() + display.connect() + found = {"value": False} + + registry = display.get_registry() + + def _on_global(_registry, _name, interface, _version): + if interface == _MANAGER_INTERFACE: + found["value"] = True + + registry.dispatcher["global"] = _on_global + display.roundtrip() + return found["value"] + except Exception as error: # pragma: no cover - depends on environment + logger.debug("wlr-foreign-toplevel probe failed: %s", error) + return False + finally: + if display is not None: + try: + display.disconnect() + except Exception: # pragma: no cover - defensive + pass + + def start(self, on_focus: OnFocus) -> None: + from pywayland.client import Display + + manager_class = _import_protocol() + if manager_class is None: + raise RuntimeError("wlr-foreign-toplevel bindings are not available") + + self._on_focus = on_focus + self._display = Display() + self._display.connect() + self._registry = self._display.get_registry() + self._registry.dispatcher["global"] = self._make_global_handler(manager_class) + self._display.roundtrip() + self._display.flush() + + self._watch_id = GLib.io_add_watch( + self._display.get_fd(), + GLib.IO_IN | GLib.IO_HUP | GLib.IO_ERR, + self._on_readable, + ) + logger.debug("wlr-foreign-toplevel focus backend started") + + def _make_global_handler(self, manager_class: Any): + def _on_global(registry, name, interface, version): + if interface != _MANAGER_INTERFACE: + return + self._manager = registry.bind(name, manager_class, version) + self._manager.dispatcher["toplevel"] = self._on_toplevel + + return _on_global + + def _on_toplevel(self, _manager, handle) -> None: + self._toplevels[handle] = {"app_id": "", "title": "", "activated": False} + handle.dispatcher["app_id"] = self._on_app_id + handle.dispatcher["title"] = self._on_title + handle.dispatcher["state"] = self._on_state + handle.dispatcher["done"] = self._on_done + handle.dispatcher["closed"] = self._on_closed + + def _on_app_id(self, handle, app_id) -> None: + if handle in self._toplevels: + self._toplevels[handle]["app_id"] = app_id or "" + + def _on_title(self, handle, title) -> None: + if handle in self._toplevels: + self._toplevels[handle]["title"] = title or "" + + def _on_state(self, handle, states) -> None: + if handle not in self._toplevels: + return + # `states` is a wl_array of 32-bit ints. + values = list(states) if states is not None else [] + self._toplevels[handle]["activated"] = _STATE_ACTIVATED in values + + def _on_done(self, handle) -> None: + info = self._toplevels.get(handle) + if info is None or not info["activated"]: + return + event = FocusEvent( + app_id=info["app_id"], title=info["title"], backend=self.name + ) + self._current = event + if self._on_focus is not None: + self._on_focus(event) + + def _on_closed(self, handle) -> None: + self._toplevels.pop(handle, None) + + def _on_readable(self, _fd: int, condition: int) -> bool: + if condition & (GLib.IO_HUP | GLib.IO_ERR): + logger.error("Wayland connection lost") + return False + + try: + self._display.dispatch(block=False) + self._display.flush() + except Exception as error: # pragma: no cover - depends on environment + logger.error("Error dispatching Wayland events: %s", error) + return False + return True + + def get_current(self) -> Optional[FocusEvent]: + return self._current + + def stop(self) -> None: + if self._watch_id is not None: + GLib.source_remove(self._watch_id) + self._watch_id = None + if self._display is not None: + try: + self._display.disconnect() + except Exception: # pragma: no cover - defensive + pass + self._display = None + self._manager = None + self._registry = None + self._toplevels.clear() + self._current = None diff --git a/inputremapper/focus/backends/xorg.py b/inputremapper/focus/backends/xorg.py new file mode 100644 index 0000000..f091888 --- /dev/null +++ b/inputremapper/focus/backends/xorg.py @@ -0,0 +1,181 @@ +# -*- coding: utf-8 -*- +# input-remapper - GUI for device specific keyboard mappings +# Copyright (C) 2025 sezanzeb +# +# This file is part of input-remapper. +# +# input-remapper is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# input-remapper is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with input-remapper. If not, see . + +"""Focus backend for X11 / XWayland using python-xlib. + +The root window is watched for ``_NET_ACTIVE_WINDOW`` property changes. The +X connection's file descriptor is integrated into the GLib main loop so no +extra thread is needed. +""" + +from __future__ import annotations + +import os +from typing import Any, Optional + +import gi + +from inputremapper.focus.focus_backend import FocusBackend, FocusEvent, OnFocus +from inputremapper.logging.logger import logger + +gi.require_version("GLib", "2.0") +from gi.repository import GLib # noqa: E402 + + +class XorgFocusBackend(FocusBackend): + """Detects focus changes via _NET_ACTIVE_WINDOW on the X root window.""" + + name = "xorg" + + def __init__(self) -> None: + self._display: Any = None + self._root: Any = None + self._watch_id: Optional[int] = None + self._on_focus: Optional[OnFocus] = None + self._net_active_window: Any = None + self._net_wm_name: Any = None + self._utf8_string: Any = None + + @staticmethod + def is_available() -> bool: + if not os.environ.get("DISPLAY"): + return False + try: + from Xlib import display as xdisplay + except ImportError: + logger.debug("python-xlib is not installed") + return False + + try: + connection = xdisplay.Display() + connection.close() + except Exception as error: # pragma: no cover - depends on environment + logger.debug("Could not open X display: %s", error) + return False + return True + + def start(self, on_focus: OnFocus) -> None: + from Xlib import X, display as xdisplay + + self._on_focus = on_focus + self._display = xdisplay.Display() + self._root = self._display.screen().root + + self._net_active_window = self._display.intern_atom("_NET_ACTIVE_WINDOW") + self._net_wm_name = self._display.intern_atom("_NET_WM_NAME") + self._utf8_string = self._display.intern_atom("UTF8_STRING") + + self._root.change_attributes(event_mask=X.PropertyChangeMask) + self._display.flush() + + self._watch_id = GLib.io_add_watch( + self._display.fileno(), + GLib.IO_IN | GLib.IO_HUP | GLib.IO_ERR, + self._on_readable, + ) + logger.debug("Xorg focus backend watching root window") + + def _on_readable(self, _fd: int, condition: int) -> bool: + from Xlib import X + + if condition & (GLib.IO_HUP | GLib.IO_ERR): + logger.error("X connection lost") + return False + + try: + pending = self._display.pending_events() + for _ in range(pending): + event = self._display.next_event() + if ( + event.type == X.PropertyNotify + and event.atom == self._net_active_window + ): + self._emit_current() + except Exception as error: # pragma: no cover - depends on environment + logger.error("Error while reading X events: %s", error) + return False + + return True + + def _emit_current(self) -> None: + event = self.get_current() + if event is not None and self._on_focus is not None: + self._on_focus(event) + + def _active_window(self) -> Any: + prop = self._root.get_full_property(self._net_active_window, 0) + if prop is None or not prop.value: + return None + window_id = prop.value[0] + if not window_id: + return None + return self._display.create_resource_object("window", window_id) + + def _read_title(self, window: Any) -> str: + from Xlib import X + + try: + prop = window.get_full_property(self._net_wm_name, self._utf8_string) + if prop and prop.value: + value = prop.value + if isinstance(value, bytes): + return value.decode(errors="replace") + return str(value) + prop = window.get_full_property(X.XA_WM_NAME, X.AnyPropertyType) + if prop and prop.value: + value = prop.value + if isinstance(value, bytes): + return value.decode(errors="replace") + return str(value) + except Exception: # pragma: no cover - depends on environment + pass + return "" + + def get_current(self) -> Optional[FocusEvent]: + if self._display is None: + return None + + from Xlib.error import XError + + try: + window = self._active_window() + if window is None: + return FocusEvent(app_id="", title="", backend=self.name) + + wm_class = window.get_wm_class() + # get_wm_class returns (instance, class); the class is the canonical id + app_id = wm_class[1] if wm_class else "" + title = self._read_title(window) + except XError as error: + logger.debug("X error while reading active window: %s", error) + return None + + return FocusEvent(app_id=app_id or "", title=title, backend=self.name) + + def stop(self) -> None: + if self._watch_id is not None: + GLib.source_remove(self._watch_id) + self._watch_id = None + if self._display is not None: + try: + self._display.close() + except Exception: # pragma: no cover - defensive + pass + self._display = None + self._root = None diff --git a/inputremapper/focus/focus_backend.py b/inputremapper/focus/focus_backend.py new file mode 100644 index 0000000..576d5c7 --- /dev/null +++ b/inputremapper/focus/focus_backend.py @@ -0,0 +1,88 @@ +# -*- coding: utf-8 -*- +# input-remapper - GUI for device specific keyboard mappings +# Copyright (C) 2025 sezanzeb +# +# This file is part of input-remapper. +# +# input-remapper is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# input-remapper is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with input-remapper. If not, see . + +"""The base class and event for all focus backends. + +This lives in its own module to avoid a circular import between the backend +registry (``focus_watcher``) and the concrete backends. +""" + +from __future__ import annotations + +from abc import ABC, abstractmethod +from dataclasses import dataclass +from typing import Callable, Optional + + +@dataclass(frozen=True) +class FocusEvent: + """A normalized description of the currently focused window. + + Attributes + ---------- + app_id + The canonical application identifier (WM_CLASS on X11, app_id on + Wayland). This is the value bindings are matched against. May be empty + for transient focus changes (menus, popups, no focus). + title + The human readable window title (_NET_WM_NAME / title). + backend + The name of the backend that produced this event. + """ + + app_id: str + title: str + backend: str + + +# Type alias for the focus callback passed to FocusBackend.start. +OnFocus = Callable[[FocusEvent], None] + + +class FocusBackend(ABC): + """Detects the focused window for a specific compositor / display server. + + Backends integrate into the GLib main loop (e.g. via ``GLib.io_add_watch`` + on a file descriptor) and call the ``on_focus`` callback whenever the + focused window changes. + """ + + # A short, human readable identifier, also used as FocusEvent.backend. + name: str = "base" + + @staticmethod + @abstractmethod + def is_available() -> bool: + """Whether this backend can run in the current session.""" + raise NotImplementedError + + @abstractmethod + def start(self, on_focus: OnFocus) -> None: + """Begin watching for focus changes, invoking on_focus on each change.""" + raise NotImplementedError + + @abstractmethod + def stop(self) -> None: + """Stop watching and release all resources.""" + raise NotImplementedError + + @abstractmethod + def get_current(self) -> Optional[FocusEvent]: + """Return the currently focused window, or None if unavailable.""" + raise NotImplementedError diff --git a/inputremapper/focus/focus_service.py b/inputremapper/focus/focus_service.py new file mode 100644 index 0000000..1f205ad --- /dev/null +++ b/inputremapper/focus/focus_service.py @@ -0,0 +1,284 @@ +# -*- coding: utf-8 -*- +# input-remapper - GUI for device specific keyboard mappings +# Copyright (C) 2025 sezanzeb +# +# This file is part of input-remapper. +# +# input-remapper is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# input-remapper is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with input-remapper. If not, see . + +"""User-level service that applies presets based on the focused application. + +This process runs as the logged-in user (never root, no pkexec). It reads the +user config, detects the focused window through a ``FocusBackend`` and drives +the existing root daemon over the system D-Bus. It exposes a small interface on +the session bus (``inputremapper.Focus``) for the GUI. +""" + +from __future__ import annotations + +import json +from typing import Dict, List, Optional + +import gi +from dasbus.connection import SessionMessageBus +from dasbus.identifier import DBusServiceIdentifier +from dasbus.loop import EventLoop +from dasbus.signal import Signal + +from inputremapper.configs.app_binding import AppBinding +from inputremapper.configs.global_config import GlobalConfig +from inputremapper.daemon import DaemonProxy +from inputremapper.focus.focus_backend import FocusBackend, FocusEvent +from inputremapper.logging.logger import logger + +gi.require_version("GLib", "2.0") +from gi.repository import GLib # noqa: E402 + +SESSION_BUS = SessionMessageBus() + +FOCUS = DBusServiceIdentifier( + namespace=("inputremapper", "Focus"), + message_bus=SESSION_BUS, +) + +# How long to wait for the focus to settle before reconciling, in milliseconds. +# This collapses rapid alt-tabbing into a single reconciliation. +DEFAULT_DEBOUNCE_MS = 150 + + +class FocusService: + """Reconciles injected presets with the currently focused application. + + The service only ever stops injections that it started itself (tracked in + ``_app_controlled``). Presets applied manually (e.g. from the GUI) are left + untouched until another *bound* application takes focus and reclaims that + device, which implements the "manual apply is a temporary override" + behavior. + """ + + __dbus_xml__ = f""" + + + + + + + + + + + + + + + + + + + """ + + def __init__( + self, + global_config: GlobalConfig, + backend: FocusBackend, + daemon: DaemonProxy, + debounce_ms: int = DEFAULT_DEBOUNCE_MS, + ) -> None: + self.global_config = global_config + self.backend = backend + self.daemon = daemon + self._debounce_ms = debounce_ms + + # The signal attribute that dasbus connects to the bus. + self.focus_changed = Signal() + + self._bindings: List[AppBinding] = [] + self._enabled = False + # group_key -> preset currently injected because of a binding + self._app_controlled: Dict[str, str] = {} + + self._current_event: Optional[FocusEvent] = None + self._pending_event: Optional[FocusEvent] = None + self._debounce_id: Optional[int] = None + + self._load_config() + + # -- lifecycle --------------------------------------------------------- + + def _load_config(self) -> None: + self.global_config.load_config() + self._enabled = self.global_config.get_app_binding_enabled() + self._bindings = self.global_config.get_app_bindings() + logger.info( + "Loaded %d app binding(s), enabled=%s", + len(self._bindings), + self._enabled, + ) + + def publish(self) -> None: + """Make the session-bus interface available to the GUI.""" + try: + SESSION_BUS.publish_object(FOCUS.object_path, self) + SESSION_BUS.register_service(FOCUS.service_name) + except ConnectionError as error: + logger.error("Is the focus-service already running? (%s)", str(error)) + raise RuntimeError("Failed to publish focus-service on D-Bus") from error + + def start(self) -> None: + """Publish the interface and start watching for focus changes.""" + self.publish() + self.backend.start(self._on_focus) + self._current_event = self.backend.get_current() + if self._current_event is not None and self._enabled: + self._reconcile(self._current_event) + logger.info("Focus-service started with backend '%s'", self.backend.name) + + def run(self) -> None: + """Start the service and block on the GLib main loop.""" + self.start() + loop = EventLoop() + loop.run() + + def stop(self) -> None: + """Stop the backend and any injections this service controls.""" + if self._debounce_id is not None: + GLib.source_remove(self._debounce_id) + self._debounce_id = None + self.backend.stop() + for group_key in list(self._app_controlled): + self.daemon.stop_injecting(group_key) + self._app_controlled.clear() + + # -- focus handling ---------------------------------------------------- + + def _on_focus(self, event: FocusEvent) -> None: + """Called by the backend on every focus change (debounced here).""" + if not event.app_id: + # Transient focus (menus, popups, no focused window). Ignore it so + # we don't tear down injections during alt-tab. + logger.debug("Ignoring transient focus event: %s", event) + return + + self._pending_event = event + if self._debounce_id is not None: + GLib.source_remove(self._debounce_id) + self._debounce_id = GLib.timeout_add(self._debounce_ms, self._flush) + + def _flush(self) -> bool: + self._debounce_id = None + event = self._pending_event + if event is None: + return False + + changed = event != self._current_event + self._current_event = event + + if changed: + self._emit_focus_changed(event) + + if self._enabled: + self._reconcile(event) + + return False # GLib: do not repeat + + def _compute_desired(self, event: FocusEvent) -> Dict[str, str]: + """Map of group_key -> preset that should be injected for this event.""" + desired: Dict[str, str] = {} + for binding in self._bindings: + if binding.matches(event): + for bound in binding.presets: + desired[bound.group_key] = bound.preset + return desired + + def _reconcile(self, event: FocusEvent) -> None: + desired = self._compute_desired(event) + + # Stop app-controlled injections that are no longer wanted. Manually + # applied presets are not in _app_controlled, so they are never stopped + # here. + for group_key in list(self._app_controlled): + if group_key not in desired: + logger.info("Focus reconcile: stopping '%s'", group_key) + self.daemon.stop_injecting(group_key) + del self._app_controlled[group_key] + + # Start or replace the wanted injections. + for group_key, preset in desired.items(): + if self._app_controlled.get(group_key) == preset: + continue + logger.info( + "Focus reconcile: injecting '%s' on '%s' for app '%s'", + preset, + group_key, + event.app_id, + ) + if self.daemon.start_injecting(group_key, preset): + self._app_controlled[group_key] = preset + else: + logger.error( + "Failed to start injecting '%s' on '%s'", preset, group_key + ) + + def _emit_focus_changed(self, event: FocusEvent) -> None: + self.focus_changed(self._event_to_json(event)) + + @staticmethod + def _event_to_json(event: Optional[FocusEvent]) -> str: + if event is None: + return json.dumps({"app_id": "", "title": ""}) + return json.dumps({"app_id": event.app_id, "title": event.title}) + + # -- D-Bus interface --------------------------------------------------- + + def get_focused_app(self) -> str: + """Return json {app_id, title} for the currently focused window.""" + return self._event_to_json(self._current_event) + + def reload(self) -> None: + """Re-read the config from disk and reconcile the current focus.""" + logger.info("Reloading focus-service config") + self._load_config() + if self._current_event is not None and self._enabled: + self._reconcile(self._current_event) + elif not self._enabled: + self._stop_app_controlled() + + def set_enabled(self, enabled: bool) -> None: + """Toggle focus-driven binding at runtime (does not write the config).""" + logger.info("Setting focus-binding enabled=%s", enabled) + self._enabled = bool(enabled) + if self._enabled: + if self._current_event is not None: + self._reconcile(self._current_event) + else: + self._stop_app_controlled() + + def get_status(self) -> str: + """Return json describing the service state for the GUI.""" + return json.dumps( + { + "backend": self.backend.name, + "enabled": self._enabled, + "focused": { + "app_id": self._current_event.app_id if self._current_event else "", + "title": self._current_event.title if self._current_event else "", + }, + "app_controlled": dict(self._app_controlled), + } + ) + + def _stop_app_controlled(self) -> None: + for group_key in list(self._app_controlled): + self.daemon.stop_injecting(group_key) + del self._app_controlled[group_key] diff --git a/inputremapper/focus/focus_watcher.py b/inputremapper/focus/focus_watcher.py new file mode 100644 index 0000000..64f006d --- /dev/null +++ b/inputremapper/focus/focus_watcher.py @@ -0,0 +1,83 @@ +# -*- coding: utf-8 -*- +# input-remapper - GUI for device specific keyboard mappings +# Copyright (C) 2025 sezanzeb +# +# This file is part of input-remapper. +# +# input-remapper is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# input-remapper is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with input-remapper. If not, see . + +"""Abstractions to detect the currently focused window across compositors. + +Focus detection is implemented as a registry of backends. To add support for a +new compositor, implement a new ``FocusBackend`` and append its class to +``focus_backend_classes`` (ordered by priority). ``select_backend`` returns the +first backend that reports itself as available, so the most specific backend +for the current session wins. +""" + +from __future__ import annotations + +from typing import List, Type + +from inputremapper.focus.backends.hyprland import HyprlandFocusBackend +from inputremapper.focus.backends.null import NullBackend +from inputremapper.focus.backends.sway import SwayFocusBackend +from inputremapper.focus.backends.wlr_foreign_toplevel import ( + WlrForeignToplevelBackend, +) +from inputremapper.focus.backends.xorg import XorgFocusBackend +from inputremapper.focus.focus_backend import FocusBackend, FocusEvent +from inputremapper.logging.logger import logger + +# Re-exported for convenience / backwards compatible imports. +__all__ = [ + "FocusEvent", + "FocusBackend", + "focus_backend_classes", + "select_backend", +] + + +# Ordered by priority. The first one that reports is_available() wins. +# SWAYSOCK -> HYPRLAND_INSTANCE_SIGNATURE -> wlr-foreign-toplevel -> DISPLAY -> Null +focus_backend_classes: List[Type[FocusBackend]] = [ + SwayFocusBackend, + HyprlandFocusBackend, + WlrForeignToplevelBackend, + XorgFocusBackend, + NullBackend, +] + + +def select_backend() -> FocusBackend: + """Instantiate the first available focus backend by priority.""" + for backend_class in focus_backend_classes: + try: + available = backend_class.is_available() + except Exception as error: # pragma: no cover - defensive + logger.debug( + "Backend %s failed its availability check: %s", + backend_class.__name__, + error, + ) + available = False + + if available: + logger.info("Using focus backend %s", backend_class.name) + return backend_class() + + # NullBackend.is_available() always returns True, so this is unreachable + # as long as it stays in the registry. + logger.warning("No focus backend available, falling back to NullBackend") + return NullBackend() diff --git a/inputremapper/gui/components/app_bindings.py b/inputremapper/gui/components/app_bindings.py new file mode 100644 index 0000000..1710279 --- /dev/null +++ b/inputremapper/gui/components/app_bindings.py @@ -0,0 +1,539 @@ +# -*- coding: utf-8 -*- +# input-remapper - GUI for device specific keyboard mappings +# Copyright (C) 2025 sezanzeb +# +# This file is part of input-remapper. +# +# input-remapper is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# input-remapper is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with input-remapper. If not, see . + + +"""Components for the focus-driven "Application Bindings" page. + +This page lets the user bind presets to applications. When the feature is +enabled, the user-level focus-service watches the focused window and applies the +bound presets automatically. The whole page is built dynamically because the +list of bindings (and the presets within each binding) is fully variable. + +Everything is driven through the MessageBroker: the editor rebuilds itself from +``AppBindingsData`` and persists changes through the Controller, which republishes +the new state. Self-originated saves are guarded so the editor does not rebuild +(and lose focus) while the user is typing. +""" + +from __future__ import annotations + +import contextlib +from typing import Callable, List, Optional, Tuple + +from gi.repository import Gtk + +from inputremapper.configs.app_binding import AppBinding, BoundPreset, MatchType +from inputremapper.gui.controller import Controller +from inputremapper.gui.gettext import _ +from inputremapper.gui.messages.message_broker import ( + MessageBroker, + MessageType, +) +from inputremapper.gui.messages.message_data import ( + AppBindingsData, + FocusAppData, + GroupsData, +) +from inputremapper.gui.utils import CTX_MAPPING, HandlerDisabled, debounce +from inputremapper.logging.logger import logger + +# human readable labels for the match types +MATCH_TYPE_LABELS = { + MatchType.wm_class: _("Window class"), + MatchType.title_regex: _("Title (regex)"), +} + + +def _maybe_blocked(widget: Gtk.Widget, handler: Callable, block: bool): + """Block ``handler`` while modifying ``widget`` only when ``block`` is True. + + Used so the initial population of a freshly built combo (before its handlers + are connected) does not log spurious HandlerDisabled warnings. + """ + if block: + return HandlerDisabled(widget, handler) + return contextlib.nullcontext() + + +class AppBindingsEditor: + """The whole "Application Bindings" page. + + Owns a global enable switch, a status label (e.g. for unsupported + environments) and a dynamic list of binding rows. + """ + + def __init__( + self, + message_broker: MessageBroker, + controller: Controller, + enabled_switch: Gtk.Switch, + status_label: Gtk.Label, + listbox: Gtk.ListBox, + add_button: Gtk.Button, + ): + self._message_broker = message_broker + self._controller = controller + self._enabled_switch = enabled_switch + self._status_label = status_label + self._listbox = listbox + self._add_button = add_button + + # available device groups (group keys), updated from the "groups" message + self._groups: Tuple[str, ...] = () + # whether focus detection is usable (service reachable + backend present) + self._detection_available = False + # the binding row currently waiting for a detected app, if any + self._detecting_row: Optional[_BindingRow] = None + # guards rebuilds caused by our own saves + self._editing = False + + self._rows: List[_BindingRow] = [] + + self._enabled_switch.connect("state-set", self._on_enabled_toggled) + self._add_button.connect("clicked", self._on_add_binding_clicked) + + self._message_broker.subscribe(MessageType.app_bindings, self._on_app_bindings) + self._message_broker.subscribe(MessageType.groups, self._on_groups) + self._message_broker.subscribe(MessageType.focused_app, self._on_focused_app) + + # -- message listeners ------------------------------------------------- + + def _on_groups(self, data: GroupsData): + self._groups = tuple(data.groups.keys()) + # refresh the device selectors of every existing row + for row in self._rows: + row.refresh_groups(self._groups) + + def _on_app_bindings(self, data: AppBindingsData): + with HandlerDisabled(self._enabled_switch, self._on_enabled_toggled): + self._enabled_switch.set_active(data.enabled) + + self._detection_available = data.supported + self._update_status_label(data) + self._update_detect_sensitivity() + + if self._editing: + # this is the echo of our own save; do not rebuild and steal focus + return + + self._rebuild(data.bindings) + + def _on_focused_app(self, data: FocusAppData): + if self._detecting_row is None: + return + + if not data.app_id: + # transient focus, keep waiting + return + + row = self._detecting_row + self._stop_detection() + row.set_app_id(data.app_id) + title = data.title or data.app_id + self._controller.show_status( + CTX_MAPPING, _('Detected application "%s"') % title + ) + self.save() + + # -- gtk handlers ------------------------------------------------------ + + def _on_enabled_toggled(self, _switch, state: bool): + self._controller.set_app_binding_enabled(state) + return False # let GTK update the switch visual state + + def _on_add_binding_clicked(self, *_args): + row = self._build_row(AppBinding(app_id=_("new application"))) + self._rows.append(row) + self._listbox.insert(row.widget, -1) + self._listbox.show_all() + self.save() + + # -- detection --------------------------------------------------------- + + def request_detection(self, row: "_BindingRow"): + """Begin (or restart) focus detection for the given binding row.""" + if self._detecting_row is row: + # toggling off + self._stop_detection() + return + + # only one row can detect at a time + if self._detecting_row is not None: + self._detecting_row.set_detecting(False) + + self._detecting_row = row + row.set_detecting(True) + self._controller.start_app_detection() + self._controller.show_status( + CTX_MAPPING, _("Switch to the application you want to bind…") + ) + + def _stop_detection(self): + if self._detecting_row is not None: + self._detecting_row.set_detecting(False) + self._detecting_row = None + self._controller.stop_app_detection() + + # -- persistence ------------------------------------------------------- + + def save(self): + """Collect every binding from the UI and persist it if all are valid.""" + try: + bindings = self.to_model() + except ValueError as error: + self._controller.show_status(CTX_MAPPING, str(error)) + return + + self._editing = True + try: + self._controller.update_app_bindings(bindings) + finally: + self._editing = False + + def to_model(self) -> List[AppBinding]: + """Return all bindings, or raise if any visible row is invalid.""" + bindings: List[AppBinding] = [] + first_error: Optional[ValueError] = None + for row in self._rows: + try: + model = row.to_model() + bindings.append(model) + row.set_error("") + except ValueError as error: + row.set_error(str(error)) + if first_error is None: + first_error = error + + if first_error is not None: + raise first_error + + return bindings + + # -- helpers ----------------------------------------------------------- + + def get_presets_for_group(self, group_key: str) -> Tuple[str, ...]: + """Proxy used by binding rows to populate their preset selectors.""" + return self._controller.get_presets_for_group(group_key) + + @property + def groups(self) -> Tuple[str, ...]: + return self._groups + + @property + def detection_available(self) -> bool: + return self._detection_available + + def _rebuild(self, bindings: Tuple[AppBinding, ...]): + self._stop_detection() + for child in self._listbox.get_children(): + self._listbox.remove(child) + self._rows = [] + + for binding in bindings: + row = self._build_row(binding) + self._rows.append(row) + self._listbox.insert(row.widget, -1) + + self._listbox.show_all() + + def _build_row(self, binding: AppBinding) -> "_BindingRow": + return _BindingRow(self, binding) + + def remove_row(self, row: "_BindingRow"): + """Remove a binding row from the list and persist.""" + if row is self._detecting_row: + self._stop_detection() + if row in self._rows: + self._rows.remove(row) + self._listbox.remove(row.widget) + self.save() + + def _update_detect_sensitivity(self): + for row in self._rows: + row.set_detect_sensitive(self._detection_available) + + def _update_status_label(self, data: AppBindingsData): + if not data.enabled: + self._status_label.set_text( + _("Enable application bindings to apply presets based on focus.") + ) + return + + if not data.reachable: + self._status_label.set_text(_("Starting the focus-detection service…")) + return + + if not data.supported: + self._status_label.set_text( + _( + "Focus detection is not supported on your environment " + "(e.g. GNOME on Wayland)." + ) + ) + return + + self._status_label.set_text( + _("Focus detection is active (backend: %s).") % data.backend + ) + + +class _PresetRow: + """A single (device, preset) selector inside a binding.""" + + def __init__(self, binding_row: "_BindingRow", bound: Optional[BoundPreset]): + self._binding_row = binding_row + self._editor = binding_row.editor + + self.widget = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=6) + self.widget.set_margin_start(12) + + self._device_combo = Gtk.ComboBoxText() + self._device_combo.set_hexpand(True) + self._preset_combo = Gtk.ComboBoxText() + self._preset_combo.set_hexpand(True) + remove_button = Gtk.Button.new_from_icon_name( + "edit-delete", Gtk.IconSize.BUTTON + ) + remove_button.set_tooltip_text(_("Remove this preset")) + + self.widget.pack_start(self._device_combo, True, True, 0) + self.widget.pack_start(self._preset_combo, True, True, 0) + self.widget.pack_start(remove_button, False, False, 0) + + self._selected_group = bound.group_key if bound else "" + self._selected_preset = bound.preset if bound else "" + + # populate before connecting, so the initial selection does not trigger a save + self._populate_devices(block=False) + self._populate_presets(block=False) + + self._device_combo.connect("changed", self._on_device_changed) + self._preset_combo.connect("changed", self._on_preset_changed) + remove_button.connect("clicked", self._on_remove_clicked) + + def _populate_devices(self, block: bool = True): + with _maybe_blocked(self._device_combo, self._on_device_changed, block): + self._device_combo.remove_all() + group_keys = list(self._editor.groups) + # keep a stored group even if the device is not currently plugged in + if self._selected_group and self._selected_group not in group_keys: + group_keys.append(self._selected_group) + for group_key in group_keys: + self._device_combo.append(group_key, group_key) + if self._selected_group: + self._device_combo.set_active_id(self._selected_group) + + def _populate_presets(self, block: bool = True): + with _maybe_blocked(self._preset_combo, self._on_preset_changed, block): + self._preset_combo.remove_all() + presets: Tuple[str, ...] = () + if self._selected_group: + presets = self._editor.get_presets_for_group(self._selected_group) + preset_names = list(presets) + if self._selected_preset and self._selected_preset not in preset_names: + preset_names.append(self._selected_preset) + for preset_name in preset_names: + self._preset_combo.append(preset_name, preset_name) + if self._selected_preset: + self._preset_combo.set_active_id(self._selected_preset) + + def refresh_groups(self): + self._populate_devices() + + def _on_device_changed(self, *_): + self._selected_group = self._device_combo.get_active_id() or "" + # changing the device invalidates the chosen preset + self._selected_preset = "" + self._populate_presets() + self._binding_row.editor.save() + + def _on_preset_changed(self, *_): + self._selected_preset = self._preset_combo.get_active_id() or "" + self._binding_row.editor.save() + + def _on_remove_clicked(self, *_): + self._binding_row.remove_preset(self) + + def to_model(self) -> BoundPreset: + if not self._selected_group or not self._selected_preset: + raise ValueError(_("Select both a device and a preset.")) + + return BoundPreset(group_key=self._selected_group, preset=self._selected_preset) + + +class _BindingRow: + """A single application binding (app_id + match type + bound presets).""" + + def __init__(self, editor: AppBindingsEditor, binding: AppBinding): + self.editor = editor + self._preset_rows: List[_PresetRow] = [] + + self.widget = Gtk.ListBoxRow() + self.widget.set_selectable(False) + self.widget.set_activatable(False) + + frame = Gtk.Frame() + frame.set_margin_start(12) + frame.set_margin_end(12) + frame.set_margin_top(6) + frame.set_margin_bottom(6) + self.widget.add(frame) + + outer = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=6) + outer.set_margin_start(12) + outer.set_margin_end(12) + outer.set_margin_top(12) + outer.set_margin_bottom(12) + frame.add(outer) + + # header row: app_id entry + match selector + detect + remove + header = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=6) + outer.pack_start(header, False, False, 0) + + self._app_id_entry = Gtk.Entry() + self._app_id_entry.set_hexpand(True) + self._app_id_entry.set_placeholder_text( + _("application identifier or title pattern") + ) + self._app_id_entry.set_text(binding.app_id) + header.pack_start(self._app_id_entry, True, True, 0) + + self._match_combo = Gtk.ComboBoxText() + for match_type in MatchType: + self._match_combo.append(match_type.value, MATCH_TYPE_LABELS[match_type]) + self._match_combo.set_active_id(binding.match.value) + header.pack_start(self._match_combo, False, False, 0) + + self._detect_button = Gtk.Button.new_with_label(_("Detect")) + self._detect_button.set_tooltip_text( + _("Detect the focused application automatically") + ) + header.pack_start(self._detect_button, False, False, 0) + + remove_button = Gtk.Button.new_from_icon_name( + "edit-delete", Gtk.IconSize.BUTTON + ) + remove_button.set_tooltip_text(_("Remove this binding")) + header.pack_start(remove_button, False, False, 0) + + # presets container + self._presets_box = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=6) + outer.pack_start(self._presets_box, False, False, 0) + + add_preset_button = Gtk.Button.new_with_label(_("Add preset")) + add_preset_button.set_halign(Gtk.Align.START) + outer.pack_start(add_preset_button, False, False, 0) + + self._error_label = Gtk.Label() + self._error_label.set_halign(Gtk.Align.START) + self._error_label.set_line_wrap(True) + self._error_label.set_no_show_all(True) + outer.pack_start(self._error_label, False, False, 0) + + for bound in binding.presets: + self._add_preset_row(bound) + + # signals + self._app_id_entry.connect("changed", self._on_app_id_changed) + self._match_combo.connect("changed", self._on_match_changed) + self._detect_button.connect("clicked", self._on_detect_clicked) + remove_button.connect("clicked", self._on_remove_clicked) + add_preset_button.connect("clicked", self._on_add_preset_clicked) + + self.set_detect_sensitive(editor.detection_available) + + # -- preset rows ------------------------------------------------------- + + def _add_preset_row(self, bound: Optional[BoundPreset]): + preset_row = _PresetRow(self, bound) + self._preset_rows.append(preset_row) + self._presets_box.pack_start(preset_row.widget, False, False, 0) + return preset_row + + def remove_preset(self, preset_row: _PresetRow): + if preset_row in self._preset_rows: + self._preset_rows.remove(preset_row) + self._presets_box.remove(preset_row.widget) + self.editor.save() + + def refresh_groups(self, _groups: Tuple[str, ...]): + for preset_row in self._preset_rows: + preset_row.refresh_groups() + + # -- detection --------------------------------------------------------- + + def set_detecting(self, detecting: bool): + self._detect_button.set_label(_("Cancel") if detecting else _("Detect")) + + def set_detect_sensitive(self, sensitive: bool): + self._detect_button.set_sensitive(sensitive) + + def set_app_id(self, app_id: str): + with HandlerDisabled(self._app_id_entry, self._on_app_id_changed): + self._app_id_entry.set_text(app_id) + + def set_error(self, error: str): + self._error_label.set_text(error) + if error: + self._error_label.show() + else: + self._error_label.hide() + + # -- gtk handlers ------------------------------------------------------ + + @debounce(500) + def _on_app_id_changed(self, *_): + self.editor.save() + + def _on_match_changed(self, *_): + self.editor.save() + + def _on_detect_clicked(self, *_): + self.editor.request_detection(self) + + def _on_remove_clicked(self, *_): + self.editor.remove_row(self) + + def _on_add_preset_clicked(self, *_): + self._add_preset_row(None) + self._presets_box.show_all() + self.editor.save() + + # -- model ------------------------------------------------------------- + + def to_model(self) -> AppBinding: + app_id = self._app_id_entry.get_text().strip() + if not app_id: + raise ValueError(_("Application identifier must not be empty.")) + + match_value = self._match_combo.get_active_id() or MatchType.wm_class.value + presets = [] + for preset_row in self._preset_rows: + presets.append(preset_row.to_model()) + + try: + return AppBinding( + app_id=app_id, + match=MatchType(match_value), + presets=presets, + ) + except ValueError as error: + # e.g. an invalid title regex; keep the row so the user can fix it + logger.debug("Invalid app binding: %s", error) + raise ValueError(str(error)) from error diff --git a/inputremapper/gui/controller.py b/inputremapper/gui/controller.py index 93e051d..5870b94 100644 --- a/inputremapper/gui/controller.py +++ b/inputremapper/gui/controller.py @@ -35,7 +35,7 @@ from typing import ( ) from evdev.ecodes import EV_KEY, EV_REL, EV_ABS -from gi.repository import Gtk +from gi.repository import Gtk, GLib from inputremapper.configs.input_config import InputCombination, InputConfig from inputremapper.configs.mapping import ( @@ -53,9 +53,11 @@ from inputremapper.configs.validation_errors import ( WrongMappingTypeForKeyError, OutputSymbolVariantError, ) +from inputremapper.configs.app_binding import AppBinding from inputremapper.exceptions import DataManagementError from inputremapper.gui.components.output_type_names import OutputTypeNames from inputremapper.gui.data_manager import DataManager, DEFAULT_PRESET_NAME +from inputremapper.gui.focus_service_client import ensure_focus_service_running from inputremapper.gui.gettext import _ from inputremapper.gui.messages.message_broker import ( MessageBroker, @@ -120,6 +122,7 @@ class Controller: # initial groups self.data_manager.publish_groups() self.data_manager.publish_uinputs() + self.data_manager.publish_app_bindings() def _on_groups_changed(self, _): """Load the newest group as soon as everyone got notified @@ -603,6 +606,47 @@ class Controller: self.data_manager.set_autoload(autoload) self.data_manager.refresh_service_config_path() + def load_app_bindings(self): + """(Re)publish the current application bindings and service state.""" + self.data_manager.publish_app_bindings() + + def set_app_binding_enabled(self, enabled: bool): + """Enable or disable focus-driven preset binding.""" + if enabled: + # make sure the user-level focus-service is running, mirroring how + # the GUI launches the reader-service. + ensure_focus_service_running() + + self.data_manager.set_app_binding_enabled(enabled) + + if enabled: + # the service might need a moment to come up and detect its backend; + # refresh the status once it had time to settle. + GLib.timeout_add(1500, self._refresh_app_bindings_once) + + def _refresh_app_bindings_once(self) -> bool: + self.data_manager.publish_app_bindings() + return False # GLib: do not repeat + + def update_app_bindings(self, bindings: List[AppBinding]): + """Persist the given application bindings.""" + try: + self.data_manager.set_app_bindings(bindings) + except PermissionError as e: + self.show_status(CTX_ERROR, _("Permission denied!"), str(e)) + + def get_presets_for_group(self, group_key: str) -> Tuple[str, ...]: + """List the presets available for an arbitrary device group.""" + return self.data_manager.get_presets_for_group(group_key) + + def start_app_detection(self): + """Start listening for focus changes to auto-fill an app_id.""" + self.data_manager.start_app_detection() + + def stop_app_detection(self): + """Stop listening for focus changes.""" + self.data_manager.stop_app_detection() + def save(self): """Save all data to the disc.""" try: diff --git a/inputremapper/gui/data_manager.py b/inputremapper/gui/data_manager.py index b077928..ac34e4d 100644 --- a/inputremapper/gui/data_manager.py +++ b/inputremapper/gui/data_manager.py @@ -21,10 +21,11 @@ import glob import os import re import time -from typing import Optional, List, Tuple, Set +from typing import Optional, List, Tuple, Set, Dict from gi.repository import GLib +from inputremapper.configs.app_binding import AppBinding from inputremapper.configs.global_config import GlobalConfig from inputremapper.configs.input_config import InputCombination, InputConfig from inputremapper.configs.mapping import UIMapping, MappingData @@ -43,7 +44,10 @@ from inputremapper.gui.messages.message_data import ( GroupData, PresetData, CombinationUpdate, + AppBindingsData, + FocusAppData, ) +from inputremapper.gui.focus_service_client import FocusServiceClient from inputremapper.gui.reader_client import ReaderClient from inputremapper.injection.global_uinputs import GlobalUInputs from inputremapper.injection.injector import ( @@ -74,12 +78,14 @@ class DataManager: daemon: DaemonProxy, uinputs: GlobalUInputs, keyboard_layout: KeyboardLayout, + focus_service: Optional[FocusServiceClient] = None, ): self.message_broker = message_broker self._reader_client = reader_client self._daemon = daemon self._uinputs = uinputs self._keyboard_layout = keyboard_layout + self._focus_service = focus_service or FocusServiceClient() uinputs.prepare_all() self._config = config @@ -156,6 +162,88 @@ class DataManager: self.message_broker.publish(InjectorStateMessage(self.get_state())) + def publish_app_bindings(self): + """Publish the "app_bindings" message with the current service state.""" + status = self._focus_service.get_status() + reachable = status is not None + backend = status.get("backend") if status else None + self.message_broker.publish( + AppBindingsData( + enabled=self._config.get_app_binding_enabled(), + bindings=tuple(self._config.get_app_bindings()), + backend=backend, + supported=reachable and backend is not None, + reachable=reachable, + ) + ) + + def get_app_bindings(self) -> List[AppBinding]: + """Get the configured application bindings.""" + return self._config.get_app_bindings() + + def set_app_bindings(self, bindings: List[AppBinding]): + """Persist the application bindings and reconcile the focus-service. + + Will send the "app_bindings" message on the MessageBroker. + """ + self._config.set_app_bindings(bindings) + self._focus_service.reload() + self.publish_app_bindings() + + def get_app_binding_enabled(self) -> bool: + """Whether focus-driven preset binding is enabled.""" + return self._config.get_app_binding_enabled() + + def set_app_binding_enabled(self, enabled: bool): + """Enable or disable focus-driven preset binding. + + Writes the config and toggles the running service. Will send the + "app_bindings" message on the MessageBroker. + """ + self._config.set_app_binding_enabled(enabled) + self._focus_service.set_enabled(enabled) + self.publish_app_bindings() + + def start_app_detection(self): + """Listen for focus changes to auto-fill an app_id. + + Will send "focused_app" messages as the focus changes. + """ + self._focus_service.connect_focus_changed(self._on_focus_changed) + + def stop_app_detection(self): + """Stop listening for focus changes.""" + self._focus_service.disconnect_focus_changed() + + def _on_focus_changed(self, app: Dict[str, str]): + self.message_broker.publish( + FocusAppData( + app_id=app.get("app_id", ""), + title=app.get("title", ""), + ) + ) + + def get_presets_for_group(self, group_key: str) -> Tuple[Name, ...]: + """Get all preset names for an arbitrary group, sorted by age. + + Unlike get_preset_names this does not depend on the active group, so it + can be used to populate the application-binding preset selectors. + """ + group = self._reader_client.groups.find(key=group_key) + if not group: + return tuple() + + device_folder = PathUtils.get_preset_path(group.name) + PathUtils.mkdir(device_folder) + + paths = glob.glob(os.path.join(glob.escape(device_folder), "*.json")) + presets = [ + os.path.splitext(os.path.basename(path))[0] + for path in sorted(paths, key=os.path.getmtime) + ] + presets.reverse() + return tuple(presets) + @property def active_group(self) -> Optional[_Group]: """The currently loaded group.""" diff --git a/inputremapper/gui/focus_service_client.py b/inputremapper/gui/focus_service_client.py new file mode 100644 index 0000000..2ec52c1 --- /dev/null +++ b/inputremapper/gui/focus_service_client.py @@ -0,0 +1,145 @@ +# -*- coding: utf-8 -*- +# input-remapper - GUI for device specific keyboard mappings +# Copyright (C) 2025 sezanzeb +# +# This file is part of input-remapper. +# +# input-remapper is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# input-remapper is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with input-remapper. If not, see . + +"""GUI-side client for the user-level focus-service (``inputremapper.Focus``). + +The focus-service exposes a small interface on the session bus. This client +wraps the dasbus proxy and stays defensive: when the service is not running (the +feature might be disabled, or the environment might not support focus detection) +every call degrades gracefully instead of raising, so the GUI keeps working. +""" + +from __future__ import annotations + +import json +import subprocess +from typing import Any, Callable, Dict, Optional + +from dasbus.connection import SessionMessageBus + +from inputremapper.bin.process_utils import ProcessUtils +from inputremapper.logging.logger import logger + +FOCUS_SERVICE_NAME = "inputremapper.Focus" +FOCUS_OBJECT_PATH = "/inputremapper/Focus" +FOCUS_SERVICE_EXECUTABLE = "input-remapper-focus-service" + + +def ensure_focus_service_running() -> None: + """Launch the user-level focus-service if it is not already running. + + Mirrors how the GUI starts the reader-service, but the focus-service runs as + the logged-in user (no pkexec / no root). + """ + if ProcessUtils.count_python_processes(FOCUS_SERVICE_EXECUTABLE) >= 1: + logger.info("Found an input-remapper-focus-service to already be running") + return + + try: + # start_new_session detaches the service from the GUI process group so it + # keeps running and is not torn down together with the GUI. + subprocess.Popen( # pylint: disable=consider-using-with + [FOCUS_SERVICE_EXECUTABLE], + start_new_session=True, + ) + logger.info("Started the input-remapper-focus-service") + except Exception as error: # pylint: disable=broad-except + logger.error("Failed to start the focus-service: %s", error) + + +class FocusServiceClient: + """Thin, defensive wrapper around the focus-service session-bus interface.""" + + def __init__(self, bus: Optional[SessionMessageBus] = None) -> None: + self._bus = bus or SessionMessageBus() + self._proxy: Any = None + self._focus_changed_handler: Optional[Callable[[str], None]] = None + + def _get_proxy(self) -> Any: + if self._proxy is None: + self._proxy = self._bus.get_proxy(FOCUS_SERVICE_NAME, FOCUS_OBJECT_PATH) + return self._proxy + + def _reset(self) -> None: + """Drop the cached proxy so the next call reconnects.""" + self._proxy = None + self._focus_changed_handler = None + + def get_status(self) -> Optional[Dict[str, Any]]: + """Return the service status as a dict, or None if it is unreachable.""" + try: + return json.loads(self._get_proxy().get_status()) + except Exception as error: # pylint: disable=broad-except + logger.debug("focus-service get_status failed: %s", error) + self._reset() + return None + + def get_focused_app(self) -> Dict[str, str]: + """Return the currently focused window as {"app_id", "title"}.""" + try: + return json.loads(self._get_proxy().get_focused_app()) + except Exception as error: # pylint: disable=broad-except + logger.debug("focus-service get_focused_app failed: %s", error) + self._reset() + return {"app_id": "", "title": ""} + + def reload(self) -> None: + """Ask the service to re-read the config from disk and reconcile.""" + try: + self._get_proxy().reload() + except Exception as error: # pylint: disable=broad-except + logger.debug("focus-service reload failed: %s", error) + self._reset() + + def set_enabled(self, enabled: bool) -> None: + """Toggle focus-driven binding at runtime (does not write the config).""" + try: + self._get_proxy().set_enabled(enabled) + except Exception as error: # pylint: disable=broad-except + logger.debug("focus-service set_enabled failed: %s", error) + self._reset() + + def connect_focus_changed(self, callback: Callable[[Dict[str, str]], None]) -> bool: + """Subscribe to the focus_changed signal. Returns whether it succeeded.""" + + def handler(app_json: str) -> None: + try: + callback(json.loads(app_json)) + except (ValueError, TypeError) as error: + logger.error("Invalid focus_changed payload %r: %s", app_json, error) + + try: + self._get_proxy().focus_changed.connect(handler) + self._focus_changed_handler = handler + return True + except Exception as error: # pylint: disable=broad-except + logger.debug("focus-service focus_changed.connect failed: %s", error) + self._reset() + return False + + def disconnect_focus_changed(self) -> None: + """Unsubscribe from the focus_changed signal.""" + if self._focus_changed_handler is None: + return + try: + self._get_proxy().focus_changed.disconnect(self._focus_changed_handler) + except Exception as error: # pylint: disable=broad-except + logger.debug("focus-service focus_changed.disconnect failed: %s", error) + finally: + self._focus_changed_handler = None diff --git a/inputremapper/gui/messages/message_data.py b/inputremapper/gui/messages/message_data.py index 94ca174..fc7b04c 100644 --- a/inputremapper/gui/messages/message_data.py +++ b/inputremapper/gui/messages/message_data.py @@ -21,6 +21,7 @@ import re from dataclasses import dataclass from typing import Dict, Tuple, Optional, Callable +from inputremapper.configs.app_binding import AppBinding from inputremapper.configs.input_config import InputCombination from inputremapper.configs.mapping import MappingData from inputremapper.gui.messages.message_types import ( @@ -124,3 +125,31 @@ class DoStackSwitch: message_type = MessageType.do_stack_switch page_index: int + + +@dataclass(frozen=True) +class AppBindingsData: + """Message with the current focus-driven application bindings and service state. + + ``backend`` is the name of the focus backend reported by the focus-service + (e.g. "xorg", "sway", "hyprland", "wlr-foreign-toplevel"). ``supported`` is + False when the service is reachable but reports no usable backend (e.g. on + GNOME-Wayland). ``reachable`` is False when the focus-service is not running + yet, in which case the backend state is simply unknown. + """ + + message_type = MessageType.app_bindings + enabled: bool + bindings: Tuple[AppBinding, ...] + backend: Optional[str] = None + supported: bool = False + reachable: bool = False + + +@dataclass(frozen=True) +class FocusAppData: + """Message with the latest focused application, emitted during detection.""" + + message_type = MessageType.focused_app + app_id: str + title: str = "" diff --git a/inputremapper/gui/messages/message_types.py b/inputremapper/gui/messages/message_types.py index c11f30b..dc083b8 100644 --- a/inputremapper/gui/messages/message_types.py +++ b/inputremapper/gui/messages/message_types.py @@ -55,6 +55,10 @@ class MessageType(Enum): do_stack_switch = "do_stack_switch" + # focus-driven preset binding (application bindings) + app_bindings = "app_bindings" + focused_app = "focused_app" + # for unit tests: test1 = "test1" test2 = "test2" diff --git a/inputremapper/gui/user_interface.py b/inputremapper/gui/user_interface.py index 09043de..a996c66 100644 --- a/inputremapper/gui/user_interface.py +++ b/inputremapper/gui/user_interface.py @@ -19,6 +19,7 @@ """User Interface.""" + from typing import Dict, Callable from gi.repository import Gtk, GtkSource, Gdk, GObject @@ -49,6 +50,7 @@ from inputremapper.gui.components.editor import ( RequireActiveMapping, GdkEventRecorder, ) +from inputremapper.gui.components.app_bindings import AppBindingsEditor from inputremapper.gui.components.main import Stack, StatusBar from inputremapper.gui.components.presets import PresetSelection from inputremapper.gui.controller import Controller @@ -203,6 +205,15 @@ class UserInterface: self.get("expo-scale"), ) + AppBindingsEditor( + message_broker, + controller, + self.get("app_binding_enabled_switch"), + self.get("app_binding_status_label"), + self.get("app_bindings_listbox"), + self.get("app_binding_add_button"), + ) + GdkEventRecorder(self.window, self.get("gdk-event-recorder-label")) RequireActiveMapping( diff --git a/inputremapper/installation_info.py b/inputremapper/installation_info.py index 339e394..f737f08 100644 --- a/inputremapper/installation_info.py +++ b/inputremapper/installation_info.py @@ -24,7 +24,7 @@ These defaults might be overwritten by package maintainers. """ COMMIT_HASH = "unknown" -VERSION = "2.2.1" +VERSION = "2.3.0" # depending on where this file is installed to, make sure to use the proper # prefix path for data. DATA_DIR = "/usr/share/input-remapper" diff --git a/install/check_dependencies.py b/install/check_dependencies.py index 83f8ae4..458c5aa 100644 --- a/install/check_dependencies.py +++ b/install/check_dependencies.py @@ -35,9 +35,20 @@ def check_dependencies() -> None: import dasbus import pygobject import pydantic + import Xlib # python-xlib, required by the X11/XWayland focus backend print("All required Python modules found") except ImportError as e: print(f"\033[93mMissing Python module: {e}\033[0m") + + try: + import pywayland # optional, enables the wlroots/KDE-Wayland focus backend + + del pywayland + except ImportError: + print( + "\033[93mOptional module 'pywayland' not found: the wlroots/KDE-Wayland " + "focus backend will be disabled\033[0m" + ) except Exception as e: print(f"\033[93mException while checking dependencies: {e}\033[0m") diff --git a/install/data_files.py b/install/data_files.py index d4bf331..4a377d6 100644 --- a/install/data_files.py +++ b/install/data_files.py @@ -40,6 +40,8 @@ def get_data_files() -> list[tuple[str, list[str]]]: ("usr/share/icons/hicolor/scalable/apps/", ["data/input-remapper.svg"]), ("usr/share/polkit-1/actions/", ["data/input-remapper.policy"]), ("usr/lib/systemd/system", ["data/input-remapper.service"]), + # The focus-service is a per-user service running in the graphical session. + ("usr/lib/systemd/user", ["data/input-remapper-focus.service"]), # Fun fact: At some point during development and testing on arch, I ended up # with an empty inputremapper.Control.conf file, causing dbus to fail to start, # which rendered the whole operating system unusable. @@ -51,6 +53,7 @@ def get_data_files() -> list[tuple[str, list[str]]]: ("usr/bin/", ["bin/input-remapper-service"]), ("usr/bin/", ["bin/input-remapper-control"]), ("usr/bin/", ["bin/input-remapper-reader-service"]), + ("usr/bin/", ["bin/input-remapper-focus-service"]), ] diff --git a/pyproject.toml b/pyproject.toml index 5baedad..d677cf4 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "input-remapper" -version = "2.2.1" +version = "2.3.0" description = "A tool to change the mapping of your input device buttons" authors = [ { name = "Sezanzeb", email = "b8x45ygc9@mozmail.com" }, @@ -16,6 +16,14 @@ dependencies = [ "PyGObject", "pydantic", "packaging", + "python-xlib", +] + +[project.optional-dependencies] +# Enables the wlroots / KDE-Wayland focus backend (zwlr_foreign_toplevel_manager_v1). +# The focus-service degrades gracefully to other backends when this is absent. +wayland = [ + "pywayland", ] [tool.setuptools.packages.find] diff --git a/scripts/build-deb.sh b/scripts/build-deb.sh index 6ff94f4..d104056 100755 --- a/scripts/build-deb.sh +++ b/scripts/build-deb.sh @@ -11,7 +11,7 @@ build_deb() { find ./build -empty -delete cp ./DEBIAN build/deb -r - dpkg-deb -Z gzip -b build/deb dist/input-remapper-2.2.1.deb + dpkg-deb -Z gzip -b build/deb dist/input-remapper-2.3.0.deb } build_deb diff --git a/tests/system/gui/test_app_bindings.py b/tests/system/gui/test_app_bindings.py new file mode 100644 index 0000000..ea98dcc --- /dev/null +++ b/tests/system/gui/test_app_bindings.py @@ -0,0 +1,290 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +# input-remapper - GUI for device specific keyboard mappings +# Copyright (C) 2025 sezanzeb +# +# This file is part of input-remapper. +# +# input-remapper is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# input-remapper is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with input-remapper. If not, see . + +import unittest +from unittest.mock import MagicMock + +import gi + +gi.require_version("Gtk", "3.0") +gi.require_version("GLib", "2.0") +from gi.repository import Gtk, GLib + +from inputremapper.configs.app_binding import AppBinding, BoundPreset, MatchType +from inputremapper.gui.components.app_bindings import ( + AppBindingsEditor, + _BindingRow, +) +from inputremapper.gui.controller import Controller +from inputremapper.gui.messages.message_broker import MessageBroker, MessageType +from inputremapper.gui.messages.message_data import ( + AppBindingsData, + FocusAppData, + GroupsData, +) +from inputremapper.gui.utils import gtk_iteration +from tests.lib.test_setup import test_setup + + +@test_setup +class TestAppBindingsEditor(unittest.TestCase): + def setUp(self) -> None: + self.message_broker = MessageBroker() + self.controller_mock: Controller = MagicMock() + self.controller_mock.get_presets_for_group.return_value = ("preset1", "preset2") + + self.enabled_switch = Gtk.Switch() + self.status_label = Gtk.Label() + self.listbox = Gtk.ListBox() + self.add_button = Gtk.Button() + + self.editor = AppBindingsEditor( + self.message_broker, + self.controller_mock, + self.enabled_switch, + self.status_label, + self.listbox, + self.add_button, + ) + + def tearDown(self) -> None: + super().tearDown() + self.message_broker.signal(MessageType.terminate) + for widget in ( + self.enabled_switch, + self.status_label, + self.listbox, + self.add_button, + ): + GLib.timeout_add(0, widget.destroy) + GLib.timeout_add(0, Gtk.main_quit) + Gtk.main() + + def _publish_bindings(self, bindings=(), enabled=False, **kwargs): + defaults = dict(backend="xorg", supported=True, reachable=True) + defaults.update(kwargs) + self.message_broker.publish( + AppBindingsData(enabled=enabled, bindings=tuple(bindings), **defaults) + ) + gtk_iteration() + + def _binding(self, app_id="firefox", presets=()): + return AppBinding( + app_id=app_id, + presets=[BoundPreset(group_key=g, preset=p) for g, p in presets], + ) + + # -- enable toggle ----------------------------------------------------- + + def test_toggle_enabled_calls_controller(self): + self.enabled_switch.emit("state-set", True) + gtk_iteration() + self.controller_mock.set_app_binding_enabled.assert_called_with(True) + + def test_app_bindings_message_updates_switch_without_callback(self): + # the incoming state must not echo back into the controller + self.controller_mock.set_app_binding_enabled.reset_mock() + self._publish_bindings(enabled=True) + self.assertTrue(self.enabled_switch.get_active()) + self.controller_mock.set_app_binding_enabled.assert_not_called() + + # -- add / remove binding ---------------------------------------------- + + def test_add_binding(self): + self.add_button.emit("clicked") + gtk_iteration() + self.assertEqual(len(self.listbox.get_children()), 1) + self.assertEqual(len(self.editor._rows), 1) + self.controller_mock.update_app_bindings.assert_called() + + def test_rebuild_from_message(self): + self._publish_bindings( + [ + self._binding("firefox", [("Foo Device", "preset1")]), + self._binding("kitty"), + ], + enabled=True, + ) + self.assertEqual(len(self.listbox.get_children()), 2) + self.assertEqual(len(self.editor._rows), 2) + + def test_remove_binding(self): + self._publish_bindings([self._binding("firefox")], enabled=True) + row = self.editor._rows[0] + self.controller_mock.update_app_bindings.reset_mock() + self.editor.remove_row(row) + gtk_iteration() + self.assertEqual(len(self.editor._rows), 0) + self.assertEqual(len(self.listbox.get_children()), 0) + self.controller_mock.update_app_bindings.assert_called() + + def test_editing_guard_prevents_rebuild(self): + self._publish_bindings([self._binding("firefox")], enabled=True) + self.assertEqual(len(self.editor._rows), 1) + # simulate being mid-save: the echo of our own write must not rebuild + self.editor._editing = True + self._publish_bindings([self._binding("a"), self._binding("b")], enabled=True) + self.assertEqual(len(self.editor._rows), 1) + + # -- preset rows ------------------------------------------------------- + + def test_add_and_remove_preset(self): + self._publish_bindings([self._binding("firefox")], enabled=True) + row: _BindingRow = self.editor._rows[0] + self.assertEqual(len(row._preset_rows), 0) + + row._on_add_preset_clicked() + gtk_iteration() + self.assertEqual(len(row._preset_rows), 1) + + preset_row = row._preset_rows[0] + row.remove_preset(preset_row) + gtk_iteration() + self.assertEqual(len(row._preset_rows), 0) + + def test_binding_to_model(self): + self._publish_bindings( + [self._binding("firefox", [("Foo Device", "preset1")])], enabled=True + ) + row: _BindingRow = self.editor._rows[0] + model = row.to_model() + self.assertIsNotNone(model) + self.assertEqual(model.app_id, "firefox") + self.assertEqual(model.match, MatchType.wm_class) + self.assertEqual(model.presets[0].group_key, "Foo Device") + + def test_empty_app_id_row_raises(self): + self.add_button.emit("clicked") + gtk_iteration() + row: _BindingRow = self.editor._rows[0] + row.set_app_id("") + with self.assertRaises(ValueError): + row.to_model() + + def test_save_rejects_empty_app_id_with_visible_feedback(self): + self.add_button.emit("clicked") + gtk_iteration() + row: _BindingRow = self.editor._rows[0] + row.set_app_id("") + + self.controller_mock.update_app_bindings.reset_mock() + self.editor.save() + gtk_iteration() + + self.controller_mock.update_app_bindings.assert_not_called() + self.assertTrue(row._error_label.get_visible()) + self.assertIn("identifier", row._error_label.get_text()) + + def test_save_rejects_invalid_title_regex_with_visible_feedback(self): + self._publish_bindings([self._binding("firefox")], enabled=True) + row: _BindingRow = self.editor._rows[0] + row.set_app_id("(") + row._match_combo.set_active_id(MatchType.title_regex.value) + + self.controller_mock.update_app_bindings.reset_mock() + self.editor.save() + gtk_iteration() + + self.controller_mock.update_app_bindings.assert_not_called() + self.assertTrue(row._error_label.get_visible()) + self.assertIn("valid title regex", row._error_label.get_text()) + + def test_save_rejects_partial_preset_with_visible_feedback(self): + self.message_broker.publish(GroupsData({"Foo Device": []})) + gtk_iteration() + self._publish_bindings([self._binding("firefox")], enabled=True) + row: _BindingRow = self.editor._rows[0] + row._on_add_preset_clicked() + preset_row = row._preset_rows[0] + preset_row._device_combo.set_active_id("Foo Device") + + self.controller_mock.update_app_bindings.reset_mock() + self.editor.save() + gtk_iteration() + + self.controller_mock.update_app_bindings.assert_not_called() + self.assertTrue(row._error_label.get_visible()) + self.assertIn("device and a preset", row._error_label.get_text()) + + # -- detection --------------------------------------------------------- + + def test_detect_flow(self): + self._publish_bindings([self._binding("firefox")], enabled=True) + row: _BindingRow = self.editor._rows[0] + + self.editor.request_detection(row) + self.controller_mock.start_app_detection.assert_called_once() + self.assertIs(self.editor._detecting_row, row) + + # a transient (empty) focus is ignored, detection keeps running + self.message_broker.publish(FocusAppData(app_id="", title="")) + gtk_iteration() + self.assertIs(self.editor._detecting_row, row) + + # a real focus fills the app_id and stops detection + self.controller_mock.update_app_bindings.reset_mock() + self.message_broker.publish(FocusAppData(app_id="kitty", title="Terminal")) + gtk_iteration() + self.assertIsNone(self.editor._detecting_row) + self.controller_mock.stop_app_detection.assert_called() + self.controller_mock.update_app_bindings.assert_called() + self.assertEqual(row.to_model().app_id, "kitty") + + def test_detect_toggle_off(self): + self._publish_bindings([self._binding("firefox")], enabled=True) + row: _BindingRow = self.editor._rows[0] + self.editor.request_detection(row) + # requesting again on the same row toggles detection off + self.editor.request_detection(row) + self.assertIsNone(self.editor._detecting_row) + self.controller_mock.stop_app_detection.assert_called() + + # -- status label ------------------------------------------------------ + + def test_status_label_disabled(self): + self._publish_bindings(enabled=False) + self.assertIn("Enable application bindings", self.status_label.get_text()) + + def test_status_label_starting(self): + self._publish_bindings(enabled=True, reachable=False, supported=False) + self.assertIn("Starting", self.status_label.get_text()) + + def test_status_label_unsupported(self): + self._publish_bindings( + enabled=True, reachable=True, supported=False, backend=None + ) + self.assertIn("not supported", self.status_label.get_text()) + + def test_status_label_active(self): + self._publish_bindings( + enabled=True, reachable=True, supported=True, backend="xorg" + ) + self.assertIn("xorg", self.status_label.get_text()) + + # -- groups ------------------------------------------------------------ + + def test_groups_message_updates_available_groups(self): + self.message_broker.publish(GroupsData({"Foo Device": [], "Bar": []})) + gtk_iteration() + self.assertEqual(self.editor.groups, ("Foo Device", "Bar")) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/unit/test_app_binding.py b/tests/unit/test_app_binding.py new file mode 100644 index 0000000..4571482 --- /dev/null +++ b/tests/unit/test_app_binding.py @@ -0,0 +1,154 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +# input-remapper - GUI for device specific keyboard mappings +# Copyright (C) 2025 sezanzeb +# +# This file is part of input-remapper. +# +# input-remapper is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# input-remapper is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with input-remapper. If not, see . + +import unittest + +try: + from pydantic.v1 import ValidationError +except ImportError: + from pydantic import ValidationError # type: ignore[assignment, no-redef] + +from inputremapper.configs.app_binding import ( + AppBinding, + BoundPreset, + MatchType, + normalize_app_id, +) +from inputremapper.focus.focus_backend import FocusEvent +from tests.lib.test_setup import test_setup + + +def _event(app_id="", title="", backend="fake") -> FocusEvent: + return FocusEvent(app_id=app_id, title=title, backend=backend) + + +@test_setup +class TestNormalizeAppId(unittest.TestCase): + def test_lower_and_strip(self): + self.assertEqual(normalize_app_id(" FireFox "), "firefox") + self.assertEqual(normalize_app_id("ALACRITTY"), "alacritty") + self.assertEqual(normalize_app_id(""), "") + + +@test_setup +class TestBoundPreset(unittest.TestCase): + def test_valid(self): + bound = BoundPreset(group_key="Foo Device", preset="preset1") + self.assertEqual(bound.group_key, "Foo Device") + self.assertEqual(bound.preset, "preset1") + + def test_empty_group_key_rejected(self): + with self.assertRaises(ValidationError): + BoundPreset(group_key="", preset="preset1") + with self.assertRaises(ValidationError): + BoundPreset(group_key=" ", preset="preset1") + + def test_empty_preset_rejected(self): + with self.assertRaises(ValidationError): + BoundPreset(group_key="Foo Device", preset="") + with self.assertRaises(ValidationError): + BoundPreset(group_key="Foo Device", preset=" ") + + +@test_setup +class TestAppBinding(unittest.TestCase): + def test_defaults(self): + binding = AppBinding(app_id="firefox") + self.assertEqual(binding.match, MatchType.wm_class) + self.assertEqual(binding.presets, []) + + def test_default_presets_are_not_shared(self): + binding_a = AppBinding(app_id="firefox") + binding_b = AppBinding(app_id="chromium") + binding_a.presets.append(BoundPreset(group_key="Foo Device", preset="preset1")) + self.assertEqual(binding_b.presets, []) + + def test_empty_app_id_rejected(self): + with self.assertRaises(ValidationError): + AppBinding(app_id="") + with self.assertRaises(ValidationError): + AppBinding(app_id=" ") + + def test_invalid_regex_rejected(self): + with self.assertRaises(ValidationError): + AppBinding(app_id="(unclosed", match=MatchType.title_regex) + + def test_invalid_regex_allowed_for_wm_class(self): + # an unbalanced parenthesis is a fine literal wm_class, only title_regex + # compiles it as a pattern. + binding = AppBinding(app_id="(unclosed", match=MatchType.wm_class) + self.assertEqual(binding.app_id, "(unclosed") + + def test_with_presets(self): + binding = AppBinding( + app_id="firefox", + presets=[BoundPreset(group_key="Foo Device", preset="preset1")], + ) + self.assertEqual(len(binding.presets), 1) + self.assertEqual(binding.presets[0].group_key, "Foo Device") + + # -- matches() --------------------------------------------------------- + + def test_matches_wm_class_normalized(self): + binding = AppBinding(app_id="FireFox", match=MatchType.wm_class) + self.assertTrue(binding.matches(_event(app_id="firefox"))) + self.assertTrue(binding.matches(_event(app_id=" FIREFOX "))) + self.assertFalse(binding.matches(_event(app_id="chromium"))) + + def test_matches_wm_class_ignores_title(self): + binding = AppBinding(app_id="firefox", match=MatchType.wm_class) + self.assertFalse( + binding.matches(_event(app_id="chromium", title="firefox - page")) + ) + + def test_matches_title_regex(self): + binding = AppBinding(app_id="Privat.*Browsing", match=MatchType.title_regex) + self.assertTrue( + binding.matches(_event(app_id="firefox", title="Private Browsing")) + ) + self.assertFalse( + binding.matches(_event(app_id="firefox", title="Normal Window")) + ) + + def test_matches_title_regex_searches_substring(self): + binding = AppBinding(app_id="vim", match=MatchType.title_regex) + # re.search, not fullmatch + self.assertTrue(binding.matches(_event(title="file.py - nvim"))) + + def test_matches_title_regex_case_sensitive(self): + # unlike wm_class, the title regex is not normalized + binding = AppBinding(app_id="Private", match=MatchType.title_regex) + self.assertFalse(binding.matches(_event(title="private browsing"))) + + def test_roundtrip_through_json(self): + binding = AppBinding( + app_id="firefox", + match=MatchType.wm_class, + presets=[BoundPreset(group_key="Foo Device", preset="preset1")], + ) + import json + + as_dict = json.loads(binding.json()) + restored = AppBinding(**as_dict) + self.assertEqual(restored, binding) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/unit/test_controller.py b/tests/unit/test_controller.py index c328833..185d17b 100644 --- a/tests/unit/test_controller.py +++ b/tests/unit/test_controller.py @@ -1625,3 +1625,73 @@ class TestController(unittest.TestCase): output_type=None, output_code=None, ) + + +@test_setup +class TestControllerAppBindings(unittest.TestCase): + def setUp(self) -> None: + self.message_broker = MessageBroker() + self.data_manager = MagicMock() + self.controller = Controller(self.message_broker, self.data_manager) + self.user_interface = MagicMock() + self.controller.set_gui(self.user_interface) + + def test_load_app_bindings(self): + self.controller.load_app_bindings() + self.data_manager.publish_app_bindings.assert_called_once() + + def test_set_app_binding_enabled_true_starts_service(self): + with patch( + "inputremapper.gui.controller.ensure_focus_service_running" + ) as ensure, patch( + "inputremapper.gui.controller.GLib.timeout_add" + ) as timeout_add: + self.controller.set_app_binding_enabled(True) + + ensure.assert_called_once() + self.data_manager.set_app_binding_enabled.assert_called_once_with(True) + # a delayed status refresh is scheduled + timeout_add.assert_called_once() + + def test_set_app_binding_enabled_false_does_not_start_service(self): + with patch( + "inputremapper.gui.controller.ensure_focus_service_running" + ) as ensure, patch( + "inputremapper.gui.controller.GLib.timeout_add" + ) as timeout_add: + self.controller.set_app_binding_enabled(False) + + ensure.assert_not_called() + self.data_manager.set_app_binding_enabled.assert_called_once_with(False) + timeout_add.assert_not_called() + + def test_refresh_app_bindings_once(self): + result = self.controller._refresh_app_bindings_once() + self.assertFalse(result) # GLib: do not repeat + self.data_manager.publish_app_bindings.assert_called_once() + + def test_update_app_bindings(self): + from inputremapper.configs.app_binding import AppBinding + + bindings = [AppBinding(app_id="firefox")] + self.controller.update_app_bindings(bindings) + self.data_manager.set_app_bindings.assert_called_once_with(bindings) + + def test_update_app_bindings_permission_error(self): + self.data_manager.set_app_bindings.side_effect = PermissionError("nope") + with patch.object(self.controller, "show_status") as show_status: + self.controller.update_app_bindings([]) + show_status.assert_called_once() + self.assertEqual(show_status.call_args[0][0], CTX_ERROR) + + def test_get_presets_for_group(self): + self.data_manager.get_presets_for_group.return_value = ("a", "b") + result = self.controller.get_presets_for_group("Foo Device") + self.assertEqual(result, ("a", "b")) + self.data_manager.get_presets_for_group.assert_called_once_with("Foo Device") + + def test_start_and_stop_app_detection(self): + self.controller.start_app_detection() + self.data_manager.start_app_detection.assert_called_once() + self.controller.stop_app_detection() + self.data_manager.stop_app_detection.assert_called_once() diff --git a/tests/unit/test_data_manager.py b/tests/unit/test_data_manager.py index fefbcde..a9e2baf 100644 --- a/tests/unit/test_data_manager.py +++ b/tests/unit/test_data_manager.py @@ -969,3 +969,168 @@ class TestDataManager(unittest.TestCase): def test_cannot_get_injector_state_without_group(self): self.assertRaises(DataManagementError, self.data_manager.get_state) + + +class FakeFocusServiceClient: + """A FocusServiceClient stand-in that never touches the session bus.""" + + def __init__(self, status=None): + self._status = status + self.calls: List = [] + self.handler = None + + def get_status(self): + self.calls.append("get_status") + return self._status + + def reload(self): + self.calls.append("reload") + + def set_enabled(self, enabled): + self.calls.append(("set_enabled", enabled)) + + def connect_focus_changed(self, callback): + self.handler = callback + self.calls.append("connect") + return True + + def disconnect_focus_changed(self): + self.calls.append("disconnect") + self.handler = None + + +@test_setup +class TestDataManagerAppBindings(unittest.TestCase): + def _make_data_manager(self, status=None) -> DataManager: + message_broker = MessageBroker() + uinputs = GlobalUInputs(FrontendUInput) + uinputs.prepare_all() + self.focus_service = FakeFocusServiceClient(status=status) + return DataManager( + message_broker, + GlobalConfig(), + ReaderClient(message_broker, _Groups()), + FakeDaemonProxy(), + uinputs, + keyboard_layout, + focus_service=self.focus_service, + ) + + def setUp(self) -> None: + self.data_manager = self._make_data_manager() + self.message_broker = self.data_manager.message_broker + + @staticmethod + def _binding(app_id="firefox", presets=()): + from inputremapper.configs.app_binding import AppBinding, BoundPreset + + return AppBinding( + app_id=app_id, + presets=[BoundPreset(group_key=g, preset=p) for g, p in presets], + ) + + def test_publish_app_bindings_unreachable(self): + from inputremapper.gui.messages.message_data import AppBindingsData + + listener = Listener() + self.message_broker.subscribe(MessageType.app_bindings, listener) + self.data_manager.publish_app_bindings() + + data: AppBindingsData = listener.calls[0] + self.assertFalse(data.reachable) + self.assertFalse(data.supported) + self.assertIsNone(data.backend) + self.assertFalse(data.enabled) + self.assertEqual(data.bindings, ()) + + def test_publish_app_bindings_reachable(self): + self.data_manager = self._make_data_manager(status={"backend": "xorg"}) + self.message_broker = self.data_manager.message_broker + listener = Listener() + self.message_broker.subscribe(MessageType.app_bindings, listener) + self.data_manager.publish_app_bindings() + + data = listener.calls[0] + self.assertTrue(data.reachable) + self.assertTrue(data.supported) + self.assertEqual(data.backend, "xorg") + + def test_publish_app_bindings_reachable_but_unsupported(self): + # service reachable but reports no backend (e.g. GNOME-Wayland / null) + self.data_manager = self._make_data_manager(status={"backend": None}) + self.message_broker = self.data_manager.message_broker + listener = Listener() + self.message_broker.subscribe(MessageType.app_bindings, listener) + self.data_manager.publish_app_bindings() + + data = listener.calls[0] + self.assertTrue(data.reachable) + self.assertFalse(data.supported) + + def test_set_and_get_app_bindings(self): + bindings = [self._binding("firefox", [("Foo Device", "preset1")])] + listener = Listener() + self.message_broker.subscribe(MessageType.app_bindings, listener) + + self.data_manager.set_app_bindings(bindings) + + self.assertEqual(self.data_manager.get_app_bindings(), bindings) + self.assertIn("reload", self.focus_service.calls) + # publishes an app_bindings update + self.assertEqual(listener.calls[-1].bindings, tuple(bindings)) + + def test_set_and_get_app_binding_enabled(self): + listener = Listener() + self.message_broker.subscribe(MessageType.app_bindings, listener) + + self.data_manager.set_app_binding_enabled(True) + + self.assertTrue(self.data_manager.get_app_binding_enabled()) + self.assertIn(("set_enabled", True), self.focus_service.calls) + self.assertTrue(listener.calls[-1].enabled) + + def test_start_and_stop_app_detection(self): + self.data_manager.start_app_detection() + self.assertIn("connect", self.focus_service.calls) + self.assertIsNotNone(self.focus_service.handler) + + self.data_manager.stop_app_detection() + self.assertIn("disconnect", self.focus_service.calls) + self.assertIsNone(self.focus_service.handler) + + def test_on_focus_changed_publishes_focused_app(self): + from inputremapper.gui.messages.message_data import FocusAppData + + listener = Listener() + self.message_broker.subscribe(MessageType.focused_app, listener) + + self.data_manager.start_app_detection() + # simulate the service emitting a focus_changed signal + self.focus_service.handler({"app_id": "firefox", "title": "Mozilla"}) + + data: FocusAppData = listener.calls[0] + self.assertEqual(data.app_id, "firefox") + self.assertEqual(data.title, "Mozilla") + + def test_on_focus_changed_handles_missing_fields(self): + from inputremapper.gui.messages.message_data import FocusAppData + + listener = Listener() + self.message_broker.subscribe(MessageType.focused_app, listener) + self.data_manager.start_app_detection() + self.focus_service.handler({}) + + data: FocusAppData = listener.calls[0] + self.assertEqual(data.app_id, "") + self.assertEqual(data.title, "") + + def test_get_presets_for_group(self): + prepare_presets() + presets = self.data_manager.get_presets_for_group("Foo Device") + # newest first + self.assertEqual(presets, ("preset3", "preset2", "preset1")) + + def test_get_presets_for_unknown_group(self): + self.assertEqual( + self.data_manager.get_presets_for_group("Does Not Exist"), tuple() + ) diff --git a/tests/unit/test_focus_service.py b/tests/unit/test_focus_service.py new file mode 100644 index 0000000..aca9a63 --- /dev/null +++ b/tests/unit/test_focus_service.py @@ -0,0 +1,342 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +# input-remapper - GUI for device specific keyboard mappings +# Copyright (C) 2025 sezanzeb +# +# This file is part of input-remapper. +# +# input-remapper is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# input-remapper is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with input-remapper. If not, see . + +import json +import unittest +from typing import List, Optional +from unittest.mock import MagicMock + +from inputremapper.configs.app_binding import AppBinding, BoundPreset, MatchType +from inputremapper.configs.global_config import GlobalConfig +from inputremapper.focus.focus_backend import FocusBackend, FocusEvent +from inputremapper.focus.focus_service import FocusService +from tests.lib.patches import FakeDaemonProxy +from tests.lib.test_setup import test_setup + + +def _event(app_id="firefox", title="", backend="fake") -> FocusEvent: + return FocusEvent(app_id=app_id, title=title, backend=backend) + + +class FakeBackend(FocusBackend): + name = "fake" + + def __init__(self, current: Optional[FocusEvent] = None): + self._current = current + self.started = False + self.stopped = False + self._on_focus = None + self.calls = [] + + @staticmethod + def is_available() -> bool: + return True + + def start(self, on_focus) -> None: + self.started = True + self._on_focus = on_focus + self.calls.append("start") + + def stop(self) -> None: + self.stopped = True + + def get_current(self) -> Optional[FocusEvent]: + self.calls.append("get_current") + return self._current + + def emit(self, event: FocusEvent) -> None: + assert self._on_focus is not None + self._on_focus(event) + + +@test_setup +class TestFocusService(unittest.TestCase): + def setUp(self) -> None: + self.daemon = FakeDaemonProxy() + self.backend = FakeBackend() + + def _make_service( + self, + bindings: Optional[List[AppBinding]] = None, + enabled: bool = True, + ) -> FocusService: + config = GlobalConfig() + config.set_app_bindings(bindings or []) + config.set_app_binding_enabled(enabled) + service = FocusService(config, self.backend, self.daemon, debounce_ms=10) + # never touch the real session bus in tests + service.publish = MagicMock() + self.addCleanup(service.stop) + return service + + @staticmethod + def _binding(app_id, presets, match=MatchType.wm_class) -> AppBinding: + return AppBinding( + app_id=app_id, + match=match, + presets=[BoundPreset(group_key=g, preset=p) for g, p in presets], + ) + + # -- reconciliation ---------------------------------------------------- + + def test_loads_config_on_init(self): + service = self._make_service( + [self._binding("firefox", [("devX", "p1")])], enabled=True + ) + self.assertTrue(service._enabled) + self.assertEqual(len(service._bindings), 1) + + def test_focus_bound_app_starts_injecting(self): + service = self._make_service( + [self._binding("firefox", [("devX", "p1"), ("devY", "p2")])] + ) + service._reconcile(_event("firefox")) + self.assertCountEqual( + self.daemon.calls["start_injecting"], + [("devX", "p1"), ("devY", "p2")], + ) + self.assertEqual(service._app_controlled, {"devX": "p1", "devY": "p2"}) + + def test_switching_apps_replaces_and_stops(self): + service = self._make_service( + [ + self._binding("a", [("devX", "p1")]), + self._binding("b", [("devX", "p2"), ("devY", "p3")]), + self._binding("c", [("devY", "p3")]), + ] + ) + service._reconcile(_event("a")) + self.assertEqual(service._app_controlled, {"devX": "p1"}) + + # switch to b: devX replaced (p1->p2), devY added, nothing stopped + self.daemon.calls["start_injecting"].clear() + service._reconcile(_event("b")) + self.assertCountEqual( + self.daemon.calls["start_injecting"], [("devX", "p2"), ("devY", "p3")] + ) + self.assertEqual(self.daemon.calls["stop_injecting"], []) + self.assertEqual(service._app_controlled, {"devX": "p2", "devY": "p3"}) + + # switch to c: only devY wanted (already correct), devX stopped + self.daemon.calls["start_injecting"].clear() + service._reconcile(_event("c")) + self.assertEqual(self.daemon.calls["stop_injecting"], ["devX"]) + # devY p3 already running -> not restarted + self.assertEqual(self.daemon.calls["start_injecting"], []) + self.assertEqual(service._app_controlled, {"devY": "p3"}) + + def test_focus_unbound_app_stops_everything(self): + service = self._make_service([self._binding("firefox", [("devX", "p1")])]) + service._reconcile(_event("firefox")) + self.daemon.calls["start_injecting"].clear() + + service._reconcile(_event("someuncontrolledapp")) + self.assertEqual(self.daemon.calls["stop_injecting"], ["devX"]) + self.assertEqual(service._app_controlled, {}) + + def test_replace_does_not_restart_identical(self): + service = self._make_service([self._binding("firefox", [("devX", "p1")])]) + service._reconcile(_event("firefox")) + service._reconcile(_event("firefox")) + # only injected once even though reconciled twice + self.assertEqual(self.daemon.calls["start_injecting"], [("devX", "p1")]) + + def test_start_injecting_failure_not_tracked(self): + service = self._make_service([self._binding("firefox", [("devX", "p1")])]) + self.daemon.start_injecting = MagicMock(return_value=False) + service._reconcile(_event("firefox")) + self.assertEqual(service._app_controlled, {}) + + # -- manual override --------------------------------------------------- + + def test_manual_preset_is_not_stopped(self): + # a preset applied manually is not in _app_controlled, so focusing an + # unbound app must not stop it. + service = self._make_service([self._binding("firefox", [("devX", "p1")])]) + # devY has a manually applied preset, unknown to the service + service._reconcile(_event("unbound")) + self.assertEqual(self.daemon.calls["stop_injecting"], []) + + def test_bound_app_reclaims_manually_used_device(self): + # another bound app taking focus replaces whatever is on the device + service = self._make_service([self._binding("firefox", [("devX", "p9")])]) + service._reconcile(_event("firefox")) + self.assertEqual(self.daemon.calls["start_injecting"], [("devX", "p9")]) + + # -- enable / disable -------------------------------------------------- + + def test_set_enabled_false_stops_app_controlled(self): + service = self._make_service([self._binding("firefox", [("devX", "p1")])]) + service._current_event = _event("firefox") + service._reconcile(service._current_event) + self.assertEqual(service._app_controlled, {"devX": "p1"}) + + service.set_enabled(False) + self.assertFalse(service._enabled) + self.assertEqual(self.daemon.calls["stop_injecting"], ["devX"]) + self.assertEqual(service._app_controlled, {}) + + def test_set_enabled_true_reconciles_current(self): + service = self._make_service( + [self._binding("firefox", [("devX", "p1")])], enabled=False + ) + service._current_event = _event("firefox") + service.set_enabled(True) + self.assertTrue(service._enabled) + self.assertEqual(self.daemon.calls["start_injecting"], [("devX", "p1")]) + + def test_set_enabled_true_without_current_event(self): + service = self._make_service( + [self._binding("firefox", [("devX", "p1")])], enabled=False + ) + service.set_enabled(True) + self.assertEqual(self.daemon.calls["start_injecting"], []) + + # -- reload ------------------------------------------------------------ + + def test_reload_picks_up_new_bindings(self): + service = self._make_service([], enabled=True) + service._current_event = _event("firefox") + + service.global_config.set_app_bindings( + [self._binding("firefox", [("devX", "p1")])] + ) + service.reload() + self.assertEqual(self.daemon.calls["start_injecting"], [("devX", "p1")]) + + def test_reload_when_disabled_stops_app_controlled(self): + service = self._make_service([self._binding("firefox", [("devX", "p1")])]) + service._current_event = _event("firefox") + service._reconcile(service._current_event) + + service.global_config.set_app_binding_enabled(False) + service.reload() + self.assertEqual(self.daemon.calls["stop_injecting"], ["devX"]) + self.assertEqual(service._app_controlled, {}) + + # -- debounce / transient focus --------------------------------------- + + def test_transient_empty_focus_is_ignored(self): + service = self._make_service([self._binding("firefox", [("devX", "p1")])]) + service._on_focus(_event(app_id="")) + self.assertIsNone(service._pending_event) + self.assertIsNone(service._debounce_id) + self.assertEqual(self.daemon.calls["start_injecting"], []) + + def test_debounce_collapses_rapid_focus(self): + service = self._make_service( + [ + self._binding("a", [("devX", "p1")]), + self._binding("b", [("devX", "p2")]), + ] + ) + service._on_focus(_event("a")) + service._on_focus(_event("b")) + # nothing reconciled until the debounce flushes + self.assertEqual(self.daemon.calls["start_injecting"], []) + self.assertEqual(service._pending_event, _event("b")) + + service._flush() + # only the most recent focus is applied + self.assertEqual(self.daemon.calls["start_injecting"], [("devX", "p2")]) + self.assertEqual(service._current_event, _event("b")) + + def test_flush_without_pending_is_noop(self): + service = self._make_service([]) + self.assertFalse(service._flush()) + + def test_flush_emits_focus_changed_on_change(self): + service = self._make_service([]) + service.focus_changed = MagicMock() + service._pending_event = _event("firefox", title="Mozilla") + service._flush() + service.focus_changed.assert_called_once() + payload = json.loads(service.focus_changed.call_args[0][0]) + self.assertEqual(payload, {"app_id": "firefox", "title": "Mozilla"}) + + def test_flush_no_emit_when_unchanged(self): + service = self._make_service([]) + service._current_event = _event("firefox") + service.focus_changed = MagicMock() + service._pending_event = _event("firefox") + service._flush() + service.focus_changed.assert_not_called() + + # -- lifecycle / dbus surface ----------------------------------------- + + def test_start_reconciles_initial_focus(self): + self.backend = FakeBackend(current=_event("firefox")) + service = self._make_service([self._binding("firefox", [("devX", "p1")])]) + service.start() + self.assertTrue(self.backend.started) + self.assertEqual(self.backend.calls[:2], ["start", "get_current"]) + self.assertEqual(self.daemon.calls["start_injecting"], [("devX", "p1")]) + + def test_start_stops_on_publish_failure(self): + service = self._make_service([]) + service.publish = MagicMock(side_effect=RuntimeError("dbus failed")) + with self.assertRaises(RuntimeError): + service.start() + self.assertFalse(self.backend.started) + + def test_stop_releases_app_controlled(self): + service = self._make_service([self._binding("firefox", [("devX", "p1")])]) + service._reconcile(_event("firefox")) + service.stop() + self.assertTrue(self.backend.stopped) + self.assertIn("devX", self.daemon.calls["stop_injecting"]) + self.assertEqual(service._app_controlled, {}) + + def test_get_focused_app_json(self): + service = self._make_service([]) + self.assertEqual( + json.loads(service.get_focused_app()), {"app_id": "", "title": ""} + ) + service._current_event = _event("firefox", title="Moz") + self.assertEqual( + json.loads(service.get_focused_app()), + {"app_id": "firefox", "title": "Moz"}, + ) + + def test_get_status_json(self): + service = self._make_service([self._binding("firefox", [("devX", "p1")])]) + service._reconcile(_event("firefox", title="Moz")) + status = json.loads(service.get_status()) + self.assertEqual(status["backend"], "fake") + self.assertTrue(status["enabled"]) + self.assertEqual(status["focused"], {"app_id": "", "title": ""}) + self.assertEqual(status["app_controlled"], {"devX": "p1"}) + + def test_compute_desired_merges_multiple_bindings(self): + # two bindings can match the same event (wm_class + title regex) + service = self._make_service( + [ + self._binding("firefox", [("devX", "p1")]), + self._binding( + "Moz", [("devY", "p2")], match=MatchType.title_regex + ), + ] + ) + desired = service._compute_desired(_event("firefox", title="Mozilla")) + self.assertEqual(desired, {"devX": "p1", "devY": "p2"}) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/unit/test_focus_service_client.py b/tests/unit/test_focus_service_client.py new file mode 100644 index 0000000..6381fc4 --- /dev/null +++ b/tests/unit/test_focus_service_client.py @@ -0,0 +1,173 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +# input-remapper - GUI for device specific keyboard mappings +# Copyright (C) 2025 sezanzeb +# +# This file is part of input-remapper. +# +# input-remapper is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# input-remapper is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with input-remapper. If not, see . + +import unittest +from unittest.mock import MagicMock, patch + +from inputremapper.gui.focus_service_client import ( + FocusServiceClient, + ensure_focus_service_running, +) +from tests.lib.test_setup import test_setup + + +@test_setup +class TestFocusServiceClient(unittest.TestCase): + def setUp(self) -> None: + self.bus = MagicMock() + self.proxy = MagicMock() + self.bus.get_proxy.return_value = self.proxy + self.client = FocusServiceClient(bus=self.bus) + + # -- get_status -------------------------------------------------------- + + def test_get_status_success(self): + self.proxy.get_status.return_value = '{"backend": "xorg", "enabled": true}' + status = self.client.get_status() + self.assertEqual(status, {"backend": "xorg", "enabled": True}) + + def test_get_status_failure_returns_none_and_resets(self): + self.proxy.get_status.side_effect = Exception("no service") + self.assertIsNone(self.client.get_status()) + # the cached proxy is dropped so the next call reconnects + self.assertIsNone(self.client._proxy) + + def test_get_focused_app_success(self): + self.proxy.get_focused_app.return_value = ( + '{"app_id": "firefox", "title": "Moz"}' + ) + self.assertEqual( + self.client.get_focused_app(), {"app_id": "firefox", "title": "Moz"} + ) + + def test_get_focused_app_failure_returns_empty(self): + self.proxy.get_focused_app.side_effect = Exception("boom") + self.assertEqual( + self.client.get_focused_app(), {"app_id": "", "title": ""} + ) + + # -- reload / set_enabled (fire and forget) ---------------------------- + + def test_reload_success(self): + self.client.reload() + self.proxy.reload.assert_called_once() + + def test_reload_failure_is_swallowed(self): + self.proxy.reload.side_effect = Exception("boom") + # must not raise + self.client.reload() + self.assertIsNone(self.client._proxy) + + def test_set_enabled(self): + self.client.set_enabled(True) + self.proxy.set_enabled.assert_called_once_with(True) + + def test_set_enabled_failure_is_swallowed(self): + self.proxy.set_enabled.side_effect = Exception("boom") + self.client.set_enabled(False) + self.assertIsNone(self.client._proxy) + + # -- focus_changed signal --------------------------------------------- + + def test_connect_focus_changed_success(self): + received = [] + ok = self.client.connect_focus_changed(received.append) + self.assertTrue(ok) + self.proxy.focus_changed.connect.assert_called_once() + + # the wrapped handler parses the json payload before calling back + handler = self.proxy.focus_changed.connect.call_args[0][0] + handler('{"app_id": "firefox", "title": "Moz"}') + self.assertEqual(received, [{"app_id": "firefox", "title": "Moz"}]) + + def test_connect_focus_changed_invalid_payload_does_not_raise(self): + received = [] + self.client.connect_focus_changed(received.append) + handler = self.proxy.focus_changed.connect.call_args[0][0] + # invalid json must be swallowed (no callback, no exception) + handler("not json") + self.assertEqual(received, []) + + def test_connect_focus_changed_failure(self): + self.proxy.focus_changed.connect.side_effect = Exception("boom") + ok = self.client.connect_focus_changed(lambda app: None) + self.assertFalse(ok) + self.assertIsNone(self.client._proxy) + + def test_disconnect_without_handler_is_noop(self): + self.client.disconnect_focus_changed() + self.proxy.focus_changed.disconnect.assert_not_called() + + def test_disconnect_after_connect(self): + self.client.connect_focus_changed(lambda app: None) + self.client.disconnect_focus_changed() + self.proxy.focus_changed.disconnect.assert_called_once() + self.assertIsNone(self.client._focus_changed_handler) + + def test_disconnect_failure_clears_handler(self): + self.client.connect_focus_changed(lambda app: None) + self.proxy.focus_changed.disconnect.side_effect = Exception("boom") + self.client.disconnect_focus_changed() + # handler is cleared even if the disconnect failed + self.assertIsNone(self.client._focus_changed_handler) + + +@test_setup +class TestEnsureFocusServiceRunning(unittest.TestCase): + def test_does_not_start_when_already_running(self): + with patch( + "inputremapper.gui.focus_service_client.ProcessUtils." + "count_python_processes", + return_value=1, + ), patch( + "inputremapper.gui.focus_service_client.subprocess.Popen" + ) as popen: + ensure_focus_service_running() + popen.assert_not_called() + + def test_starts_when_not_running(self): + with patch( + "inputremapper.gui.focus_service_client.ProcessUtils." + "count_python_processes", + return_value=0, + ), patch( + "inputremapper.gui.focus_service_client.subprocess.Popen" + ) as popen: + ensure_focus_service_running() + popen.assert_called_once() + self.assertEqual( + popen.call_args[0][0], ["input-remapper-focus-service"] + ) + + def test_start_failure_is_swallowed(self): + with patch( + "inputremapper.gui.focus_service_client.ProcessUtils." + "count_python_processes", + return_value=0, + ), patch( + "inputremapper.gui.focus_service_client.subprocess.Popen", + side_effect=OSError("boom"), + ): + # must not raise + ensure_focus_service_running() + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/unit/test_focus_watcher.py b/tests/unit/test_focus_watcher.py new file mode 100644 index 0000000..a30df6e --- /dev/null +++ b/tests/unit/test_focus_watcher.py @@ -0,0 +1,326 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +# input-remapper - GUI for device specific keyboard mappings +# Copyright (C) 2025 sezanzeb +# +# This file is part of input-remapper. +# +# input-remapper is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# input-remapper is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with input-remapper. If not, see . + +import json +import unittest +from unittest.mock import patch + +from inputremapper.focus.backends.hyprland import HyprlandFocusBackend +from inputremapper.focus.backends.null import NullBackend +from inputremapper.focus.backends.sway import ( + SwayFocusBackend, + _container_to_event, + _find_focused, +) +from inputremapper.focus.backends.wlr_foreign_toplevel import ( + WlrForeignToplevelBackend, +) +from inputremapper.focus.backends.xorg import XorgFocusBackend +from inputremapper.focus.focus_watcher import focus_backend_classes, select_backend +from tests.lib.test_setup import test_setup + + +@test_setup +class TestSelectBackend(unittest.TestCase): + def _patch_availability(self, available, exclude=()): + """Patch is_available of every backend class from a {name: bool} map.""" + patchers = [] + excluded = set(exclude) + for backend_class in focus_backend_classes: + if backend_class.name in excluded: + continue + value = available.get(backend_class.name, False) + patcher = patch.object( + backend_class, + "is_available", + staticmethod(lambda v=value: v), + ) + patcher.start() + patchers.append(patcher) + self.addCleanup(lambda: [p.stop() for p in patchers]) + + def test_priority_order(self): + # registry order: sway, hyprland, wlr, xorg, null + self.assertEqual( + [c.name for c in focus_backend_classes], + ["sway", "hyprland", "wlr-foreign-toplevel", "xorg", "null"], + ) + + def test_selects_sway_first(self): + self._patch_availability({"sway": True, "xorg": True}) + self.assertIsInstance(select_backend(), SwayFocusBackend) + + def test_selects_hyprland_when_no_sway(self): + self._patch_availability({"hyprland": True, "xorg": True}) + self.assertIsInstance(select_backend(), HyprlandFocusBackend) + + def test_selects_xorg_when_only_xorg(self): + self._patch_availability({"xorg": True}) + self.assertIsInstance(select_backend(), XorgFocusBackend) + + def test_falls_back_to_null(self): + self._patch_availability({}) + self.assertIsInstance(select_backend(), NullBackend) + + def test_availability_exception_is_swallowed(self): + def boom(): + raise RuntimeError("nope") + + with patch.object(SwayFocusBackend, "is_available", staticmethod(boom)): + self._patch_availability({"xorg": True}, exclude={"sway"}) + # sway raises -> skipped, xorg wins + self.assertIsInstance(select_backend(), XorgFocusBackend) + + +@test_setup +class TestNullBackend(unittest.TestCase): + def test_always_available(self): + self.assertTrue(NullBackend.is_available()) + + def test_get_current_is_none(self): + backend = NullBackend() + self.assertIsNone(backend.get_current()) + + def test_start_and_stop_do_nothing(self): + backend = NullBackend() + calls = [] + backend.start(lambda event: calls.append(event)) + backend.stop() + self.assertEqual(calls, []) + + +@test_setup +class TestSwayParsing(unittest.TestCase): + def test_container_to_event_app_id(self): + event = _container_to_event({"app_id": "Alacritty", "name": "shell"}) + self.assertEqual(event.app_id, "Alacritty") + self.assertEqual(event.title, "shell") + self.assertEqual(event.backend, "sway") + + def test_container_to_event_xwayland_class_fallback(self): + # XWayland windows have no app_id but a window_properties.class + event = _container_to_event( + { + "app_id": None, + "name": "Firefox", + "window_properties": {"class": "firefox"}, + } + ) + self.assertEqual(event.app_id, "firefox") + self.assertEqual(event.title, "Firefox") + + def test_container_to_event_empty_container(self): + self.assertIsNone(_container_to_event({})) + + def test_container_to_event_missing_fields(self): + event = _container_to_event({"app_id": "x"}) + self.assertEqual(event.app_id, "x") + self.assertEqual(event.title, "") + + def test_find_focused_nested(self): + tree = { + "focused": False, + "nodes": [ + {"focused": False, "nodes": []}, + { + "focused": False, + "nodes": [{"focused": True, "app_id": "target", "nodes": []}], + }, + ], + "floating_nodes": [], + } + found = _find_focused(tree) + self.assertIsNotNone(found) + self.assertEqual(found["app_id"], "target") + + def test_find_focused_floating(self): + tree = { + "focused": False, + "nodes": [], + "floating_nodes": [{"focused": True, "app_id": "floater"}], + } + self.assertEqual(_find_focused(tree)["app_id"], "floater") + + def test_find_focused_none(self): + tree = {"focused": False, "nodes": [{"focused": False}]} + self.assertIsNone(_find_focused(tree)) + + def test_handle_event_emits_on_focus_change(self): + backend = SwayFocusBackend() + received = [] + backend._on_focus = received.append + payload = json.dumps( + {"change": "focus", "container": {"app_id": "kitty", "name": "term"}} + ).encode() + backend._handle_event(payload) + self.assertEqual(len(received), 1) + self.assertEqual(received[0].app_id, "kitty") + + def test_handle_event_ignores_unrelated_change(self): + backend = SwayFocusBackend() + received = [] + backend._on_focus = received.append + payload = json.dumps( + {"change": "move", "container": {"app_id": "kitty"}} + ).encode() + backend._handle_event(payload) + self.assertEqual(received, []) + + def test_handle_event_invalid_json(self): + backend = SwayFocusBackend() + received = [] + backend._on_focus = received.append + backend._handle_event(b"\xff not json") + self.assertEqual(received, []) + + def test_is_available_without_env(self): + with patch.dict("os.environ", {}, clear=True): + self.assertFalse(SwayFocusBackend.is_available()) + + +@test_setup +class TestHyprlandParsing(unittest.TestCase): + def test_handle_line_activewindow(self): + backend = HyprlandFocusBackend() + received = [] + backend._on_focus = received.append + backend._handle_line("activewindow>>firefox,Mozilla Firefox") + self.assertEqual(len(received), 1) + self.assertEqual(received[0].app_id, "firefox") + self.assertEqual(received[0].title, "Mozilla Firefox") + self.assertEqual(received[0].backend, "hyprland") + + def test_handle_line_title_with_comma(self): + backend = HyprlandFocusBackend() + received = [] + backend._on_focus = received.append + backend._handle_line("activewindow>>code,file.py, project - VSCode") + self.assertEqual(received[0].app_id, "code") + # only the first comma splits app_id from title + self.assertEqual(received[0].title, "file.py, project - VSCode") + + def test_handle_line_ignores_other_events(self): + backend = HyprlandFocusBackend() + received = [] + backend._on_focus = received.append + backend._handle_line("workspace>>2") + backend._handle_line("openwindow>>0x1,1,firefox,Title") + self.assertEqual(received, []) + + def test_handle_line_no_separator(self): + backend = HyprlandFocusBackend() + received = [] + backend._on_focus = received.append + backend._handle_line("garbage line") + self.assertEqual(received, []) + + def test_is_available_without_signature(self): + with patch.dict("os.environ", {}, clear=True): + self.assertFalse(HyprlandFocusBackend.is_available()) + + +@test_setup +class TestXorgBackend(unittest.TestCase): + def test_is_available_without_display(self): + with patch.dict("os.environ", {}, clear=True): + self.assertFalse(XorgFocusBackend.is_available()) + + def test_is_available_without_xlib(self): + with patch.dict("os.environ", {"DISPLAY": ":0"}, clear=True): + with patch.dict("sys.modules", {"Xlib": None, "Xlib.display": None}): + # importing Xlib.display raises ImportError -> not available + self.assertFalse(XorgFocusBackend.is_available()) + + def test_get_current_without_display(self): + backend = XorgFocusBackend() + self.assertIsNone(backend.get_current()) + + def test_stop_without_start(self): + backend = XorgFocusBackend() + # must not raise even though nothing was started + backend.stop() + + +@test_setup +class TestWlrBackend(unittest.TestCase): + def test_is_available_without_wayland(self): + with patch.dict("os.environ", {}, clear=True): + self.assertFalse(WlrForeignToplevelBackend.is_available()) + + def test_is_available_without_pywayland(self): + # pywayland is not installed in the dev environment, so even with + # WAYLAND_DISPLAY set the backend must report itself unavailable cleanly. + with patch.dict("os.environ", {"WAYLAND_DISPLAY": "wayland-0"}, clear=True): + with patch( + "inputremapper.focus.backends.wlr_foreign_toplevel._import_protocol", + return_value=None, + ): + self.assertFalse(WlrForeignToplevelBackend.is_available()) + + def test_get_current_initially_none(self): + backend = WlrForeignToplevelBackend() + self.assertIsNone(backend.get_current()) + + def test_toplevel_activation_flow(self): + backend = WlrForeignToplevelBackend() + received = [] + backend._on_focus = received.append + + handle = object() + backend._toplevels[handle] = {"app_id": "", "title": "", "activated": False} + backend._on_app_id(handle, "firefox") + backend._on_title(handle, "Mozilla") + backend._on_state(handle, [2]) # _STATE_ACTIVATED + backend._on_done(handle) + + self.assertEqual(len(received), 1) + self.assertEqual(received[0].app_id, "firefox") + self.assertEqual(received[0].title, "Mozilla") + self.assertEqual(backend.get_current().app_id, "firefox") + + def test_done_without_activation_is_ignored(self): + backend = WlrForeignToplevelBackend() + received = [] + backend._on_focus = received.append + handle = object() + backend._toplevels[handle] = { + "app_id": "x", + "title": "y", + "activated": False, + } + backend._on_done(handle) + self.assertEqual(received, []) + + def test_closed_removes_toplevel(self): + backend = WlrForeignToplevelBackend() + handle = object() + backend._toplevels[handle] = {"app_id": "x", "title": "", "activated": True} + backend._on_closed(handle) + self.assertNotIn(handle, backend._toplevels) + + def test_state_unknown_handle_ignored(self): + backend = WlrForeignToplevelBackend() + # must not raise for a handle we never registered + backend._on_app_id(object(), "x") + backend._on_state(object(), [2]) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/unit/test_global_config.py b/tests/unit/test_global_config.py index 719c185..a702b8f 100644 --- a/tests/unit/test_global_config.py +++ b/tests/unit/test_global_config.py @@ -19,10 +19,12 @@ # along with input-remapper. If not, see . +import json import os import unittest from inputremapper.configs.global_config import GlobalConfig +from inputremapper.logging.logger import VERSION from tests.lib.test_setup import test_setup @@ -81,8 +83,16 @@ class TestGlobalConfig(unittest.TestCase): self.assertTrue(os.path.exists(global_config.path)) with open(global_config.path, "r") as file: - contents = file.read() - self.assertIn('"autoload": {}', contents) + config = json.load(file) + self.assertEqual( + config, + { + "version": VERSION, + "autoload": {}, + "app_binding_enabled": False, + "app_bindings": [], + }, + ) def test_save_load(self): global_config = GlobalConfig() @@ -101,6 +111,71 @@ class TestGlobalConfig(unittest.TestCase): [("d1", "a"), ("d2", "b")], ) + # -- app bindings ------------------------------------------------------ + + def test_app_binding_enabled_default_false(self): + global_config = GlobalConfig() + self.assertFalse(global_config.get_app_binding_enabled()) + + def test_set_app_binding_enabled_roundtrip(self): + global_config = GlobalConfig() + global_config.set_app_binding_enabled(True) + self.assertTrue(global_config.get_app_binding_enabled()) + + # persisted and reloadable + reloaded = GlobalConfig() + reloaded.load_config() + self.assertTrue(reloaded.get_app_binding_enabled()) + + global_config.set_app_binding_enabled(False) + self.assertFalse(global_config.get_app_binding_enabled()) + + def test_app_bindings_default_empty(self): + global_config = GlobalConfig() + self.assertEqual(global_config.get_app_bindings(), []) + + def test_set_app_bindings_roundtrip(self): + from inputremapper.configs.app_binding import ( + AppBinding, + BoundPreset, + MatchType, + ) + + bindings = [ + AppBinding( + app_id="firefox", + match=MatchType.wm_class, + presets=[BoundPreset(group_key="Foo Device", preset="preset1")], + ), + AppBinding(app_id="term.*", match=MatchType.title_regex), + ] + global_config = GlobalConfig() + global_config.set_app_bindings(bindings) + + # serialized as plain dicts in config.json + with open(global_config.path, "r") as file: + raw = json.load(file) + self.assertEqual(len(raw["app_bindings"]), 2) + self.assertEqual(raw["app_bindings"][0]["app_id"], "firefox") + + # reloaded into model objects + reloaded = GlobalConfig() + reloaded.load_config() + result = reloaded.get_app_bindings() + self.assertEqual(result, bindings) + + def test_get_app_bindings_skips_invalid(self): + global_config = GlobalConfig() + global_config.load_config() + # inject an invalid binding (empty app_id) directly into the raw config + global_config._config["app_bindings"] = [ + {"app_id": "valid", "match": "wm_class", "presets": []}, + {"app_id": "", "match": "wm_class", "presets": []}, + ] + result = global_config.get_app_bindings() + self.assertEqual(len(result), 1) + self.assertEqual(result[0].app_id, "valid") + if __name__ == "__main__": unittest.main() diff --git a/tests/unit/test_migrations.py b/tests/unit/test_migrations.py index 8bc6f11..9539fdf 100644 --- a/tests/unit/test_migrations.py +++ b/tests/unit/test_migrations.py @@ -659,7 +659,13 @@ class TestMigrations(unittest.TestCase): with open(PathUtils.get_config_path("config.json"), "r") as f: config_json = json.load(f) self.assertDictEqual( - config_json, {"autoload": {device_name: "foo"}, "version": VERSION} + config_json, + { + "autoload": {device_name: "foo"}, + "version": VERSION, + "app_binding_enabled": False, + "app_bindings": [], + }, ) with open(PathUtils.get_preset_path(device_name, "foo.json"), "r") as f: os.system(f'cat { PathUtils.get_preset_path(device_name, "foo.json") }') @@ -702,7 +708,13 @@ class TestMigrations(unittest.TestCase): with open(PathUtils.get_config_path("config.json"), "r") as f: config_json = json.load(f) self.assertDictEqual( - config_json, {"autoload": {device_name: "bar"}, "version": VERSION} + config_json, + { + "autoload": {device_name: "bar"}, + "version": VERSION, + "app_binding_enabled": False, + "app_bindings": [], + }, ) with open(PathUtils.get_preset_path(device_name, "bar.json"), "r") as f: os.system(f'cat { PathUtils.get_preset_path(device_name, "bar.json") }') @@ -721,6 +733,60 @@ class TestMigrations(unittest.TestCase): ], ) + def _write_config(self, data): + config_file = os.path.join(PathUtils.config_path(), "config.json") + PathUtils.touch(config_file) + with open(config_file, "w") as file: + json.dump(data, file, indent=4) + return config_file + + def _read_config(self, config_file): + with open(config_file, "r") as file: + return json.load(file) + + def test_add_app_binding_config(self): + # an old config without the app-binding keys gets them added + config_file = self._write_config( + {"version": "2.2.1", "autoload": {"Foo Device 2": "preset2"}} + ) + self.migrations.migrate() + + config = self._read_config(config_file) + self.assertEqual(config["app_binding_enabled"], False) + self.assertEqual(config["app_bindings"], []) + # additive: existing keys are preserved + self.assertEqual(config["autoload"], {"Foo Device 2": "preset2"}) + self.assertEqual(config["version"], VERSION) + + def test_add_app_binding_config_preserves_existing(self): + # a config already carrying app-binding data must not be reset + existing_binding = { + "app_id": "firefox", + "match": "wm_class", + "presets": [{"group_key": "Foo Device", "preset": "preset1"}], + } + config_file = self._write_config( + { + "version": "2.2.1", + "autoload": {}, + "app_binding_enabled": True, + "app_bindings": [existing_binding], + } + ) + self.migrations.migrate() + + config = self._read_config(config_file) + self.assertEqual(config["app_binding_enabled"], True) + self.assertEqual(config["app_bindings"], [existing_binding]) + + def test_add_app_binding_config_no_file(self): + # no config.json yet -> migration is a no-op and does not create one + config_file = os.path.join(PathUtils.config_path(), "config.json") + if os.path.exists(config_file): + os.remove(config_file) + self.migrations._add_app_binding_config() + self.assertFalse(os.path.exists(config_file)) + if __name__ == "__main__": unittest.main()