diff --git a/README.md b/README.md index 7c6a64d..be50a2a 100644 --- a/README.md +++ b/README.md @@ -109,7 +109,7 @@ The status commands verify managed integration configuration and, where applicab `setup-greeter` shows an arrow-key menu of installed supported managers. Scripts can select one directly with `--manager plasma-login`, `--manager greetd`, or `--manager lightdm`. Setup checks the selected manager and its current configuration before writing files. It does not restart the display manager, so reboot or restart it after setup. -Plasma Login Manager and LightDM use their own greeter startup hooks. The greetd adapter saves and wraps the existing default-session command. It starts that command before Axidev OSK, keeps the greeter authoritative, and restores the exact command during removal. +Plasma Login Manager configures Axidev OSK as KWin's input method for the login manager and session lock screen. Plasma 6.7 setup also adds an exact marked block to Plasma's lock-screen QML so the password interface remains visible while using the keyboard. This integration is limited to Plasma versions from 6.7.0 up to, but not including, 7.0.0. LightDM uses its greeter startup hook. The greetd adapter saves and wraps the existing default-session command, starts that command before Axidev OSK, and restores the exact command during removal. The keyboard retries with exponential backoff while the greeter remains active. A keyboard or display-detection failure never stops the greeter. Failures appear in the system journal under `axidev-osk-greeter`. diff --git a/packaging/linux/README.md b/packaging/linux/README.md index 11e7511..3811fd8 100644 --- a/packaging/linux/README.md +++ b/packaging/linux/README.md @@ -160,7 +160,9 @@ sudo axidev-osk linux setup-greeter --manager lightdm Setup validates every required account, hook, and existing file before changing the manager. It configures one manager and never restarts it. Reboot or restart the selected display manager after setup. -Plasma Login Manager uses an `axidev-osk-greeter.service` user unit tied to `plasma-login-wayland.target`. LightDM uses a `greeter-wrapper` drop-in. greetd replaces only the default-session command line with `/etc/axidev-osk/greetd-session-wrapper`, while `/etc/axidev-osk/greeter.json` keeps the exact original value for removal. +Plasma Login Manager configures Axidev OSK as KWin's input method for both the login manager and the session lock screen. On Plasma versions from 6.7.0 up to, but not including, 7.0.0, setup also adds one marked block to `/usr/share/plasma/shells/org.kde.plasma.desktop/contents/lockscreen/LockScreenUi.qml`. The block keeps the password interface visible while the pointer uses Axidev OSK. Running setup again restores a block removed by a Plasma package update before reporting any other managed-file drift. Setup and removal refuse partial or edited markers. + +LightDM uses a `greeter-wrapper` drop-in. greetd replaces only the default-session command line with `/etc/axidev-osk/greetd-session-wrapper`, while `/etc/axidev-osk/greeter.json` keeps the exact original value for removal. The LightDM and greetd shell wrappers start the original greeter before invoking the Axidev launcher. A missing Python package, incompatible Qt runtime, keyboard crash, or display-detection error cannot stop the original greeter. The keyboard retries after 1, 2, 4, 8, 16, 32, and then 60 seconds until the greeter exits. @@ -203,7 +205,7 @@ Uninstall stops when permission or autostart cleanup fails. `uninstall --force` Permission setup manages `/etc/modules-load.d/axidev-osk-uinput.conf` and `/etc/udev/rules.d/70-axidev-io-uinput.rules` through the application command line. Removal deletes only files whose contents still match Axidev OSK's definitions, and it does not unload the shared `uinput` module. Autostart setup manages the selected user's XDG autostart file. -Files owned by a configured manager are conditional. `/etc/axidev-osk/greeter.json` records the selected adapter. Plasma Login Manager owns its user service and target link. LightDM owns its drop-in and wrapper. greetd owns its wrapper and restores the previous command during removal. +Files owned by a configured manager are conditional. `/etc/axidev-osk/greeter.json` records the selected adapter. Plasma Login Manager owns its KWin input-method desktop entry, systemd drop-in, and KWin configuration. It also adds and removes only the exact marked block in Plasma's lock-screen QML. LightDM owns its drop-in and wrapper. greetd owns its wrapper and restores the previous command during removal. The uninstaller removes exact managed greeter files before it disables the Axidev OSK uinput rule. It stops on changed files unless `--force` is selected. Shared `uinput` memberships remain in place. diff --git a/src/axidev_osk/app.py b/src/axidev_osk/app.py index 0fd59f7..690100f 100644 --- a/src/axidev_osk/app.py +++ b/src/axidev_osk/app.py @@ -4,6 +4,7 @@ import ctypes import logging +import os import sys from importlib.metadata import PackageNotFoundError, version from importlib.resources import files @@ -12,8 +13,11 @@ from PySide6.QtWidgets import QApplication from .runtime.application import ApplicationRuntime +from .runtime.registries import ServiceRegistry +from .services.keyboard import KeyboardService +from .services.kwin_lock import KWinLockService from .services.single_instance import ExistingInstanceActivated -from .windows.overlay import prepare_always_on_top_window_environment +from .windows.overlay import OverlayBackend, prepare_always_on_top_window_environment _logger = logging.getLogger(__name__) @@ -48,6 +52,21 @@ def _set_application_icon(app: QApplication) -> None: app.setWindowIcon(icon) +def _input_panel_services( + app: QApplication, + backend: OverlayBackend, + *, + lock_lifecycle: bool, +) -> ServiceRegistry | None: + if backend != OverlayBackend.WAYLAND_INPUT_PANEL: + return None + services = ServiceRegistry() + services.register("keyboard", KeyboardService(), autostart=not lock_lifecycle) + if lock_lifecycle: + services.register("kwin_lock", KWinLockService(parent=app)) + return services + + def main() -> int: """Run the Axidev OSK Qt application. @@ -68,12 +87,21 @@ def main() -> int: ) _set_process_name("axidev-osk") _logger.info("Starting axidev-osk v%s", _package_version()) - prepare_always_on_top_window_environment() + overlay_backend = prepare_always_on_top_window_environment() + lock_lifecycle = ( + overlay_backend == OverlayBackend.WAYLAND_INPUT_PANEL + and os.environ.get("AXIDEV_OSK_GREETER") != "1" + ) app = QApplication(sys.argv) app.setApplicationName("axidev-osk") _set_application_icon(app) app.setQuitOnLastWindowClosed(False) - runtime = ApplicationRuntime(app) + runtime = ApplicationRuntime( + app, + services=_input_panel_services(app, overlay_backend, lock_lifecycle=lock_lifecycle), + confirm_quit=overlay_backend != OverlayBackend.WAYLAND_INPUT_PANEL, + show_startup_windows=not lock_lifecycle, + ) try: return runtime.start() except ExistingInstanceActivated: diff --git a/src/axidev_osk/cli/linux_greeter.py b/src/axidev_osk/cli/linux_greeter.py index 05c8072..2656650 100644 --- a/src/axidev_osk/cli/linux_greeter.py +++ b/src/axidev_osk/cli/linux_greeter.py @@ -32,6 +32,21 @@ PLASMA_WANTS_PATH = Path( "/etc/systemd/user/plasma-login-wayland.target.wants/axidev-osk-greeter.service" ) +PLASMA_INPUT_METHOD_PATH = Path( + "/usr/local/share/applications/axidev-osk-input-panel.desktop" +) +KWIN_CONFIG_PATH = Path("/etc/xdg/kwinrc") +PLASMA_LOCK_SCREEN_UI_PATH = Path( + "/usr/share/plasma/shells/org.kde.plasma.desktop/contents/lockscreen/LockScreenUi.qml" +) +PLASMA_KWIN_UNIT_PATHS = ( + Path("/usr/local/lib/systemd/user/plasma-login-kwin_wayland.service"), + Path("/usr/lib/systemd/user/plasma-login-kwin_wayland.service"), + Path("/lib/systemd/user/plasma-login-kwin_wayland.service"), +) +PLASMA_KWIN_DROPIN_PATH = Path( + "/etc/systemd/user/plasma-login-kwin_wayland.service.d/50-axidev-osk.conf" +) LIGHTDM_CONFIG_PATH = Path("/etc/lightdm/lightdm.conf.d/99-axidev-osk.conf") LIGHTDM_WRAPPER_PATH = Path("/etc/axidev-osk/lightdm-greeter-wrapper") GREETD_WRAPPER_PATH = Path("/etc/axidev-osk/greetd-session-wrapper") @@ -47,6 +62,24 @@ HEALTHY_RUNTIME_SECONDS = 60.0 POLL_SECONDS = 0.1 +PLASMA_LOCK_SCREEN_PATCH_START = "// BEGIN AXIDEV OSK MANAGED" +PLASMA_LOCK_SCREEN_PATCH_END = "// END AXIDEV OSK MANAGED" +PLASMA_LOCK_SCREEN_MIN_VERSION = (6, 7, 0) +PLASMA_LOCK_SCREEN_MAX_VERSION = (7, 0, 0) +PLASMA_LOCK_SCREEN_PATCH = ( + " // BEGIN AXIDEV OSK MANAGED\n" + " Connections {\n" + " target: lockScreenRoot\n" + " Component.onCompleted: lockScreenRoot.uiVisible = true\n\n" + " function onUiVisibleChanged() {\n" + " if (!lockScreenRoot.uiVisible) {\n" + " lockScreenRoot.uiVisible = true;\n" + " }\n" + " }\n" + " }\n" + " // END AXIDEV OSK MANAGED\n" +) + @dataclass(frozen=True) class GreetdConfig: """The exact greetd command assignment that setup may replace.""" @@ -84,6 +117,10 @@ def symlink(self, path: Path, target: Path) -> None: path.unlink(missing_ok=True) path.symlink_to(target) + def remove(self, path: Path) -> None: + self._remember(path) + path.unlink(missing_ok=True) + def rollback(self) -> None: for path, kind, value, mode in reversed(self._originals): try: @@ -153,23 +190,38 @@ def run_runtime_command(namespace: argparse.Namespace, argv: list[str]) -> int: def _setup(requested_manager: str | None) -> int: existing = _load_state(required=False) - if existing is not None: + legacy_plasma = existing is not None and _is_legacy_plasma_state(existing) + if existing is not None and not legacy_plasma: if requested_manager is not None and existing["manager"] != requested_manager: raise linux.LinuxSetupError( f"greeter integration already manages {existing['manager']}; remove it first" ) + if existing["manager"] == "plasma-login": + _require_supported_plasma_lock_screen_version() + repaired_lock_screen = _repair_plasma_lock_screen_patch(existing) + if repaired_lock_screen: + print("Restored the managed Plasma lock-screen visibility block.") if _status_state(existing) == 0: print(f"Greeter startup is already configured for {existing['manager']}.") return 0 raise linux.LinuxSetupError("managed greeter state is incomplete; remove it before setup") - manager = requested_manager or _select_manager(_installed_managers()) + if legacy_plasma: + assert existing is not None + if requested_manager is not None and requested_manager != "plasma-login": + raise linux.LinuxSetupError( + "greeter integration already manages plasma-login; remove it first" + ) + manager = "plasma-login" + else: + manager = requested_manager or _select_manager(_installed_managers()) adapter = _manager_adapter(manager) if not _manager_installed(adapter): raise linux.LinuxSetupError(f"{adapter.label} is not installed") launcher = _installed_launcher() account, details = adapter.prepare(launcher) + details["legacy_plasma"] = legacy_plasma linux._setup_permissions(account) _install_manager(manager, adapter, account, launcher, details) @@ -198,6 +250,28 @@ def _status_state(state: dict[str, Any]) -> int: return 0 if permission_status == 0 and all(passed for _, passed in checks) else 1 +def _repair_plasma_lock_screen_patch(state: dict[str, Any]) -> bool: + """Restore a missing managed QML block independently of other status checks.""" + + if _state_manager(state) != "plasma-login": + return False + if _plasma_lock_screen_patch_is_current(linux._read_text(PLASMA_LOCK_SCREEN_UI_PATH)): + return False + lock_screen_ui = linux._read_text(PLASMA_LOCK_SCREEN_UI_PATH) + if lock_screen_ui is None: + raise linux.LinuxSetupError( + f"Plasma lock-screen QML does not exist: {PLASMA_LOCK_SCREEN_UI_PATH}" + ) + _require_writable_regular_file(PLASMA_LOCK_SCREEN_UI_PATH) + managed = _plasma_lock_screen_ui_text(lock_screen_ui) + linux._write_atomic( + PLASMA_LOCK_SCREEN_UI_PATH, + managed, + PLASMA_LOCK_SCREEN_UI_PATH.stat().st_mode & 0o777, + ) + return True + + def _remove() -> int: state = _load_state(required=False) if state is None: @@ -316,11 +390,31 @@ def _installed_launcher() -> Path: def _prepare_plasma(launcher: Path) -> tuple[linux.Account, dict[str, Any]]: + _require_supported_plasma_lock_screen_version() account = linux._resolve_account("plasmalogin", require_home=False) - _require_compatible_file(NATIVE_SUPERVISOR_PATH, _native_supervisor_text(launcher)) - _require_compatible_file(PLASMA_SERVICE_PATH, _plasma_service_text()) - _require_compatible_symlink(PLASMA_WANTS_PATH, PLASMA_SERVICE_PATH) - return account, {} + original_kwinrc = linux._read_text(KWIN_CONFIG_PATH) + lock_screen_ui = linux._read_text(PLASMA_LOCK_SCREEN_UI_PATH) + if lock_screen_ui is None: + raise linux.LinuxSetupError( + f"Plasma lock-screen QML does not exist: {PLASMA_LOCK_SCREEN_UI_PATH}" + ) + managed_kwinrc = _plasma_kwin_config_text(original_kwinrc) + managed_lock_screen_ui = _plasma_lock_screen_ui_text(lock_screen_ui) + _require_compatible_file(PLASMA_INPUT_METHOD_PATH, _plasma_input_method_text(launcher)) + _require_compatible_file(PLASMA_KWIN_DROPIN_PATH, _plasma_kwin_dropin_text(launcher)) + if PLASMA_INPUT_METHOD_PATH.exists() or PLASMA_INPUT_METHOD_PATH.is_symlink(): + _require_writable_regular_file(PLASMA_INPUT_METHOD_PATH) + if original_kwinrc is not None: + _require_writable_regular_file(KWIN_CONFIG_PATH) + _require_writable_regular_file(PLASMA_LOCK_SCREEN_UI_PATH) + return account, { + "kwinrc_existed": original_kwinrc is not None, + "kwinrc_mode": KWIN_CONFIG_PATH.stat().st_mode & 0o777 if original_kwinrc is not None else 0o644, + "original_kwinrc": original_kwinrc or "", + "managed_kwinrc": managed_kwinrc, + "lock_screen_ui_mode": PLASMA_LOCK_SCREEN_UI_PATH.stat().st_mode & 0o777, + "managed_lock_screen_ui": managed_lock_screen_ui, + } def _prepare_lightdm(launcher: Path) -> tuple[linux.Account, dict[str, Any]]: @@ -380,11 +474,26 @@ def _install_plasma( launcher: Path, details: dict[str, Any], ) -> dict[str, Any]: - del details - transaction.write(NATIVE_SUPERVISOR_PATH, _native_supervisor_text(launcher), 0o755) - transaction.write(PLASMA_SERVICE_PATH, _plasma_service_text()) - transaction.symlink(PLASMA_WANTS_PATH, PLASMA_SERVICE_PATH) - return {} + if bool(details.get("legacy_plasma")): + _require_removable_symlink(PLASMA_WANTS_PATH, PLASMA_SERVICE_PATH) + _require_removable_file(PLASMA_SERVICE_PATH, _plasma_service_text()) + _require_removable_file(NATIVE_SUPERVISOR_PATH, _native_supervisor_text(launcher)) + transaction.remove(PLASMA_WANTS_PATH) + transaction.remove(PLASMA_SERVICE_PATH) + transaction.remove(NATIVE_SUPERVISOR_PATH) + transaction.write(PLASMA_INPUT_METHOD_PATH, _plasma_input_method_text(launcher)) + transaction.write(PLASMA_KWIN_DROPIN_PATH, _plasma_kwin_dropin_text(launcher)) + transaction.write(KWIN_CONFIG_PATH, _state_string(details, "managed_kwinrc")) + transaction.write( + PLASMA_LOCK_SCREEN_UI_PATH, + _state_string(details, "managed_lock_screen_ui"), + int(details["lock_screen_ui_mode"]), + ) + return { + "kwinrc_existed": bool(details["kwinrc_existed"]), + "kwinrc_mode": int(details["kwinrc_mode"]), + "original_kwinrc": _state_text(details, "original_kwinrc"), + } def _install_lightdm( @@ -415,13 +524,44 @@ def _install_greetd( def _check_plasma(launcher: Path, state: dict[str, Any]) -> list[tuple[str, bool]]: - del state - service_ok = linux._read_text(PLASMA_SERVICE_PATH) == _plasma_service_text() - link_ok = PLASMA_WANTS_PATH.is_symlink() and PLASMA_WANTS_PATH.resolve() == PLASMA_SERVICE_PATH + version_check = ( + "Plasma version >=6.7.0,<7.0.0", + _plasma_lock_screen_version_supported(), + ) + if _is_legacy_plasma_state(state): + service_ok = linux._read_text(PLASMA_SERVICE_PATH) == _plasma_service_text() + link_ok = ( + PLASMA_WANTS_PATH.is_symlink() + and PLASMA_WANTS_PATH.resolve() == PLASMA_SERVICE_PATH.resolve() + ) + return [ + version_check, + ( + str(NATIVE_SUPERVISOR_PATH), + linux._read_text(NATIVE_SUPERVISOR_PATH) == _native_supervisor_text(launcher), + ), + (str(PLASMA_SERVICE_PATH), service_ok), + (str(PLASMA_WANTS_PATH), link_ok), + ] + original_kwinrc = _state_text(state, "original_kwinrc") return [ - (str(NATIVE_SUPERVISOR_PATH), linux._read_text(NATIVE_SUPERVISOR_PATH) == _native_supervisor_text(launcher)), - (str(PLASMA_SERVICE_PATH), service_ok), - (str(PLASMA_WANTS_PATH), link_ok), + version_check, + ( + str(PLASMA_INPUT_METHOD_PATH), + linux._read_text(PLASMA_INPUT_METHOD_PATH) == _plasma_input_method_text(launcher), + ), + ( + str(PLASMA_KWIN_DROPIN_PATH), + linux._read_text(PLASMA_KWIN_DROPIN_PATH) == _plasma_kwin_dropin_text(launcher), + ), + ( + str(KWIN_CONFIG_PATH), + linux._read_text(KWIN_CONFIG_PATH) == _plasma_kwin_config_text(original_kwinrc or None), + ), + ( + str(PLASMA_LOCK_SCREEN_UI_PATH), + _plasma_lock_screen_patch_is_current(linux._read_text(PLASMA_LOCK_SCREEN_UI_PATH)), + ), ] @@ -452,13 +592,41 @@ def _check_greetd(launcher: Path, state: dict[str, Any]) -> list[tuple[str, bool def _remove_plasma(launcher: Path, state: dict[str, Any]) -> None: - del state - _require_removable_symlink(PLASMA_WANTS_PATH, PLASMA_SERVICE_PATH) - _require_removable_file(PLASMA_SERVICE_PATH, _plasma_service_text()) - _require_removable_file(NATIVE_SUPERVISOR_PATH, _native_supervisor_text(launcher)) - _remove_owned_symlink(PLASMA_WANTS_PATH, PLASMA_SERVICE_PATH) - linux._remove_owned_file(PLASMA_SERVICE_PATH, _plasma_service_text()) - linux._remove_owned_file(NATIVE_SUPERVISOR_PATH, _native_supervisor_text(launcher)) + if _is_legacy_plasma_state(state): + _require_removable_symlink(PLASMA_WANTS_PATH, PLASMA_SERVICE_PATH) + _require_removable_file(PLASMA_SERVICE_PATH, _plasma_service_text()) + _require_removable_file(NATIVE_SUPERVISOR_PATH, _native_supervisor_text(launcher)) + _remove_owned_symlink(PLASMA_WANTS_PATH, PLASMA_SERVICE_PATH) + linux._remove_owned_file(PLASMA_SERVICE_PATH, _plasma_service_text()) + linux._remove_owned_file(NATIVE_SUPERVISOR_PATH, _native_supervisor_text(launcher)) + return + original_kwinrc = _state_text(state, "original_kwinrc") + managed_kwinrc = _plasma_kwin_config_text(original_kwinrc or None) + _require_removable_file(PLASMA_INPUT_METHOD_PATH, _plasma_input_method_text(launcher)) + _require_removable_file(PLASMA_KWIN_DROPIN_PATH, _plasma_kwin_dropin_text(launcher)) + _require_removable_file(KWIN_CONFIG_PATH, managed_kwinrc) + lock_screen_ui = linux._read_text(PLASMA_LOCK_SCREEN_UI_PATH) + unmanaged_lock_screen_ui = ( + _plasma_lock_screen_ui_without_patch(lock_screen_ui) + if lock_screen_ui is not None + else None + ) + linux._remove_owned_file(PLASMA_INPUT_METHOD_PATH, _plasma_input_method_text(launcher)) + linux._remove_owned_file(PLASMA_KWIN_DROPIN_PATH, _plasma_kwin_dropin_text(launcher)) + try: + PLASMA_KWIN_DROPIN_PATH.parent.rmdir() + except OSError: + pass + if bool(state.get("kwinrc_existed")): + linux._write_atomic(KWIN_CONFIG_PATH, original_kwinrc, _state_mode(state, "kwinrc_mode")) + else: + linux._remove_owned_file(KWIN_CONFIG_PATH, managed_kwinrc) + if lock_screen_ui is not None and unmanaged_lock_screen_ui != lock_screen_ui: + linux._write_atomic( + PLASMA_LOCK_SCREEN_UI_PATH, + unmanaged_lock_screen_ui, + PLASMA_LOCK_SCREEN_UI_PATH.stat().st_mode & 0o777, + ) def _remove_lightdm(launcher: Path, state: dict[str, Any]) -> None: @@ -539,6 +707,216 @@ def _plasma_service_text() -> str: ) +def _plasma_input_method_text(launcher: Path) -> str: + return ( + "[Desktop Entry]\n" + "Name=Axidev OSK\n" + f"Exec={launcher}\n" + "Type=Application\n" + "X-KDE-Wayland-VirtualKeyboard=true\n" + "NoDisplay=true\n" + "Icon=axidev-osk\n" + ) + + +def _plasma_kwin_dropin_text(launcher: Path) -> str: + unit_path = next((path for path in PLASMA_KWIN_UNIT_PATHS if path.is_file()), None) + if unit_path is None: + raise linux.LinuxSetupError("Plasma Login Manager KWin service is missing") + unit_text = linux._read_text(unit_path) + assert unit_text is not None + command = _systemd_service_command(unit_text, "ExecStart") + try: + arguments = shlex.split(command) + except ValueError as exc: + raise linux.LinuxSetupError("Plasma Login Manager KWin command is invalid") from exc + try: + input_method_index = arguments.index("--inputmethod") + except ValueError as exc: + raise linux.LinuxSetupError( + "Plasma Login Manager KWin command has no --inputmethod option" + ) from exc + if input_method_index + 1 >= len(arguments): + raise linux.LinuxSetupError("Plasma Login Manager KWin input method is missing") + arguments[input_method_index + 1] = str(launcher) + return ( + "[Service]\n" + "Environment=AXIDEV_OSK_GREETER=1\n" + "ExecStart=\n" + f"ExecStart={shlex.join(arguments)}\n" + ) + + +def _systemd_service_command(unit_text: str, key: str) -> str: + section = "" + matches: list[str] = [] + for raw_line in unit_text.splitlines(): + line = raw_line.strip() + if line.startswith("[") and line.endswith("]"): + section = line[1:-1] + continue + if section == "Service" and line.startswith(f"{key}="): + value = line.split("=", 1)[1].strip() + if value: + matches.append(value) + if len(matches) != 1: + raise linux.LinuxSetupError( + f"Plasma Login Manager KWin service requires exactly one {key} command" + ) + return matches[0] + + +def _plasma_kwin_config_text(original: str | None) -> str: + lines = [] if original is None else original.splitlines(keepends=True) + if lines and not lines[-1].endswith(("\n", "\r")): + lines[-1] += "\n" + + section_start = next( + (index for index, line in enumerate(lines) if line.strip() == "[Wayland]"), + None, + ) + if section_start is None: + if lines and lines[-1].strip(): + lines.append("\n") + lines.extend( + ( + "[Wayland]\n", + f"InputMethod={PLASMA_INPUT_METHOD_PATH}\n", + "VirtualKeyboardMode=2\n", + ) + ) + return "".join(lines) + + section_end = next( + ( + index + for index in range(section_start + 1, len(lines)) + if lines[index].lstrip().startswith("[") + ), + len(lines), + ) + managed = { + "InputMethod": f"InputMethod={PLASMA_INPUT_METHOD_PATH}\n", + "VirtualKeyboardMode": "VirtualKeyboardMode=2\n", + } + found: set[str] = set() + rewritten: list[str] = [] + for line in lines[section_start + 1 : section_end]: + key = line.split("=", 1)[0].strip() if "=" in line else "" + if key not in managed: + rewritten.append(line) + elif key not in found: + rewritten.append(managed[key]) + found.add(key) + for key, line in managed.items(): + if key not in found: + rewritten.append(line) + lines[section_start + 1 : section_end] = rewritten + return "".join(lines) + + +def _plasma_lock_screen_patch_is_current(text: str | None) -> bool: + """Return whether QML contains exactly one unmodified managed block.""" + + return bool( + text is not None + and text.count(PLASMA_LOCK_SCREEN_PATCH) == 1 + and text.count(PLASMA_LOCK_SCREEN_PATCH_START) == 1 + and text.count(PLASMA_LOCK_SCREEN_PATCH_END) == 1 + ) + + +def _plasma_version() -> tuple[int, int, int] | None: + """Return the version of the package that owns Plasma's lock-screen QML.""" + + path = str(PLASMA_LOCK_SCREEN_UI_PATH) + version_text: str | None = None + rpm = shutil.which("rpm") + dpkg_query = shutil.which("dpkg-query") + pacman = shutil.which("pacman") + if rpm is not None: + version_text = _command_output([rpm, "-qf", "--queryformat", "%{VERSION}", path]) + elif dpkg_query is not None: + owner = _command_output([dpkg_query, "-S", path]) + if owner is not None and ": " in owner: + package = owner.split(": ", 1)[0] + version_text = _command_output([dpkg_query, "-W", "-f=${Version}", package]) + elif pacman is not None: + package = _command_output([pacman, "-Qqo", path]) + if package is not None: + installed = _command_output([pacman, "-Q", package]) + if installed is not None and " " in installed: + version_text = installed.split(" ", 1)[1] + if version_text is None: + return None + version_text = version_text.strip().split(":", 1)[-1] + match = re.search(r"\b(\d+)\.(\d+)(?:\.(\d+))?\b", version_text) + if match is None: + return None + major, minor, patch = match.groups() + return int(major), int(minor), int(patch or 0) + + +def _command_output(arguments: list[str]) -> str | None: + """Run a metadata command and return non-empty stdout on success.""" + + try: + completed = subprocess.run( + arguments, + check=False, + capture_output=True, + text=True, + ) + except OSError: + return None + output = completed.stdout.strip() + return output if completed.returncode == 0 and output else None + + +def _plasma_lock_screen_version_supported() -> bool: + version = _plasma_version() + return bool( + version is not None + and PLASMA_LOCK_SCREEN_MIN_VERSION <= version < PLASMA_LOCK_SCREEN_MAX_VERSION + ) + + +def _require_supported_plasma_lock_screen_version() -> None: + version = _plasma_version() + if version is None: + raise linux.LinuxSetupError("cannot determine the installed Plasma version") + if not PLASMA_LOCK_SCREEN_MIN_VERSION <= version < PLASMA_LOCK_SCREEN_MAX_VERSION: + rendered = ".".join(str(part) for part in version) + raise linux.LinuxSetupError( + f"Plasma lock-screen integration supports versions >=6.7.0 and <7.0.0; found {rendered}" + ) + + +def _plasma_lock_screen_ui_text(original: str) -> str: + """Add the managed always-visible unlock UI block to Plasma QML.""" + + if _plasma_lock_screen_patch_is_current(original): + return original + if PLASMA_LOCK_SCREEN_PATCH_START in original or PLASMA_LOCK_SCREEN_PATCH_END in original: + raise linux.LinuxSetupError("refusing to replace a changed Axidev lock-screen QML block") + anchor = " MouseArea {\n id: lockScreenRoot\n" + if original.count(anchor) != 1: + raise linux.LinuxSetupError( + "Plasma lock-screen QML does not contain the supported lockScreenRoot structure" + ) + return original.replace(anchor, anchor + "\n" + PLASMA_LOCK_SCREEN_PATCH, 1) + + +def _plasma_lock_screen_ui_without_patch(managed: str) -> str: + """Remove only the exact managed block from Plasma QML.""" + + if _plasma_lock_screen_patch_is_current(managed): + return managed.replace("\n" + PLASMA_LOCK_SCREEN_PATCH, "", 1) + if PLASMA_LOCK_SCREEN_PATCH_START in managed or PLASMA_LOCK_SCREEN_PATCH_END in managed: + raise linux.LinuxSetupError("refusing to remove a changed Axidev lock-screen QML block") + return managed + + def _lightdm_config_text() -> str: return f"[Seat:*]\ngreeter-wrapper={LIGHTDM_WRAPPER_PATH}\n" @@ -716,17 +1094,22 @@ def _require_compatible_file(path: Path, expected: str) -> None: raise linux.LinuxSetupError(f"refusing to replace conflicting file: {path}") +def _require_writable_regular_file(path: Path) -> None: + if path.is_symlink() or not path.is_file(): + raise linux.LinuxSetupError(f"refusing to replace a non-regular file: {path}") + + def _require_compatible_symlink(path: Path, target: Path) -> None: if not path.exists() and not path.is_symlink(): return - if not path.is_symlink() or path.resolve() != target: + if not path.is_symlink() or path.resolve() != target.resolve(): raise linux.LinuxSetupError(f"refusing to replace conflicting link: {path}") def _remove_owned_symlink(path: Path, target: Path) -> None: if not path.exists() and not path.is_symlink(): return - if not path.is_symlink() or path.resolve() != target: + if not path.is_symlink() or path.resolve() != target.resolve(): raise linux.LinuxSetupError(f"refusing to remove conflicting link: {path}") path.unlink() @@ -740,7 +1123,7 @@ def _require_removable_file(path: Path, expected: str) -> None: def _require_removable_symlink(path: Path, target: Path) -> None: if not path.exists() and not path.is_symlink(): return - if not path.is_symlink() or path.resolve() != target: + if not path.is_symlink() or path.resolve() != target.resolve(): raise linux.LinuxSetupError(f"refusing to remove conflicting link: {path}") @@ -774,6 +1157,24 @@ def _state_string(state: dict[str, Any], key: str) -> str: return value +def _state_text(state: dict[str, Any], key: str) -> str: + value = state.get(key) + if not isinstance(value, str): + raise linux.LinuxSetupError(f"managed greeter state is missing {key}") + return value + + +def _state_mode(state: dict[str, Any], key: str) -> int: + value = state.get(key) + if not isinstance(value, int) or not 0 <= value <= 0o777: + raise linux.LinuxSetupError(f"managed greeter state is missing {key}") + return value + + +def _is_legacy_plasma_state(state: dict[str, Any]) -> bool: + return state.get("manager") == "plasma-login" and "original_kwinrc" not in state + + def _runtime_launcher() -> Path: launcher = shutil.which("axidev-osk") return Path(launcher).resolve() if launcher else DEFAULT_LAUNCHER_PATH diff --git a/src/axidev_osk/platform/kwin_input_panel.py b/src/axidev_osk/platform/kwin_input_panel.py new file mode 100644 index 0000000..fac3d1c --- /dev/null +++ b/src/axidev_osk/platform/kwin_input_panel.py @@ -0,0 +1,449 @@ +"""KWin input-panel surface integration through the Wayland client ABI.""" + +from __future__ import annotations + +import ctypes +import ctypes.util +from collections.abc import Iterable +from dataclasses import dataclass, field + + +class KWinInputPanelError(RuntimeError): + """Raised when KWin cannot assign an input-panel role to a window.""" + + +class _WlInterface(ctypes.Structure): + pass + + +class _WlMessage(ctypes.Structure): + _fields_ = [ + ("name", ctypes.c_char_p), + ("signature", ctypes.c_char_p), + ("types", ctypes.POINTER(ctypes.POINTER(_WlInterface))), + ] + + +_WlInterface._fields_ = [ + ("name", ctypes.c_char_p), + ("version", ctypes.c_int), + ("method_count", ctypes.c_int), + ("methods", ctypes.POINTER(_WlMessage)), + ("event_count", ctypes.c_int), + ("events", ctypes.POINTER(_WlMessage)), +] + + +_RegistryGlobal = ctypes.CFUNCTYPE( + None, + ctypes.c_void_p, + ctypes.c_void_p, + ctypes.c_uint32, + ctypes.c_char_p, + ctypes.c_uint32, +) +_RegistryGlobalRemove = ctypes.CFUNCTYPE( + None, + ctypes.c_void_p, + ctypes.c_void_p, + ctypes.c_uint32, +) + + +class _WlRegistryListener(ctypes.Structure): + _fields_ = [("global_", _RegistryGlobal), ("global_remove", _RegistryGlobalRemove)] + + +_OutputGeometry = ctypes.CFUNCTYPE( + None, + ctypes.c_void_p, + ctypes.c_void_p, + ctypes.c_int32, + ctypes.c_int32, + ctypes.c_int32, + ctypes.c_int32, + ctypes.c_int32, + ctypes.c_char_p, + ctypes.c_char_p, + ctypes.c_int32, +) +_OutputMode = ctypes.CFUNCTYPE( + None, + ctypes.c_void_p, + ctypes.c_void_p, + ctypes.c_uint32, + ctypes.c_int32, + ctypes.c_int32, + ctypes.c_int32, +) +_OutputDone = ctypes.CFUNCTYPE(None, ctypes.c_void_p, ctypes.c_void_p) +_OutputScale = ctypes.CFUNCTYPE(None, ctypes.c_void_p, ctypes.c_void_p, ctypes.c_int32) +_OutputName = ctypes.CFUNCTYPE(None, ctypes.c_void_p, ctypes.c_void_p, ctypes.c_char_p) +_OutputDescription = ctypes.CFUNCTYPE(None, ctypes.c_void_p, ctypes.c_void_p, ctypes.c_char_p) + + +class _WlOutputListener(ctypes.Structure): + _fields_ = [ + ("geometry", _OutputGeometry), + ("mode", _OutputMode), + ("done", _OutputDone), + ("scale", _OutputScale), + ("name", _OutputName), + ("description", _OutputDescription), + ] + + +@dataclass +class _Output: + global_name: int + proxy: int + display_name: str | None = None + callbacks: list[object] = field(default_factory=list) + listener: _WlOutputListener | None = None + + +class _Protocol: + def __init__(self, library: ctypes.CDLL) -> None: + self.library = library + self.registry = _WlInterface.in_dll(library, "wl_registry_interface") + self.surface = _WlInterface.in_dll(library, "wl_surface_interface") + self.output = _WlInterface.in_dll(library, "wl_output_interface") + + input_panel_surface_types = (ctypes.POINTER(_WlInterface) * 2)( + ctypes.pointer(self.output), + ctypes.POINTER(_WlInterface)(), + ) + self.input_panel_surface_methods = (_WlMessage * 2)( + _WlMessage(b"set_toplevel", b"ou", input_panel_surface_types), + _WlMessage(b"set_overlay_panel", b"", None), + ) + self.input_panel_surface = _WlInterface( + b"zwp_input_panel_surface_v1", + 1, + len(self.input_panel_surface_methods), + self.input_panel_surface_methods, + 0, + None, + ) + + input_panel_types = (ctypes.POINTER(_WlInterface) * 2)( + ctypes.pointer(self.input_panel_surface), + ctypes.pointer(self.surface), + ) + self.input_panel_methods = (_WlMessage * 1)( + _WlMessage(b"get_input_panel_surface", b"no", input_panel_types), + ) + self.input_panel = _WlInterface( + b"zwp_input_panel_v1", + 1, + len(self.input_panel_methods), + self.input_panel_methods, + 0, + None, + ) + + +class KWinInputPanelAttachment: + """Own the client-side proxy for one input-panel surface role.""" + + def __init__(self, client: "_InputPanelClient", panel_surface: int) -> None: + self._client = client + self._panel_surface = panel_surface + + def close(self) -> None: + """Release the client-side role proxy exactly once.""" + + if self._panel_surface: + self._client.destroy_proxy(self._panel_surface) + self._panel_surface = 0 + + +class _InputPanelClient: + def __init__(self, library: ctypes.CDLL, display: int) -> None: + self.library = library + self.display = display + self.protocol = _Protocol(library) + self.marshal = library.wl_proxy_marshal_flags + self.marshal.restype = ctypes.c_void_p + library.wl_proxy_add_listener.argtypes = [ + ctypes.c_void_p, + ctypes.POINTER(ctypes.c_void_p), + ctypes.c_void_p, + ] + library.wl_proxy_add_listener.restype = ctypes.c_int + library.wl_display_roundtrip.argtypes = [ctypes.c_void_p] + library.wl_display_roundtrip.restype = ctypes.c_int + library.wl_display_flush.argtypes = [ctypes.c_void_p] + library.wl_display_flush.restype = ctypes.c_int + library.wl_proxy_destroy.argtypes = [ctypes.c_void_p] + library.wl_proxy_destroy.restype = None + + self.registry = self.marshal( + ctypes.c_void_p(display), + ctypes.c_uint32(1), + ctypes.pointer(self.protocol.registry), + ctypes.c_uint32(1), + ctypes.c_uint32(0), + None, + ) + if not self.registry: + raise KWinInputPanelError("cannot read the Wayland registry") + + self.input_panel = 0 + self._input_panel_global: tuple[int, int] | None = None + self._advertised_outputs: dict[int, int] = {} + self.outputs: dict[int, _Output] = {} + self._initialized = False + self._registry_callbacks = ( + _RegistryGlobal(self._registry_global), + _RegistryGlobalRemove(self._registry_global_remove), + ) + self._registry_listener = _WlRegistryListener(*self._registry_callbacks) + listener_pointer = ctypes.cast( + ctypes.pointer(self._registry_listener), + ctypes.POINTER(ctypes.c_void_p), + ) + if library.wl_proxy_add_listener(self.registry, listener_pointer, None) != 0: + self.destroy_proxy(self.registry) + self.registry = 0 + raise KWinInputPanelError("cannot listen to the Wayland registry") + try: + self._roundtrip() + if self._input_panel_global is None: + raise KWinInputPanelError("KWin did not expose zwp_input_panel_v1") + if not self._advertised_outputs: + raise KWinInputPanelError("KWin did not expose a Wayland output") + for name, version in self._advertised_outputs.items(): + self._bind_output(name, version) + self._roundtrip() + if not any(output.display_name for output in self.outputs.values()): + raise KWinInputPanelError( + "KWin outputs do not expose names through wl_output version 4" + ) + if self._input_panel_global is None: + raise KWinInputPanelError("KWin removed zwp_input_panel_v1 during setup") + self.input_panel = _bind_global( + self.marshal, + self.registry, + self.protocol.input_panel, + self._input_panel_global, + ) + self._initialized = True + except Exception: + self._cleanup_initialization() + raise + + def attach(self, surface: int, output_name: str) -> KWinInputPanelAttachment: + """Assign an input-panel role on the Qt-selected output.""" + + self._roundtrip() + try: + output = _select_output(self.outputs.values(), output_name) + except KWinInputPanelError: + self._roundtrip() + output = _select_output(self.outputs.values(), output_name) + panel_surface = self.marshal( + ctypes.c_void_p(self.input_panel), + ctypes.c_uint32(0), + ctypes.pointer(self.protocol.input_panel_surface), + ctypes.c_uint32(1), + ctypes.c_uint32(0), + None, + ctypes.c_void_p(surface), + ) + if not panel_surface: + raise KWinInputPanelError("KWin refused the input-panel surface") + self.marshal( + ctypes.c_void_p(panel_surface), + ctypes.c_uint32(0), + None, + ctypes.c_uint32(1), + ctypes.c_uint32(0), + ctypes.c_void_p(output.proxy), + ctypes.c_uint32(0), + ) + if self.library.wl_display_flush(self.display) < 0: + self.destroy_proxy(panel_surface) + raise KWinInputPanelError("KWin closed the input-method connection") + return KWinInputPanelAttachment(self, panel_surface) + + def destroy_proxy(self, proxy: int) -> None: + """Destroy one local Wayland proxy without touching Qt's display or surface.""" + + if proxy: + self.library.wl_proxy_destroy(ctypes.c_void_p(proxy)) + + def _roundtrip(self) -> None: + if self.library.wl_display_roundtrip(self.display) < 0: + raise KWinInputPanelError("KWin closed the input-method connection") + + def _registry_global( + self, + data: ctypes.c_void_p, + registry_proxy: ctypes.c_void_p, + name: int, + interface: bytes, + version: int, + ) -> None: + del data, registry_proxy + try: + if interface == b"zwp_input_panel_v1" and self._input_panel_global is None: + self._input_panel_global = (name, version) + elif interface == b"wl_output": + self._advertised_outputs[name] = version + if self._initialized: + self._bind_output(name, version) + except KWinInputPanelError: + pass + + def _registry_global_remove( + self, + data: ctypes.c_void_p, + registry_proxy: ctypes.c_void_p, + name: int, + ) -> None: + del data, registry_proxy + if self._input_panel_global is not None and self._input_panel_global[0] == name: + self._input_panel_global = None + self._advertised_outputs.pop(name, None) + output = self.outputs.pop(name, None) + if output is not None: + self.destroy_proxy(output.proxy) + + def _bind_output(self, name: int, version: int) -> None: + proxy = _bind_global( + self.marshal, + self.registry, + self.protocol.output, + (name, version), + maximum_version=4, + ) + output = _Output(name, proxy) + + @_OutputGeometry + def geometry(*args: object) -> None: + del args + + @_OutputMode + def mode(*args: object) -> None: + del args + + @_OutputDone + def done(*args: object) -> None: + del args + + @_OutputScale + def scale(*args: object) -> None: + del args + + @_OutputName + def output_name(data: ctypes.c_void_p, output_proxy: ctypes.c_void_p, value: bytes) -> None: + del data, output_proxy + output.display_name = value.decode("utf-8", errors="replace") + + @_OutputDescription + def description(*args: object) -> None: + del args + + output.callbacks.extend((geometry, mode, done, scale, output_name, description)) + output.listener = _WlOutputListener(*output.callbacks) + listener_pointer = ctypes.cast( + ctypes.pointer(output.listener), + ctypes.POINTER(ctypes.c_void_p), + ) + if self.library.wl_proxy_add_listener(proxy, listener_pointer, None) != 0: + self.destroy_proxy(proxy) + raise KWinInputPanelError("cannot listen to a Wayland output") + self.outputs[name] = output + + def _cleanup_initialization(self) -> None: + for output in self.outputs.values(): + self.destroy_proxy(output.proxy) + self.outputs.clear() + if self.input_panel: + self.destroy_proxy(self.input_panel) + self.input_panel = 0 + if self.registry: + self.destroy_proxy(self.registry) + self.registry = 0 + + +_clients: dict[int, _InputPanelClient] = {} + + +def attach_kwin_input_panel( + window_id: int, + *, + output_name: str, +) -> KWinInputPanelAttachment: + """Assign KWin's keyboard role to an unmapped Qt surface on its selected output.""" + + if not window_id: + raise KWinInputPanelError("Qt did not create a Wayland surface") + if not output_name: + raise KWinInputPanelError("Qt did not select a Wayland output") + library_name = ctypes.util.find_library("wayland-client") + if not library_name: + raise KWinInputPanelError("libwayland-client is unavailable") + + library = ctypes.CDLL(library_name) + library.wl_proxy_get_display.argtypes = [ctypes.c_void_p] + library.wl_proxy_get_display.restype = ctypes.c_void_p + surface = ctypes.c_void_p(window_id) + display = library.wl_proxy_get_display(surface) + if not display: + raise KWinInputPanelError("Qt window ID is not a Wayland surface") + + display_id = int(display) + client = _client_for_display(library, display_id) + return client.attach(window_id, output_name) + + +def _client_for_display(library: ctypes.CDLL, display: int) -> _InputPanelClient: + client = _clients.get(display) + if client is None: + client = _InputPanelClient(library, display) + _clients[display] = client + return client + + +def _select_output(outputs: Iterable[_Output], output_name: str) -> _Output: + candidates = tuple(outputs) + for output in candidates: + if output.display_name == output_name: + return output + available = ", ".join( + sorted(output.display_name for output in candidates if output.display_name) + ) + detail = available or "none named" + raise KWinInputPanelError( + f"KWin did not expose Qt output {output_name!r}; available outputs: {detail}" + ) + + +def _bind_global( + marshal: object, + registry: int, + interface: _WlInterface, + advertised: tuple[int, int], + *, + maximum_version: int | None = None, +) -> int: + name, advertised_version = advertised + version = min(interface.version, advertised_version) + if maximum_version is not None: + version = min(version, maximum_version) + proxy = marshal( + ctypes.c_void_p(registry), + ctypes.c_uint32(0), + ctypes.pointer(interface), + ctypes.c_uint32(version), + ctypes.c_uint32(0), + ctypes.c_uint32(name), + interface.name, + ctypes.c_uint32(version), + None, + ) + if not proxy: + raise KWinInputPanelError(f"cannot bind {interface.name.decode()}") + return proxy diff --git a/src/axidev_osk/platform/overlay.py b/src/axidev_osk/platform/overlay.py index c596527..24f16e4 100644 --- a/src/axidev_osk/platform/overlay.py +++ b/src/axidev_osk/platform/overlay.py @@ -17,6 +17,7 @@ class OverlayBackend(str, Enum): NATIVE = "native" WINDOWS_NATIVE = "windows-native" + WAYLAND_INPUT_PANEL = "wayland-input-panel" WAYLAND_LAYER_SHELL = "wayland-layer-shell" X11_UTILITY = "x11-utility" X11_UTILITY_BRIDGE = "x11-utility-bridge" diff --git a/src/axidev_osk/runtime/application.py b/src/axidev_osk/runtime/application.py index 4e95658..439543b 100644 --- a/src/axidev_osk/runtime/application.py +++ b/src/axidev_osk/runtime/application.py @@ -15,6 +15,7 @@ from ..config.models import AppConfig, ChromeConfig, PromptConfig, SurfaceConfig, WindowConfig from ..services import register_services from ..services.keyboard import KeyboardService +from ..services.kwin_lock import KWinLockService from ..styles.theme import apply_theme from ..windows.surface import register_surfaces from .context import Context @@ -25,7 +26,7 @@ route_component_pressed, route_hot_corner_triggered, ) -from .events import WindowCloseRequested +from .events import ScreenLockStateChanged, WindowCloseRequested from .prompt import PromptResolutionWaiter from .registries import ComponentRegistry, EventHandlerRegistry, ServiceRegistry, SurfaceRegistry from .state_store import StateStore @@ -44,6 +45,8 @@ def __init__( config: AppConfig | None = None, services: ServiceRegistry | None = None, event_handlers: EventHandlerRegistry | None = None, + confirm_quit: bool = True, + show_startup_windows: bool = True, ) -> None: """Create the main runtime. @@ -52,6 +55,8 @@ def __init__( config: Optional declarative app config. services: Optional pre-populated service registry for tests. event_handlers: Optional pre-populated handler registry for tests. + confirm_quit: Whether shutdown requests require confirmation. + show_startup_windows: Whether to create configured startup windows immediately. Returns: None. @@ -61,6 +66,8 @@ def __init__( """ self._app = app + self._show_startup_windows = show_startup_windows + self._screen_locked: bool | None = None self._config = config or build_default_app_config() self._dispatcher = Dispatcher() self._services = services or ServiceRegistry() @@ -91,7 +98,7 @@ def __init__( self._event_handlers.install(self._dispatcher, self) self._quit_controller = ApplicationQuitController( app, - prompt=self._show_quit_prompt, + prompt=self._show_quit_prompt if confirm_quit else lambda _parent: True, parent=app, ) self._linux_permissions = LinuxPermissionController( @@ -116,11 +123,12 @@ def start(self) -> int: """ apply_theme(self._app) - for service in self._services.services(): + for service in self._services.autostart_services(): service.start(self.context) - for window_id in self._config.startup_window_ids: - window = self._window_manager.show(window_id) - self._quit_controller.register_window(window) + if self._show_startup_windows: + for window_id in self._config.startup_window_ids: + window = self._window_manager.show(window_id) + self._quit_controller.register_window(window) for service in self._services.services(): self._quit_controller.register_quit_callback(service.stop) self._quit_controller.install_signal_handlers() @@ -144,6 +152,39 @@ def _handle_window_close_requested(self, event: object) -> None: if isinstance(event, WindowCloseRequested): self._quit_controller.request_quit() + def _handle_screen_lock_state_changed(self, event: object) -> None: + """Create or destroy secure runtime resources as KDE locks and unlocks.""" + + if not isinstance(event, ScreenLockStateChanged): + return + if event.locked == self._screen_locked: + if event.locked: + self._services.get("kwin_lock", KWinLockService).activate() + return + window_id = self._config.keyboard_window_id + if event.locked: + try: + self._keyboard.start(self.context) + window = self._window_manager.show(window_id) + window.set_close_enabled(False) + self._services.get("kwin_lock", KWinLockService).activate() + except Exception: + try: + self._window_manager.destroy(window_id) + except Exception: + _logger.exception("Failed to destroy a partially started lock window") + try: + self._keyboard.shutdown() + except Exception: + _logger.exception("Failed to shut down keyboard output after lock startup failed") + raise + else: + try: + self._window_manager.destroy(window_id) + finally: + self._keyboard.shutdown() + self._screen_locked = event.locked + def _handle_hot_corner_triggered(self, event: object) -> None: """Map hot-corner events to managed window visibility commands.""" diff --git a/src/axidev_osk/runtime/event_handlers.py b/src/axidev_osk/runtime/event_handlers.py index d62a9c5..3d6cbb7 100644 --- a/src/axidev_osk/runtime/event_handlers.py +++ b/src/axidev_osk/runtime/event_handlers.py @@ -94,6 +94,7 @@ def register_event_handlers(registry: EventHandlerRegistry) -> None: lambda runtime: lambda command: runtime._app.exit(command.exit_code), ) registry.register_event_handler(lambda runtime: runtime._handle_window_close_requested) + registry.register_event_handler(lambda runtime: runtime._handle_screen_lock_state_changed) registry.register_event_handler(lambda runtime: runtime._handle_hot_corner_triggered) registry.register_event_handler(lambda runtime: runtime._handle_component_pressed) diff --git a/src/axidev_osk/runtime/events.py b/src/axidev_osk/runtime/events.py index 42f3ebe..cc770e4 100644 --- a/src/axidev_osk/runtime/events.py +++ b/src/axidev_osk/runtime/events.py @@ -104,6 +104,13 @@ class HotCornerTriggered: corner: str +@dataclass(frozen=True, slots=True) +class ScreenLockStateChanged: + """The desktop session entered or left its locked state.""" + + locked: bool + + @dataclass(frozen=True, slots=True) class WindowCloseRequested: """A managed window requested application shutdown confirmation. @@ -136,6 +143,7 @@ class PromptResolved: | BackendKeyStateChanged | KeyLatchChanged | HotCornerTriggered + | ScreenLockStateChanged | WindowCloseRequested | PromptResolved ) diff --git a/src/axidev_osk/runtime/registries.py b/src/axidev_osk/runtime/registries.py index b748661..e852aaa 100644 --- a/src/axidev_osk/runtime/registries.py +++ b/src/axidev_osk/runtime/registries.py @@ -178,11 +178,16 @@ def __init__(self) -> None: """Create an empty service registry.""" self._services: dict[str, RuntimeService] = {} + self._deferred: set[str] = set() - def register(self, name: str, service: RuntimeService) -> None: + def register(self, name: str, service: RuntimeService, *, autostart: bool = True) -> None: """Register a runtime service under a stable name.""" self._services[name] = service + if autostart: + self._deferred.discard(name) + else: + self._deferred.add(name) def get(self, name: str, service_type: type[RuntimeT]) -> RuntimeT: """Return a named service, validating its concrete type.""" @@ -199,6 +204,11 @@ def services(self) -> Iterable[RuntimeService]: return tuple(self._services.values()) + def autostart_services(self) -> Iterable[RuntimeService]: + """Yield services that should start with the application runtime.""" + + return tuple(service for name, service in self._services.items() if name not in self._deferred) + class EventHandlerRegistry: """Stores default command and event handler factories for installation.""" diff --git a/src/axidev_osk/runtime/testing.py b/src/axidev_osk/runtime/testing.py index 79a1d33..8ba8570 100644 --- a/src/axidev_osk/runtime/testing.py +++ b/src/axidev_osk/runtime/testing.py @@ -65,6 +65,11 @@ def _handle_window_close_requested(self, event: object) -> None: if isinstance(event, WindowCloseRequested): self._dispatcher.dispatch_command(AppQuit()) + def _handle_screen_lock_state_changed(self, event: object) -> None: + """Ignore platform lock-state events in the generic test runtime.""" + + del event + def _handle_hot_corner_triggered(self, event: object) -> None: """Route hot-corner visibility commands through production helper.""" diff --git a/src/axidev_osk/runtime/window_manager.py b/src/axidev_osk/runtime/window_manager.py index d157144..ebc24c5 100644 --- a/src/axidev_osk/runtime/window_manager.py +++ b/src/axidev_osk/runtime/window_manager.py @@ -174,7 +174,7 @@ def toggle_opacity(self, window_id: str, *, component_id: str, opacity: float) - blocker = _WindowInputBlocker(window, component_id) app.installEventFilter(blocker) self._input_blockers[window_id] = blocker - window.setWindowOpacity(opacity) + window.set_visual_opacity(opacity) def _restore_interaction(self, window_id: str, window: QWidget) -> None: """Restore configured opacity and remove any temporary input blocker.""" @@ -183,7 +183,7 @@ def _restore_interaction(self, window_id: str, window: QWidget) -> None: app = QApplication.instance() if blocker is not None and app is not None: app.removeEventFilter(blocker) - window.setWindowOpacity(self._configs[window_id].opacity) + window.set_visual_opacity(self._configs[window_id].opacity) def close(self, window_id: str) -> None: """Close and forget a managed window if it exists.""" @@ -194,6 +194,17 @@ def close(self, window_id: str) -> None: self._restore_interaction(window_id, window) window.close() + def destroy(self, window_id: str) -> None: + """Hide and delete a managed window without treating it as an app quit request.""" + + window = self._windows.pop(window_id, None) + if window is not None: + _logger.info("Destroying runtime window %s", window_id) + self._restore_interaction(window_id, window) + window.release_platform_resources() + window.hide() + window.deleteLater() + def all_windows(self) -> list[RuntimeWindow]: """Return all live managed windows.""" diff --git a/src/axidev_osk/services/keyboard/service.py b/src/axidev_osk/services/keyboard/service.py index 7d68dbc..a76c091 100644 --- a/src/axidev_osk/services/keyboard/service.py +++ b/src/axidev_osk/services/keyboard/service.py @@ -100,6 +100,7 @@ def initialize(self) -> bool: """ initialized = self._backend.initialize() + self._shutdown = False self._ensure_backend_listener() return initialized @@ -122,7 +123,7 @@ def shutdown(self) -> None: self._shutdown = True started_at = time.perf_counter() _logger.info("Shutting down keyboard backend") - self._release_press_handles() + self.reset_state() self._backend.shutdown() _logger.info("Keyboard backend shutdown completed in %.3fs", time.perf_counter() - started_at) diff --git a/src/axidev_osk/services/kwin_lock.py b/src/axidev_osk/services/kwin_lock.py new file mode 100644 index 0000000..1a19b83 --- /dev/null +++ b/src/axidev_osk/services/kwin_lock.py @@ -0,0 +1,110 @@ +"""KWin screen-lock state integration for the secure input panel.""" + +from __future__ import annotations + +import logging +from typing import TYPE_CHECKING + +from PySide6.QtCore import QObject, SLOT, Slot +from PySide6.QtDBus import QDBusConnection, QDBusInterface, QDBusMessage + +from ..runtime.events import ScreenLockStateChanged + +if TYPE_CHECKING: + from ..runtime.context import Context + +_logger = logging.getLogger(__name__) + + +class KWinLockService(QObject): + """Observe KDE's lock state and expose KWin input-method activation.""" + + def __init__(self, parent: QObject | None = None) -> None: + super().__init__(parent) + self._context: Context | None = None + self._connection = QDBusConnection.sessionBus() + self._virtual_keyboard: QDBusInterface | None = None + + def start(self, context: Context) -> None: + """Connect lock-state signals and publish the current state.""" + + self._context = context + if not self._connection.isConnected(): + raise RuntimeError("KDE session bus is unavailable") + self._virtual_keyboard = QDBusInterface( + "org.kde.KWin", + "/VirtualKeyboard", + "org.kde.kwin.VirtualKeyboard", + self._connection, + ) + connected_about = self._connection.connect( + "org.kde.screensaver", + "/ScreenSaver", + "org.kde.screensaver", + "AboutToLock", + self, + SLOT("aboutToLock()"), + ) + connected_active = self._connection.connect( + "org.freedesktop.ScreenSaver", + "/ScreenSaver", + "org.freedesktop.ScreenSaver", + "ActiveChanged", + self, + SLOT("activeChanged(bool)"), + ) + if not connected_about or not connected_active: + raise RuntimeError("Cannot monitor KDE screen-lock state") + + screen_saver = QDBusInterface( + "org.freedesktop.ScreenSaver", + "/ScreenSaver", + "org.freedesktop.ScreenSaver", + self._connection, + ) + reply = screen_saver.call("GetActive") + if reply.type() == QDBusMessage.MessageType.ReplyMessage and reply.arguments(): + self._emit_state(bool(reply.arguments()[0])) + else: + _logger.warning("KDE screen-lock state is unavailable; keeping the secure panel hidden") + self._emit_state(False) + + def stop(self) -> None: + """Disconnect lock-state signals.""" + + self._connection.disconnect( + "org.kde.screensaver", + "/ScreenSaver", + "org.kde.screensaver", + "AboutToLock", + self, + SLOT("aboutToLock()"), + ) + self._connection.disconnect( + "org.freedesktop.ScreenSaver", + "/ScreenSaver", + "org.freedesktop.ScreenSaver", + "ActiveChanged", + self, + SLOT("activeChanged(bool)"), + ) + self._context = None + self._virtual_keyboard = None + + def activate(self) -> None: + """Ask KWin to activate its configured virtual keyboard.""" + + if self._virtual_keyboard is not None: + self._virtual_keyboard.call("forceActivate") + + @Slot() + def aboutToLock(self) -> None: + self._emit_state(True) + + @Slot(bool) + def activeChanged(self, active: bool) -> None: + self._emit_state(active) + + def _emit_state(self, locked: bool) -> None: + if self._context is not None: + self._context.dispatcher.dispatch_event(ScreenLockStateChanged(locked=locked)) diff --git a/src/axidev_osk/windows/builder.py b/src/axidev_osk/windows/builder.py index eefaf85..d6323d0 100644 --- a/src/axidev_osk/windows/builder.py +++ b/src/axidev_osk/windows/builder.py @@ -9,7 +9,8 @@ from ..config.models import WindowConfig from ..runtime.context import Context from ..runtime.events import WindowCloseRequested -from .chrome import install_overlay_chrome +from .chrome import OverlayChromeWidgets, install_overlay_chrome +from .opacity import WindowOpacityController from .overlay import configure_always_on_top_window, configure_plain_window @@ -42,28 +43,34 @@ def __init__(self, config: WindowConfig, context: Context, parent: QWidget | Non self._config = config self._context = context self._quit_controller_managed = False + self._chrome_widgets: OverlayChromeWidgets | None = None self.setProperty("componentType", "window") self.setProperty("componentId", config.id) self.setWindowTitle(config.title) - self.setWindowOpacity(config.opacity) if config.overlay.always_on_top: self._overlay = configure_always_on_top_window(self, config=config.overlay.config) else: self._overlay = configure_plain_window(self) - - central = context.surfaces.build(config.surface, context) - if config.chrome.enabled and getattr(self._overlay, "uses_custom_chrome", False): - central_layout = central.layout() - if isinstance(central_layout, QVBoxLayout): - install_overlay_chrome( - central_layout, - title=self.windowTitle(), - parent=central, - on_move=self._overlay.move_by, - on_resize=self._overlay.resize_by, - ) - self.setCentralWidget(central) - self.apply_startup_size(minimum_size=config.surface.minimum_size) + self.destroyed.connect(self._release_platform_resources_on_destroy) + try: + central = context.surfaces.build(config.surface, context) + if config.chrome.enabled and getattr(self._overlay, "uses_custom_chrome", False): + central_layout = central.layout() + if isinstance(central_layout, QVBoxLayout): + self._chrome_widgets = install_overlay_chrome( + central_layout, + title=self.windowTitle(), + parent=central, + on_move=self._overlay.move_by, + on_resize=self._overlay.resize_by, + ) + self.setCentralWidget(central) + self._opacity = WindowOpacityController(self) + self.set_visual_opacity(config.opacity) + self.apply_startup_size(minimum_size=config.surface.minimum_size) + except Exception: + self.release_platform_resources() + raise @property def window_id(self) -> str: @@ -71,6 +78,28 @@ def window_id(self) -> str: return self._config.id + def set_visual_opacity(self, opacity: float) -> None: + """Set opacity through the platform-supported window implementation.""" + + self._opacity.set_opacity(opacity) + + def set_close_enabled(self, enabled: bool) -> None: + """Set whether installed custom chrome exposes its close control.""" + + if self._chrome_widgets is not None: + self._chrome_widgets.title_bar.set_close_enabled(enabled) + + def release_platform_resources(self) -> None: + """Release native resources before Qt destroys this window.""" + + release = getattr(self._overlay, "release_resources", None) + if release is not None: + release() + + def _release_platform_resources_on_destroy(self, *args: object) -> None: + del args + self.release_platform_resources() + def set_quit_controller_managed(self, managed: bool) -> None: """Set whether close events should request managed app quit. diff --git a/src/axidev_osk/windows/chrome.py b/src/axidev_osk/windows/chrome.py index 35aeff7..fd2c103 100644 --- a/src/axidev_osk/windows/chrome.py +++ b/src/axidev_osk/windows/chrome.py @@ -69,6 +69,11 @@ def add_control(self, widget: QWidget) -> None: self._layout.insertWidget(self._layout.indexOf(self._close_button), widget) + def set_close_enabled(self, enabled: bool) -> None: + """Set whether the title bar exposes its close control.""" + + self._close_button.setVisible(enabled) + def mousePressEvent(self, event: QMouseEvent) -> None: """Begin a title-bar drag on left mouse press.""" diff --git a/src/axidev_osk/windows/opacity.py b/src/axidev_osk/windows/opacity.py new file mode 100644 index 0000000..1e0f8c7 --- /dev/null +++ b/src/axidev_osk/windows/opacity.py @@ -0,0 +1,26 @@ +"""Platform-aware visual opacity for runtime windows.""" + +from __future__ import annotations + +from PySide6.QtGui import QGuiApplication +from PySide6.QtWidgets import QGraphicsOpacityEffect, QMainWindow + + +class WindowOpacityController: + """Apply one opacity API through the primitive supported by the platform.""" + + def __init__(self, window: QMainWindow) -> None: + self._window = window + self._content_effect: QGraphicsOpacityEffect | None = None + + def set_opacity(self, opacity: float) -> None: + """Set visual opacity for the complete runtime-window content.""" + + content = self._window.centralWidget() + if QGuiApplication.platformName().lower() == "wayland" and content is not None: + if self._content_effect is None: + self._content_effect = QGraphicsOpacityEffect(content) + content.setGraphicsEffect(self._content_effect) + self._content_effect.setOpacity(opacity) + return + self._window.setWindowOpacity(opacity) diff --git a/src/axidev_osk/windows/overlay/always_on_top.py b/src/axidev_osk/windows/overlay/always_on_top.py index 7d6a942..2c57a9d 100644 --- a/src/axidev_osk/windows/overlay/always_on_top.py +++ b/src/axidev_osk/windows/overlay/always_on_top.py @@ -43,6 +43,7 @@ is_wayland_session, prepend_plugin_root, ) +from ...platform.kwin_input_panel import attach_kwin_input_panel TWindow = TypeVar("TWindow", bound=QWidget) @@ -103,6 +104,11 @@ def prepare_always_on_top_window_environment( if forced_platforms and "wayland" not in forced_platforms: return _set_overlay_backend(OverlayBackend.NATIVE) + if os.environ.get("WAYLAND_SOCKET"): + os.environ["QT_QPA_PLATFORM"] = "wayland" + os.environ["QT_WAYLAND_USE_BYPASSWINDOWMANAGERHINT"] = "1" + return _set_overlay_backend(OverlayBackend.WAYLAND_INPUT_PANEL) + if not is_wayland_session(): if os.environ.get("DISPLAY"): return _set_overlay_backend(OverlayBackend.X11_UTILITY) @@ -178,6 +184,7 @@ def __init__( self._floating_position_initialized = False self._show_adjustments_applied = False self._layer_shell_startup_refresh_applied = False + self._input_panel_attachment = None @property def backend(self) -> OverlayBackend: @@ -190,6 +197,7 @@ def uses_custom_chrome(self) -> bool: """Whether this backend requires app-provided frameless chrome.""" return self._backend in { + OverlayBackend.WAYLAND_INPUT_PANEL, OverlayBackend.WAYLAND_LAYER_SHELL, OverlayBackend.X11_UTILITY, OverlayBackend.X11_UTILITY_BRIDGE, @@ -205,6 +213,9 @@ def configure_window(self) -> None: self._window.setWindowFlag(Qt.WindowType.WindowStaysOnTopHint, True) self._window.setWindowFlag(Qt.WindowType.WindowDoesNotAcceptFocus, True) + if self._backend == OverlayBackend.WAYLAND_INPUT_PANEL: + self._window.setWindowFlag(Qt.WindowType.BypassWindowManagerHint, True) + if self.uses_custom_chrome: self._window.setWindowFlag(Qt.WindowType.FramelessWindowHint, True) @@ -215,6 +226,14 @@ def configure_window(self) -> None: if self._backend == OverlayBackend.WINDOWS_NATIVE: _set_windows_taskbar_style(int(self._window.winId())) + if self._backend == OverlayBackend.WAYLAND_INPUT_PANEL: + screen = self._window.screen() + output_name = screen.name() if screen is not None else "" + self._input_panel_attachment = attach_kwin_input_panel( + int(self._window.winId()), + output_name=output_name, + ) + self._debug_log( "configure-window", backend=self._backend.value, @@ -223,6 +242,13 @@ def configure_window(self) -> None: screen_margin=self._config.screen_margin, ) + def release_resources(self) -> None: + """Release native resources owned by this overlay controller.""" + + if self._input_panel_attachment is not None: + self._input_panel_attachment.close() + self._input_panel_attachment = None + def handle_show(self) -> bool: """Apply backend-specific adjustments after the window is shown.""" @@ -236,6 +262,9 @@ def handle_show(self) -> bool: self._refresh_wayland_layer_shell_surface_after_startup() return applied + if self._backend == OverlayBackend.WAYLAND_INPUT_PANEL: + return True + if self._backend in {OverlayBackend.X11_UTILITY, OverlayBackend.X11_UTILITY_BRIDGE, OverlayBackend.NATIVE}: self._position_floating_window_if_needed() return True @@ -256,6 +285,8 @@ def prepare_show(self) -> bool: if self._backend == OverlayBackend.WAYLAND_LAYER_SHELL: return self._sync_wayland_layer_shell() + if self._backend == OverlayBackend.WAYLAND_INPUT_PANEL: + return True if self._backend in {OverlayBackend.WINDOWS_NATIVE, OverlayBackend.X11_UTILITY, OverlayBackend.X11_UTILITY_BRIDGE, OverlayBackend.NATIVE}: self._position_floating_window_if_needed() return True @@ -371,10 +402,13 @@ def _detect_backend(self) -> OverlayBackend: platform = self._qt_platform() if platform == "wayland": selected = _read_selected_backend() - if selected == OverlayBackend.WAYLAND_LAYER_SHELL: + if selected in { + OverlayBackend.WAYLAND_INPUT_PANEL, + OverlayBackend.WAYLAND_LAYER_SHELL, + }: return selected raise RuntimeError( - "Wayland overlay backend was initialized without layer-shell support." + "Wayland overlay backend was initialized without input-panel or layer-shell support." ) if platform == "xcb": diff --git a/tests/test_app.py b/tests/test_app.py new file mode 100644 index 0000000..6eb6c29 --- /dev/null +++ b/tests/test_app.py @@ -0,0 +1,44 @@ +from __future__ import annotations + +import unittest +from unittest.mock import Mock + +from PySide6.QtCore import QObject + +from axidev_osk.app import _input_panel_services +from axidev_osk.platform.overlay import OverlayBackend + + +class InputPanelRuntimeTests(unittest.TestCase): + def test_input_panel_starts_only_keyboard_service(self) -> None: + app = QObject() + + services = _input_panel_services( + app, + OverlayBackend.WAYLAND_INPUT_PANEL, + lock_lifecycle=True, + ) + + self.assertIsNotNone(services) + self.assertEqual(len(tuple(services.services())), 2) + self.assertEqual(len(tuple(services.autostart_services())), 1) + + def test_plasma_login_input_panel_starts_keyboard_without_lock_monitor(self) -> None: + services = _input_panel_services( + QObject(), + OverlayBackend.WAYLAND_INPUT_PANEL, + lock_lifecycle=False, + ) + + self.assertIsNotNone(services) + self.assertEqual(len(tuple(services.services())), 1) + self.assertEqual(tuple(services.autostart_services()), tuple(services.services())) + + def test_ordinary_overlay_uses_default_services(self) -> None: + services = _input_panel_services( + Mock(), + OverlayBackend.WAYLAND_LAYER_SHELL, + lock_lifecycle=False, + ) + + self.assertIsNone(services) diff --git a/tests/test_application_runtime.py b/tests/test_application_runtime.py index 8efee3c..e81b2b0 100644 --- a/tests/test_application_runtime.py +++ b/tests/test_application_runtime.py @@ -2,7 +2,7 @@ import unittest from dataclasses import replace -from unittest.mock import patch +from unittest.mock import Mock, patch from PySide6.QtCore import QTimer from PySide6.QtWidgets import QApplication, QWidget @@ -11,7 +11,10 @@ from axidev_osk.config.defaults import build_default_app_config from axidev_osk.config.models import WindowConfig from axidev_osk.runtime.application import ApplicationRuntime -from axidev_osk.runtime.events import PromptResolved +from axidev_osk.runtime.events import PromptResolved, ScreenLockStateChanged +from axidev_osk.runtime.registries import ServiceRegistry +from axidev_osk.services.keyboard import KeyboardService +from axidev_osk.services.kwin_lock import KWinLockService def _app() -> QApplication: @@ -105,5 +108,104 @@ def create_transient(window_config: WindowConfig, *, parent: QWidget | None = No self.assertEqual(created[0].title, sentinel) +class SecureInputPanelLifecycleTests(unittest.TestCase): + def test_repeated_lock_cycles_rebuild_window_and_restart_keyboard(self) -> None: + backend = Mock() + backend.initialize.return_value = True + backend.add_key_state_listener.return_value = lambda: None + keyboard = KeyboardService(backend) + kwin_lock = KWinLockService() + kwin_lock.activate = Mock() + services = ServiceRegistry() + services.register("keyboard", keyboard, autostart=False) + services.register("kwin_lock", kwin_lock, autostart=False) + runtime = ApplicationRuntime(_app(), services=services, show_startup_windows=False) + lock_window = Mock() + + with ( + patch.object(runtime._window_manager, "show", return_value=lock_window) as show, + patch.object(runtime._window_manager, "destroy") as destroy, + ): + runtime.context.dispatcher.dispatch_event(ScreenLockStateChanged(locked=True)) + runtime.context.dispatcher.dispatch_event(ScreenLockStateChanged(locked=False)) + runtime.context.dispatcher.dispatch_event(ScreenLockStateChanged(locked=True)) + + self.assertEqual(backend.initialize.call_count, 2) + backend.shutdown.assert_called_once_with() + self.assertEqual(show.call_count, 2) + self.assertEqual( + lock_window.set_close_enabled.call_args_list, + [unittest.mock.call(False), unittest.mock.call(False)], + ) + destroy.assert_called_once_with(runtime._config.keyboard_window_id) + self.assertEqual(kwin_lock.activate.call_count, 2) + + def test_failed_lock_window_creation_rolls_back_and_remains_retryable(self) -> None: + backend = Mock() + backend.initialize.return_value = True + backend.add_key_state_listener.return_value = lambda: None + keyboard = KeyboardService(backend) + kwin_lock = KWinLockService() + kwin_lock.activate = Mock() + services = ServiceRegistry() + services.register("keyboard", keyboard, autostart=False) + services.register("kwin_lock", kwin_lock, autostart=False) + runtime = ApplicationRuntime(_app(), services=services, show_startup_windows=False) + + with ( + patch.object( + runtime._window_manager, + "show", + side_effect=(RuntimeError("window failed"), Mock()), + ) as show, + patch.object(runtime._window_manager, "destroy") as destroy, + ): + with self.assertRaisesRegex(RuntimeError, "window failed"): + runtime.context.dispatcher.dispatch_event(ScreenLockStateChanged(locked=True)) + self.assertIsNone(runtime._screen_locked) + + runtime.context.dispatcher.dispatch_event(ScreenLockStateChanged(locked=True)) + + self.assertEqual(show.call_count, 2) + destroy.assert_called_once_with(runtime._config.keyboard_window_id) + self.assertEqual(backend.initialize.call_count, 2) + backend.shutdown.assert_called_once_with() + kwin_lock.activate.assert_called_once_with() + self.assertTrue(runtime._screen_locked) + + def test_failed_lock_startup_preserves_error_when_cleanup_also_fails(self) -> None: + backend = Mock() + backend.initialize.return_value = True + backend.add_key_state_listener.return_value = lambda: None + keyboard = KeyboardService(backend) + kwin_lock = KWinLockService() + services = ServiceRegistry() + services.register("keyboard", keyboard, autostart=False) + services.register("kwin_lock", kwin_lock, autostart=False) + runtime = ApplicationRuntime(_app(), services=services, show_startup_windows=False) + + with ( + patch.object( + runtime._window_manager, + "show", + side_effect=RuntimeError("window failed"), + ), + patch.object( + runtime._window_manager, + "destroy", + side_effect=RuntimeError("cleanup failed"), + ), + patch("axidev_osk.runtime.application._logger") as logger, + self.assertRaisesRegex(RuntimeError, "window failed"), + ): + runtime.context.dispatcher.dispatch_event(ScreenLockStateChanged(locked=True)) + + backend.shutdown.assert_called_once_with() + logger.exception.assert_called_once_with( + "Failed to destroy a partially started lock window" + ) + self.assertIsNone(runtime._screen_locked) + + if __name__ == "__main__": unittest.main() diff --git a/tests/test_hot_corner_overlay.py b/tests/test_hot_corner_overlay.py index 7816198..7d96be6 100644 --- a/tests/test_hot_corner_overlay.py +++ b/tests/test_hot_corner_overlay.py @@ -4,7 +4,7 @@ import unittest from pathlib import Path from tempfile import TemporaryDirectory -from unittest.mock import patch +from unittest.mock import Mock, patch from PySide6.QtCore import QMargins, QPoint, QRect, Qt from PySide6.QtWidgets import QApplication @@ -41,6 +41,7 @@ def __init__(self) -> None: self._opacity = 1.0 self._window_flags = Qt.WindowType.Widget self._attributes_enabled: set[Qt.WidgetAttribute] = set() + self.lifecycle: list[str] = [] def setFocusPolicy(self, policy: Qt.FocusPolicy) -> None: self.focus_policies.append(policy) @@ -53,6 +54,7 @@ def setAttribute(self, attribute: Qt.WidgetAttribute, enabled: bool = True) -> N self._attributes_enabled.discard(attribute) def setWindowFlag(self, flag: Qt.WindowType, enabled: bool = True) -> None: + self.lifecycle.append(f"flag:{flag.name}") self.flags.append((flag, enabled)) if enabled: self._window_flags |= flag @@ -121,6 +123,7 @@ def testAttribute(self, attribute: Qt.WidgetAttribute) -> bool: return attribute in self._attributes_enabled def winId(self) -> int: + self.lifecycle.append("win-id") return 1 @@ -151,12 +154,16 @@ def handle_show(self) -> bool: class FakeScreen: - def __init__(self, geometry: QRect) -> None: + def __init__(self, geometry: QRect, name: str = "Virtual-1") -> None: self._geometry = QRect(geometry) + self._name = name def geometry(self) -> QRect: return QRect(self._geometry) + def name(self) -> str: + return self._name + class OverlayWindowControllerTests(unittest.TestCase): def test_configure_window_disables_system_background_erase(self) -> None: @@ -427,6 +434,52 @@ def test_find_qt_platform_plugin_root_detects_pyinstaller_bundle_plugins(self) - class OverlayBackendSelectionTests(unittest.TestCase): + def test_kwin_input_method_connection_selects_input_panel(self) -> None: + with patch( + "axidev_osk.windows.overlay.always_on_top.sys.platform", + "linux", + ), patch.dict( + "os.environ", + {"WAYLAND_SOCKET": "12"}, + clear=True, + ): + backend = prepare_always_on_top_window_environment() + selected_backend = os.environ["AXIDEV_OSK_OVERLAY_BACKEND"] + qt_platform = os.environ["QT_QPA_PLATFORM"] + bypass_hint = os.environ["QT_WAYLAND_USE_BYPASSWINDOWMANAGERHINT"] + + self.assertEqual(backend, OverlayBackend.WAYLAND_INPUT_PANEL) + self.assertEqual(selected_backend, "wayland-input-panel") + self.assertEqual(qt_platform, "wayland") + self.assertEqual(bypass_hint, "1") + + def test_input_panel_controller_assigns_role_before_show(self) -> None: + window = FakeWindow() + window.screen = lambda: FakeScreen(QRect(0, 0, 1920, 1080)) + attachment = Mock() + with patch.object( + AlwaysOnTopWindowController, + "_detect_backend", + return_value=OverlayBackend.WAYLAND_INPUT_PANEL, + ), patch( + "axidev_osk.windows.overlay.always_on_top.attach_kwin_input_panel", + side_effect=lambda *_args, **_kwargs: ( + window.lifecycle.append("attach") or attachment + ), + ) as attach_input_panel: + controller = AlwaysOnTopWindowController(window) + controller.configure_window() + + self.assertIn((Qt.WindowType.BypassWindowManagerHint, True), window.flags) + attach_input_panel.assert_called_once_with(1, output_name="Virtual-1") + self.assertLess( + window.lifecycle.index("flag:FramelessWindowHint"), + window.lifecycle.index("win-id"), + ) + self.assertLess(window.lifecycle.index("win-id"), window.lifecycle.index("attach")) + controller.release_resources() + attachment.close.assert_called_once_with() + def test_wayland_without_layer_shell_falls_back_to_x11_bridge_with_warning(self) -> None: with patch( "axidev_osk.windows.overlay.always_on_top.sys.platform", diff --git a/tests/test_keyboard_service.py b/tests/test_keyboard_service.py index c7f830f..8c46ade 100644 --- a/tests/test_keyboard_service.py +++ b/tests/test_keyboard_service.py @@ -328,6 +328,18 @@ def test_service_reset_state_clears_latches_for_registered_layout(self) -> None: self.assertIsNone(context.state.get(keyboard_latches_namespace(LAYOUT_ID), "shift")) self.assertFalse(context.keyboard.is_latched(LAYOUT_ID, "shift")) + def test_service_shutdown_clears_latches_for_next_start(self) -> None: + backend = FakeKeyboardBackend() + context = make_test_context(backend) + spec = KeySpec(label="Shift", row=0, column=0, key_id="shift", io_key="leftshift", latchable=True) + + context.keyboard.register_key_spec(LAYOUT_ID, spec) + context.dispatcher.dispatch_command(KeyboardSyncLatchedKey(LAYOUT_ID, spec, True)) + context.keyboard.shutdown() + + self.assertIsNone(context.state.get(keyboard_latches_namespace(LAYOUT_ID), "shift")) + self.assertFalse(context.keyboard.is_latched(LAYOUT_ID, "shift")) + def test_widget_renders_latched_style_from_snapshot(self) -> None: _app() backend = FakeKeyboardBackend() diff --git a/tests/test_kwin_input_panel.py b/tests/test_kwin_input_panel.py new file mode 100644 index 0000000..c16ba9e --- /dev/null +++ b/tests/test_kwin_input_panel.py @@ -0,0 +1,77 @@ +from __future__ import annotations + +import unittest +from types import SimpleNamespace +from unittest.mock import Mock, patch + +from axidev_osk.platform import kwin_input_panel + + +class KWinInputPanelTests(unittest.TestCase): + def tearDown(self) -> None: + kwin_input_panel._clients.clear() + + def test_select_output_matches_qt_screen_name(self) -> None: + first = kwin_input_panel._Output(10, 100, "HDMI-A-1") + selected = kwin_input_panel._Output(11, 101, "DP-1") + + self.assertIs( + kwin_input_panel._select_output((first, selected), "DP-1"), + selected, + ) + + def test_select_output_rejects_unknown_qt_screen(self) -> None: + output = kwin_input_panel._Output(10, 100, "HDMI-A-1") + + with self.assertRaisesRegex( + kwin_input_panel.KWinInputPanelError, + "available outputs: HDMI-A-1", + ): + kwin_input_panel._select_output((output,), "DP-1") + + def test_client_is_reused_for_rebuilt_windows_on_one_display(self) -> None: + library = Mock() + client = Mock() + with patch.object(kwin_input_panel, "_InputPanelClient", return_value=client) as build: + first = kwin_input_panel._client_for_display(library, 42) + second = kwin_input_panel._client_for_display(library, 42) + + self.assertIs(first, client) + self.assertIs(second, client) + build.assert_called_once_with(library, 42) + + def test_attachment_releases_proxy_once(self) -> None: + client = Mock() + attachment = kwin_input_panel.KWinInputPanelAttachment(client, 99) + + attachment.close() + attachment.close() + + client.destroy_proxy.assert_called_once_with(99) + + def test_failed_client_initialization_removes_listener_proxies(self) -> None: + library = Mock() + library.wl_proxy_marshal_flags.return_value = 77 + library.wl_proxy_add_listener.return_value = 0 + protocol = SimpleNamespace(registry=kwin_input_panel._WlInterface()) + + with ( + patch.object(kwin_input_panel, "_Protocol", return_value=protocol), + patch.object( + kwin_input_panel._InputPanelClient, + "_roundtrip", + side_effect=kwin_input_panel.KWinInputPanelError("roundtrip failed"), + ), + self.assertRaisesRegex( + kwin_input_panel.KWinInputPanelError, + "roundtrip failed", + ), + ): + kwin_input_panel._InputPanelClient(library, 42) + + destroyed = [call.args[0].value for call in library.wl_proxy_destroy.call_args_list] + self.assertEqual(destroyed, [77]) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_kwin_lock.py b/tests/test_kwin_lock.py new file mode 100644 index 0000000..5f55327 --- /dev/null +++ b/tests/test_kwin_lock.py @@ -0,0 +1,47 @@ +from __future__ import annotations + +import unittest +from unittest.mock import Mock, patch + +from PySide6.QtDBus import QDBusMessage + +from axidev_osk.services.kwin_lock import KWinLockService + + +class KWinLockServiceTests(unittest.TestCase): + def test_lock_signals_are_bound_to_screen_locker_services(self) -> None: + connection = Mock() + connection.isConnected.return_value = True + connection.connect.return_value = True + reply = Mock() + reply.type.return_value = QDBusMessage.MessageType.ReplyMessage + reply.arguments.return_value = [False] + screen_saver = Mock() + screen_saver.call.return_value = reply + + with ( + patch( + "axidev_osk.services.kwin_lock.QDBusConnection.sessionBus", + return_value=connection, + ), + patch( + "axidev_osk.services.kwin_lock.QDBusInterface", + side_effect=(Mock(), screen_saver), + ), + ): + service = KWinLockService() + service.start(Mock()) + service.stop() + + self.assertEqual( + [call.args[0] for call in connection.connect.call_args_list], + ["org.kde.screensaver", "org.freedesktop.ScreenSaver"], + ) + self.assertEqual( + [call.args[0] for call in connection.disconnect.call_args_list], + ["org.kde.screensaver", "org.freedesktop.ScreenSaver"], + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_linux_greeter.py b/tests/test_linux_greeter.py index 6c24594..dff46f3 100644 --- a/tests/test_linux_greeter.py +++ b/tests/test_linux_greeter.py @@ -120,11 +120,221 @@ def test_install_and_remove_restore_exact_config(self) -> None: class NativeAdapterTests(unittest.TestCase): - def test_plasma_service_stops_with_greeter_target(self) -> None: - text = linux_greeter._plasma_service_text() + PLASMA_KWIN_UNIT = ( + "[Unit]\n" + "Description=KDE Window Manager\n" + "[Service]\n" + "ExecStart=/usr/bin/kwin_wayland --no-lockscreen --inputmethod plasma-keyboard --locale1\n" + ) + + def test_plasma_input_method_uses_installed_launcher(self) -> None: + launcher = Path("/opt/axidev-osk/bin/axidev-osk") + text = linux_greeter._plasma_input_method_text(launcher) + + self.assertIn(f"Exec={launcher}\n", text) + self.assertIn("X-KDE-Wayland-VirtualKeyboard=true", text) + + def test_plasma_kwin_config_preserves_unmanaged_content(self) -> None: + original = ( + "# keep\n" + "[Wayland]\n" + "InputMethod=/usr/share/applications/other.desktop\n" + "Unmanaged=value\n" + "[Other]\n" + "VirtualKeyboardMode=1\n" + ) + + managed = linux_greeter._plasma_kwin_config_text(original) + + self.assertIn("# keep\n", managed) + self.assertIn("Unmanaged=value\n", managed) + self.assertIn("[Other]\nVirtualKeyboardMode=1\n", managed) + self.assertEqual(managed.count("InputMethod="), 1) + self.assertIn(f"InputMethod={linux_greeter.PLASMA_INPUT_METHOD_PATH}\n", managed) + self.assertIn("VirtualKeyboardMode=2\n", managed) + + def test_plasma_kwin_dropin_replaces_only_input_method(self) -> None: + with TemporaryDirectory() as temporary: + unit = Path(temporary) / "plasma-login-kwin_wayland.service" + unit.write_text(self.PLASMA_KWIN_UNIT, encoding="utf-8") + launcher = Path("/opt/axidev-osk/bin/axidev-osk") + + with patch.object(linux_greeter, "PLASMA_KWIN_UNIT_PATHS", (unit,)): + dropin = linux_greeter._plasma_kwin_dropin_text(launcher) + + self.assertIn("ExecStart=\n", dropin) + self.assertIn("--inputmethod", dropin) + self.assertIn(str(launcher), dropin) + self.assertIn("--no-lockscreen", dropin) + self.assertIn("--locale1", dropin) + self.assertNotIn("--inputmethod plasma-keyboard", dropin) + + def test_plasma_lock_screen_patch_is_additive_and_reversible(self) -> None: + original = ( + "Item {\n" + " MouseArea {\n" + " id: lockScreenRoot\n\n" + " property bool uiVisible: false\n" + " }\n" + "}\n" + ) + + managed = linux_greeter._plasma_lock_screen_ui_text(original) + + self.assertIn(linux_greeter.PLASMA_LOCK_SCREEN_PATCH, managed) + self.assertEqual(linux_greeter._plasma_lock_screen_ui_text(managed), managed) + self.assertEqual(linux_greeter._plasma_lock_screen_ui_without_patch(managed), original) + + def test_plasma_lock_screen_patch_rejects_changed_markers(self) -> None: + changed = ( + "Item {\n" + " MouseArea {\n" + " id: lockScreenRoot\n" + " // BEGIN AXIDEV OSK MANAGED\n" + " changed content\n" + " // END AXIDEV OSK MANAGED\n" + " }\n" + "}\n" + ) + + with self.assertRaisesRegex(linux.LinuxSetupError, "changed Axidev"): + linux_greeter._plasma_lock_screen_ui_text(changed) + with self.assertRaisesRegex(linux.LinuxSetupError, "changed Axidev"): + linux_greeter._plasma_lock_screen_ui_without_patch(changed) + + def test_plasma_version_is_read_from_owning_rpm(self) -> None: + completed = Mock(returncode=0, stdout="6.7.4") + with ( + patch.object( + linux_greeter.shutil, + "which", + side_effect=lambda command: "/usr/bin/rpm" if command == "rpm" else None, + ), + patch.object(linux_greeter.subprocess, "run", return_value=completed), + ): + version = linux_greeter._plasma_version() + + self.assertEqual(version, (6, 7, 4)) + + def test_plasma_lock_screen_version_range_excludes_plasma_7(self) -> None: + with patch.object(linux_greeter, "_plasma_version", return_value=(6, 7, 0)): + self.assertTrue(linux_greeter._plasma_lock_screen_version_supported()) + with patch.object(linux_greeter, "_plasma_version", return_value=(6, 6, 5)): + self.assertFalse(linux_greeter._plasma_lock_screen_version_supported()) + with patch.object(linux_greeter, "_plasma_version", return_value=(7, 0, 0)): + self.assertFalse(linux_greeter._plasma_lock_screen_version_supported()) + with self.assertRaisesRegex(linux.LinuxSetupError, "<7.0.0"): + linux_greeter._require_supported_plasma_lock_screen_version() + + def test_plasma_install_and_remove_restore_kwin_config(self) -> None: + with TemporaryDirectory() as temporary: + root = Path(temporary) + input_method = root / "axidev-osk-input-panel.desktop" + kwin_dropin = root / "50-axidev-osk.conf" + kwin_unit = root / "plasma-login-kwin_wayland.service" + kwinrc = root / "kwinrc" + lock_screen_ui = root / "LockScreenUi.qml" + state_path = root / "greeter.json" + original = "[Wayland]\nUnmanaged=value\n" + original_lock_screen_ui = ( + "Item {\n" + " MouseArea {\n" + " id: lockScreenRoot\n\n" + " property bool uiVisible: false\n" + " }\n" + "}\n" + ) + kwinrc.write_text(original, encoding="utf-8") + lock_screen_ui.write_text(original_lock_screen_ui, encoding="utf-8") + kwin_unit.write_text(self.PLASMA_KWIN_UNIT, encoding="utf-8") + launcher = Path("/opt/axidev-osk/bin/axidev-osk") + account = linux.Account("plasmalogin", 981, 981, root) + + with ( + patch.object(linux_greeter, "PLASMA_INPUT_METHOD_PATH", input_method), + patch.object(linux_greeter, "PLASMA_KWIN_DROPIN_PATH", kwin_dropin), + patch.object(linux_greeter, "PLASMA_KWIN_UNIT_PATHS", (kwin_unit,)), + patch.object(linux_greeter, "KWIN_CONFIG_PATH", kwinrc), + patch.object(linux_greeter, "PLASMA_LOCK_SCREEN_UI_PATH", lock_screen_ui), + patch.object(linux_greeter, "STATE_PATH", state_path), + patch.object(linux_greeter, "_plasma_version", return_value=(6, 7, 4)), + patch.object(linux, "_resolve_account", return_value=account), + ): + prepared_account, details = linux_greeter._prepare_plasma(launcher) + linux_greeter._install_manager( + "plasma-login", + linux_greeter._manager_adapter("plasma-login"), + prepared_account, + launcher, + details, + ) + state = json.loads(state_path.read_text(encoding="utf-8")) + self.assertIn("VirtualKeyboardMode=2", kwinrc.read_text(encoding="utf-8")) + self.assertTrue(input_method.is_file()) + self.assertTrue(kwin_dropin.is_file()) + self.assertIn( + linux_greeter.PLASMA_LOCK_SCREEN_PATCH, + lock_screen_ui.read_text(encoding="utf-8"), + ) + + lock_screen_ui.write_text(original_lock_screen_ui, encoding="utf-8") + kwin_dropin.write_text("changed after setup\n", encoding="utf-8") + with patch.object(linux_greeter, "_runtime_launcher", return_value=launcher): + self.assertTrue(linux_greeter._repair_plasma_lock_screen_patch(state)) + self.assertIn( + linux_greeter.PLASMA_LOCK_SCREEN_PATCH, + lock_screen_ui.read_text(encoding="utf-8"), + ) + + kwin_dropin.write_text( + linux_greeter._plasma_kwin_dropin_text(launcher), + encoding="utf-8", + ) + linux_greeter._remove_plasma(launcher, state) - self.assertIn("PartOf=plasma-login-wayland.target", text) - self.assertIn(str(linux_greeter.NATIVE_SUPERVISOR_PATH), text) + self.assertEqual(prepared_account.name, "plasmalogin") + self.assertEqual(kwinrc.read_text(encoding="utf-8"), original) + self.assertFalse(input_method.exists()) + self.assertFalse(kwin_dropin.exists()) + self.assertEqual(lock_screen_ui.read_text(encoding="utf-8"), original_lock_screen_ui) + + def test_legacy_plasma_remove_keeps_working(self) -> None: + launcher = Path("/opt/axidev-osk/bin/axidev-osk") + legacy_state = {"schema": 1, "manager": "plasma-login", "account": "plasmalogin"} + with TemporaryDirectory() as temporary: + root = Path(temporary) + supervisor = root / "supervisor" + service = root / "service" + wants = root / "wants" + with ( + patch.object(linux_greeter, "NATIVE_SUPERVISOR_PATH", supervisor), + patch.object(linux_greeter, "PLASMA_SERVICE_PATH", service), + patch.object(linux_greeter, "PLASMA_WANTS_PATH", wants), + ): + supervisor.write_text( + linux_greeter._native_supervisor_text(launcher), encoding="utf-8" + ) + service.write_text(linux_greeter._plasma_service_text(), encoding="utf-8") + wants.symlink_to(service) + linux_greeter._remove_plasma(launcher, legacy_state) + + self.assertFalse(supervisor.exists()) + self.assertFalse(service.exists()) + self.assertFalse(wants.exists()) + + def test_removable_symlink_accepts_an_equivalent_target_path(self) -> None: + with TemporaryDirectory() as temporary: + root = Path(temporary) + actual = root / "actual" + alias = root / "alias" + actual.mkdir() + alias.symlink_to(actual, target_is_directory=True) + target = alias / "service" + target.write_text("service", encoding="utf-8") + link = root / "wants" + link.symlink_to(target) + + linux_greeter._require_removable_symlink(link, target) def test_lightdm_uses_native_greeter_wrapper(self) -> None: wrapper = linux_greeter._lightdm_wrapper_text(Path("/opt/axidev-osk/bin/axidev-osk")) @@ -264,6 +474,5 @@ def test_attached_supervisor_does_not_stop_parent(self) -> None: self.assertEqual(result, 0) - if __name__ == "__main__": unittest.main() diff --git a/tests/test_service_registry.py b/tests/test_service_registry.py index 5518255..aef38ef 100644 --- a/tests/test_service_registry.py +++ b/tests/test_service_registry.py @@ -58,6 +58,16 @@ def stop(self) -> None: class ServiceRegistryTests(unittest.TestCase): + def test_deferred_service_is_excluded_only_from_autostart(self) -> None: + services = ServiceRegistry() + deferred = RecordingService("deferred", []) + automatic = RecordingService("automatic", []) + services.register("deferred", deferred, autostart=False) + services.register("automatic", automatic) + + self.assertEqual(tuple(services.services()), (deferred, automatic)) + self.assertEqual(tuple(services.autostart_services()), (automatic,)) + def test_runtime_starts_and_stops_registered_services_in_order(self) -> None: calls: list[str] = [] services = ServiceRegistry() diff --git a/tests/test_window_builder.py b/tests/test_window_builder.py index d9690cf..da1c210 100644 --- a/tests/test_window_builder.py +++ b/tests/test_window_builder.py @@ -1,7 +1,7 @@ from __future__ import annotations import unittest -from unittest.mock import patch +from unittest.mock import Mock, patch from PySide6.QtCore import Qt from PySide6.QtWidgets import QApplication, QLabel, QPushButton @@ -91,6 +91,31 @@ def _build_keyboard_window(backend: FakeKeyboardBackend): class RuntimeWindowLayoutTests(unittest.TestCase): """Tests covering the default keyboard window built via ``build_window``.""" + + def test_failed_content_build_releases_platform_resources(self) -> None: + _app() + config = build_default_app_config() + surfaces = SurfaceRegistry() + surfaces.build = Mock(side_effect=RuntimeError("surface failed")) + context = make_test_context( + FakeKeyboardBackend(ready=True), + config=config, + components=ComponentRegistry(), + surfaces=surfaces, + ) + overlay = Mock() + + with ( + patch( + "axidev_osk.windows.builder.configure_always_on_top_window", + return_value=overlay, + ), + self.assertRaisesRegex(RuntimeError, "surface failed"), + ): + build_window(config.windows[0], context) + + overlay.release_resources.assert_called_once_with() + def test_custom_chrome_puts_resize_handle_in_title_bar(self) -> None: _app() overlay = FakeOverlayController() @@ -207,6 +232,27 @@ def test_keyboard_window_uses_configured_normal_opacity(self) -> None: self.addCleanup(window.close) self.assertAlmostEqual(window.windowOpacity(), 0.85, delta=0.005) + def test_runtime_window_can_hide_and_restore_custom_close_control(self) -> None: + _app() + overlay = FakeOverlayController() + + with patch( + "axidev_osk.windows.builder.configure_always_on_top_window", + return_value=overlay, + ): + window = _build_keyboard_window(FakeKeyboardBackend(ready=True)) + + self.addCleanup(window.close) + close_button = window.findChild(QPushButton, "layerShellCloseButton") + self.assertIsNotNone(close_button) + self.assertFalse(close_button.isHidden()) + + window.set_close_enabled(False) + self.assertTrue(close_button.isHidden()) + + window.set_close_enabled(True) + self.assertFalse(close_button.isHidden()) + def test_runtime_window_and_components_expose_dynamic_identity_properties(self) -> None: _app() overlay = FakeOverlayController() diff --git a/tests/test_window_manager.py b/tests/test_window_manager.py index f88d13b..03cdbf2 100644 --- a/tests/test_window_manager.py +++ b/tests/test_window_manager.py @@ -76,7 +76,7 @@ def test_input_blocker_allows_only_the_recovery_component(self) -> None: self.assertTrue(blocker.eventFilter(window, event)) def test_toggle_opacity_restores_configured_opacity_on_second_call(self) -> None: - window = QWidget() + window = Mock() self.manager._windows = {"window:keyboard": window} self.manager.toggle_opacity( @@ -85,7 +85,7 @@ def test_toggle_opacity_restores_configured_opacity_on_second_call(self) -> None opacity=0.01, ) - self.assertAlmostEqual(window.windowOpacity(), 0.01, delta=0.005) + window.set_visual_opacity.assert_called_once_with(0.01) self.assertIn("window:keyboard", self.manager._input_blockers) self.manager.toggle_opacity( @@ -94,12 +94,14 @@ def test_toggle_opacity_restores_configured_opacity_on_second_call(self) -> None opacity=0.01, ) - self.assertAlmostEqual(window.windowOpacity(), 0.85, delta=0.005) + self.assertEqual( + window.set_visual_opacity.call_args_list, + [unittest.mock.call(0.01), unittest.mock.call(0.85)], + ) self.assertNotIn("window:keyboard", self.manager._input_blockers) def test_show_restores_configured_opacity_and_removes_input_blocker(self) -> None: - window = QWidget() - window.setWindowOpacity(0.01) + window = Mock() blocker = _WindowInputBlocker(window, "key:ghost") self.app.installEventFilter(blocker) self.manager._windows = {"window:keyboard": window} @@ -107,9 +109,8 @@ def test_show_restores_configured_opacity_and_removes_input_blocker(self) -> Non self.manager.show("window:keyboard") - self.assertAlmostEqual(window.windowOpacity(), 0.85, delta=0.005) + window.set_visual_opacity.assert_called_once_with(0.85) self.assertNotIn("window:keyboard", self.manager._input_blockers) - if __name__ == "__main__": unittest.main() diff --git a/tests/test_window_opacity.py b/tests/test_window_opacity.py new file mode 100644 index 0000000..6cf4ce5 --- /dev/null +++ b/tests/test_window_opacity.py @@ -0,0 +1,51 @@ +from __future__ import annotations + +import unittest +from unittest.mock import patch + +from PySide6.QtWidgets import QApplication, QMainWindow, QWidget + +from axidev_osk.windows.opacity import WindowOpacityController + + +class WindowOpacityControllerTests(unittest.TestCase): + @classmethod + def setUpClass(cls) -> None: + cls.app = QApplication.instance() or QApplication([]) + + def test_wayland_applies_every_opacity_value_to_content(self) -> None: + window = QMainWindow() + content = QWidget() + window.setCentralWidget(content) + controller = WindowOpacityController(window) + + with patch( + "axidev_osk.windows.opacity.QGuiApplication.platformName", + return_value="wayland", + ): + controller.set_opacity(0.85) + effect = content.graphicsEffect() + self.assertIsNotNone(effect) + self.assertAlmostEqual(effect.opacity(), 0.85, delta=0.005) + + controller.set_opacity(0.01) + self.assertIs(content.graphicsEffect(), effect) + self.assertAlmostEqual(effect.opacity(), 0.01, delta=0.005) + + def test_non_wayland_applies_opacity_to_native_window(self) -> None: + window = QMainWindow() + window.setCentralWidget(QWidget()) + controller = WindowOpacityController(window) + + with patch( + "axidev_osk.windows.opacity.QGuiApplication.platformName", + return_value="windows", + ): + controller.set_opacity(0.42) + + self.assertAlmostEqual(window.windowOpacity(), 0.42, delta=0.005) + self.assertIsNone(window.centralWidget().graphicsEffect()) + + +if __name__ == "__main__": + unittest.main()