From ed8232429085f343be170f31997ea0a31e29b59e Mon Sep 17 00:00:00 2001 From: ChristopherLandaverde Date: Mon, 27 Apr 2026 22:50:13 -0400 Subject: [PATCH 1/8] =?UTF-8?q?docs:=20Phase=206=20design=20spec=20?= =?UTF-8?q?=E2=80=94=20Qt-path=20SIGTERM=20+=20systemd=20unit?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../2026-04-27-qt-sigterm-systemd-design.md | 165 ++++++++++++++++++ 1 file changed, 165 insertions(+) create mode 100644 docs/superpowers/specs/2026-04-27-qt-sigterm-systemd-design.md diff --git a/docs/superpowers/specs/2026-04-27-qt-sigterm-systemd-design.md b/docs/superpowers/specs/2026-04-27-qt-sigterm-systemd-design.md new file mode 100644 index 0000000..ef2a02c --- /dev/null +++ b/docs/superpowers/specs/2026-04-27-qt-sigterm-systemd-design.md @@ -0,0 +1,165 @@ +# Qt-path SIGTERM + systemd unit — Design Spec + +**Date:** 2026-04-27 +**Phase:** 6 + +--- + +## Problem + +The command-only listen path handles SIGTERM cleanly via +`_sigterm_raises_keyboard_interrupt`. The Qt path (ring bindings) has no SIGTERM +handling: `app.exec()` blocks the main thread inside Qt's native event loop, +so Python signal handlers don't fire reliably. `systemctl stop` therefore +either hangs or hard-kills the process, leaving the grabbed device in a bad +state. + +This phase fixes that and adds a `logitechmouse install-service` command so the +package works as a systemd user service out of the box. + +--- + +## Architecture + +### Qt-path SIGTERM fix + +All changes are in `src/logitechmouse/cli/listen.py`, inside `_run_with_qt`, +before the `app.exec()` call. + +**Setup (once, before `app.exec()`):** + +1. Create a `socket.socketpair(AF_UNIX, SOCK_STREAM)` — a self-pipe pair. Set + both ends non-blocking (required by `set_wakeup_fd`). +2. Call `signal.set_wakeup_fd(write_fd)` — Python writes a byte to `write_fd` + whenever any signal arrives, regardless of which thread is running. +3. Install a no-op Python handler for SIGTERM: + `signal.signal(SIGTERM, lambda s, f: None)`. This is required — `set_wakeup_fd` + only fires for signals that have a Python-level handler installed (not `SIG_DFL` + or `SIG_IGN`). The no-op prevents the OS from terminating the process before + the event loop can respond. +4. Create `QSocketNotifier(read_fd, QSocketNotifier.Type.Read)` — Qt watches + the read end on the main thread's event loop. +5. Connect `notifier.activated` to a slot that drains the read end and calls + `app.quit()`. + +**Shutdown sequence on SIGTERM:** + +``` +systemctl stop + → SIGTERM delivered to process + → no-op Python handler runs (process not killed) + → Python writes byte to write_fd (set_wakeup_fd) + → QSocketNotifier fires on main thread + → slot drains read_fd, calls app.quit() + → app.exec() returns + → existing finally block: virt.close(), device.ungrab() + → worker thread sees OSError from read_loop, emits finished + → thread.wait(2000) joins worker + → process exits 0 +``` + +No new teardown code is needed — the existing `finally` block handles cleanup. +SIGINT (Ctrl+C) follows the same path since `set_wakeup_fd` catches all signals +that have a Python handler installed. Restore the previous SIGTERM handler after +`app.exec()` returns (in a `finally` block) so the context is clean. + +The `_sigterm_raises_keyboard_interrupt` context manager on the command-only +path is **unchanged**. + +**Failure mode:** If `set_wakeup_fd` fails (unusual environment, fd exhaustion), +log a warning and continue without the notifier. The process will respond to +SIGTERM via OS default (hard kill) — no worse than today. + +--- + +### `install-service` command + +New module: `src/logitechmouse/cli/install_service.py` +Registered as a subcommand in the existing CLI entry point. + +``` +logitechmouse install-service --config PATH [--force] +``` + +**Sequence:** + +1. Resolve the `logitechmouse` binary via `shutil.which("logitechmouse")`, + fallback to `sys.argv[0]`. Fail with exit 1 if neither resolves. +2. Resolve `--config PATH` to an absolute path. Fail with exit 1 if the file + does not exist. +3. Render the unit file from a hardcoded template string (no external file). +4. Target path: `~/.config/systemd/user/logitechmouse.service`. Create + intermediate directories if absent. +5. If the file already exists and `--force` is not set, exit 1 with a clear + message. +6. Write the file. +7. Run `systemctl --user daemon-reload` via `subprocess.run`. On failure, warn + only — do not exit 1 (systemd may not be running in all environments). +8. Print next-step instructions to stdout. + +**Generated unit file template:** + +```ini +[Unit] +Description=Logitech Mouse button remapper +After=graphical-session.target + +[Service] +ExecStart={exec_start} listen --config {config_path} +Restart=on-failure +RestartSec=3 + +[Install] +WantedBy=default.target +``` + +--- + +## Error handling + +| Scenario | Behaviour | +|---|---| +| Config file not found | exit 1, no unit file written | +| Binary not resolvable | exit 1, advise checking PATH | +| Unit file exists, no `--force` | exit 1, tell user to use `--force` | +| Unit file dir not writable | surface `OSError` naturally | +| `daemon-reload` fails | warn, don't fail (file is written) | +| `set_wakeup_fd` fails | log warning, continue without notifier | +| Worker thread doesn't exit in 2s | existing `thread.wait(2000)` behaviour | + +--- + +## Testing + +### `install-service` + +- Unit tests in `tests/test_install_service.py` using `tmp_path` fixture. +- Mock `shutil.which` and `subprocess.run`. +- Assert generated file content for happy path. +- Assert exit codes and messages for each error scenario. + +### Qt SIGTERM + +- Test in `tests/test_listen_qt_sigterm.py`. +- Spin up a minimal Qt app using the same `QSocketNotifier` setup, send + `signal.SIGTERM` to `os.getpid()`, assert `app.exec()` returns and the + teardown ran. +- Pattern mirrors the existing SIGTERM behavioural test in + `tests/test_listen_grab.py`. + +### End-to-end + +- Optional test gated behind `requires_uinput` (same pattern as Phase 5): + grab a real device, send SIGTERM to the Qt path, assert device is ungrabbed + and virtual device is closed. + +--- + +## Out of scope + +- Wayland support (documented as a separate phase). +- App-specific profiles. +- `logitechmouse uninstall-service` command (can be added later; manual removal + is one `rm` command). +- Packaging the unit file as a static asset (not viable due to dynamic + `ExecStart` path). From 25b03bf9e5f09a57195469ba94e734d2aa01c7df Mon Sep 17 00:00:00 2001 From: ChristopherLandaverde Date: Mon, 27 Apr 2026 22:55:45 -0400 Subject: [PATCH 2/8] =?UTF-8?q?docs:=20Phase=206=20implementation=20plan?= =?UTF-8?q?=20=E2=80=94=20Qt-path=20SIGTERM=20+=20systemd=20unit?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../plans/2026-04-27-qt-sigterm-systemd.md | 724 ++++++++++++++++++ 1 file changed, 724 insertions(+) create mode 100644 docs/superpowers/plans/2026-04-27-qt-sigterm-systemd.md diff --git a/docs/superpowers/plans/2026-04-27-qt-sigterm-systemd.md b/docs/superpowers/plans/2026-04-27-qt-sigterm-systemd.md new file mode 100644 index 0000000..d86546f --- /dev/null +++ b/docs/superpowers/plans/2026-04-27-qt-sigterm-systemd.md @@ -0,0 +1,724 @@ +# Qt-path SIGTERM + systemd unit Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Fix SIGTERM handling on the Qt listen path using `set_wakeup_fd` + `QSocketNotifier`, and add a `logitechmouse install-service` command that writes a systemd user unit file. + +**Architecture:** Before `app.exec()`, a `socketpair` + `signal.set_wakeup_fd` wires all Python-handled signals to a `QSocketNotifier` on the main thread; the notifier slot calls `app.quit()`, causing `app.exec()` to return and the existing `finally` teardown block to run. The `install-service` command resolves the installed binary path, renders a unit file template, writes it to `~/.config/systemd/user/`, and runs `systemctl --user daemon-reload`. + +**Tech Stack:** Python 3.11+, PyQt6 ~= 6.6, evdev, pytest, pytest-qt + +--- + +## File map + +| File | Action | Responsibility | +|------|--------|----------------| +| `src/logitechmouse/cli/listen.py` | Modify | Add socketpair + `set_wakeup_fd` + `QSocketNotifier` before `app.exec()` in `_run_with_qt`; add `import socket` | +| `src/logitechmouse/cli/install_service.py` | Create | `run(args)` — resolve binary, render template, write unit file, reload systemd | +| `src/logitechmouse/main.py` | Modify | Add `install-service` subparser (`--config` required, `--force` optional) and dispatch | +| `tests/test_listen_qt_sigterm.py` | Create | Unit + integration tests for the SIGTERM mechanism and `_run_with_qt` signal setup | +| `tests/test_install_service.py` | Create | Tests for all `install-service` paths (happy, errors, force) | + +--- + +## Task 1: Qt mechanism unit tests + +**Files:** +- Create: `tests/test_listen_qt_sigterm.py` + +- [ ] **Step 1: Write the two mechanism tests** + +```python +# tests/test_listen_qt_sigterm.py +from __future__ import annotations + +import os +import signal +import socket +import sys + +import pytest + +pytest.importorskip("PyQt6") + +from PyQt6.QtCore import QSocketNotifier, QTimer +from PyQt6.QtWidgets import QApplication + + +def _app() -> QApplication: + return QApplication.instance() or QApplication(sys.argv) + + +def test_socket_notifier_fires_when_written_to(): + """Writing to the write end of a socketpair causes QSocketNotifier to fire + and call app.quit(); app.exec() returns.""" + app = _app() + r, w = socket.socketpair(socket.AF_UNIX, socket.SOCK_STREAM) + r.setblocking(False) + w.setblocking(False) + + fired = [] + notifier = QSocketNotifier(r.fileno(), QSocketNotifier.Type.Read) + + def _slot(): + try: + r.recv(256) + except OSError: + pass + fired.append(True) + app.quit() + + notifier.activated.connect(_slot) + QTimer.singleShot(50, lambda: w.send(b"\x00")) + app.exec() + + notifier.setEnabled(False) + r.close() + w.close() + + assert fired, "QSocketNotifier slot was not called" + + +def test_sigterm_via_set_wakeup_fd_triggers_notifier(): + """SIGTERM delivered to this process writes a byte via set_wakeup_fd, + which fires the QSocketNotifier and causes app.exec() to return.""" + app = _app() + r, w = socket.socketpair(socket.AF_UNIX, socket.SOCK_STREAM) + r.setblocking(False) + w.setblocking(False) + + prev_fd = signal.set_wakeup_fd(w.fileno()) + prev_sigterm = signal.getsignal(signal.SIGTERM) + signal.signal(signal.SIGTERM, lambda s, f: None) # no-op; must be non-SIG_DFL + + fired = [] + notifier = QSocketNotifier(r.fileno(), QSocketNotifier.Type.Read) + + def _slot(): + try: + r.recv(256) + except OSError: + pass + fired.append(True) + app.quit() + + notifier.activated.connect(_slot) + QTimer.singleShot(50, lambda: os.kill(os.getpid(), signal.SIGTERM)) + app.exec() + + notifier.setEnabled(False) + signal.set_wakeup_fd(prev_fd) + signal.signal(signal.SIGTERM, prev_sigterm) + r.close() + w.close() + + assert fired, "SIGTERM did not trigger QSocketNotifier via set_wakeup_fd" +``` + +- [ ] **Step 2: Run the tests — expect both to PASS** (mechanism is pure Qt/Python, no listen.py changes needed) + +```bash +cd /home/chrisland/projects/logitechmouse +pytest tests/test_listen_qt_sigterm.py -v +``` + +Expected: 2 passed (or skipped if no display). If DISPLAY is unset, add `DISPLAY=:0` or run in a display session. + +- [ ] **Step 3: Commit** + +```bash +git add tests/test_listen_qt_sigterm.py +git commit -m "test(qt_sigterm): mechanism unit tests — socketpair+notifier and set_wakeup_fd" +``` + +--- + +## Task 2: Qt SIGTERM integration tests + +**Files:** +- Modify: `tests/test_listen_qt_sigterm.py` (append tests) + +- [ ] **Step 1: Append integration tests for `_run_with_qt` signal setup** + +```python +# append to tests/test_listen_qt_sigterm.py +import argparse +from unittest.mock import MagicMock, patch + +from logitechmouse.cli import listen as listen_mod + + +@pytest.fixture +def ring_cfg(tmp_path): + cfg = tmp_path / "config.toml" + cfg.write_text( + '[device]\npath = "/dev/input/event99"\n\n' + '[actions.a]\nkind = "command"\ncommand = "true"\n\n' + '[rings.r]\nsegments = [\n' + ' { action = "a", label = "A" },\n' + ' { action = "a", label = "B" },\n' + ']\n\n' + '[bindings.b1]\ntrigger = "BTN_BACK"\ntarget = "ring:r"\n' + ) + return cfg + + +def test_run_with_qt_installs_noop_sigterm_and_restores_it(ring_cfg): + """_run_with_qt must replace the SIGTERM handler with a no-op while + app.exec() runs, then restore the original handler on exit.""" + args = argparse.Namespace(config=ring_cfg, device=None) + fake_dev = MagicMock(path="/dev/input/event99", name="fake") + + original_sigterm = signal.getsignal(signal.SIGTERM) + captured = {} + + def fake_exec(): + captured["sigterm"] = signal.getsignal(signal.SIGTERM) + return 0 + + with patch.object(listen_mod.EvdevBackend, "resolve", return_value=fake_dev), \ + patch.object(listen_mod.EvdevBackend, "read_loop", return_value=iter([])), \ + patch("logitechmouse.cli.listen.try_grab", return_value=None), \ + patch("PyQt6.QtWidgets.QApplication.exec", side_effect=fake_exec), \ + patch("PyQt6.QtCore.QThread.start"), \ + patch("PyQt6.QtCore.QThread.wait"): + listen_mod.run(args) + + assert captured.get("sigterm") is not original_sigterm, \ + "SIGTERM handler must be replaced (no-op) during app.exec()" + assert signal.getsignal(signal.SIGTERM) is original_sigterm, \ + "SIGTERM handler must be restored after _run_with_qt returns" + + +def test_run_with_qt_tears_down_virt_after_sigterm(ring_cfg): + """Teardown (virt.close, device.ungrab) must run after app.exec() returns + regardless of whether exit was triggered by SIGTERM or normal finish.""" + args = argparse.Namespace(config=ring_cfg, device=None) + fake_dev = MagicMock(path="/dev/input/event99", name="fake") + fake_virt = MagicMock() + + with patch.object(listen_mod.EvdevBackend, "resolve", return_value=fake_dev), \ + patch.object(listen_mod.EvdevBackend, "read_loop", return_value=iter([])), \ + patch("logitechmouse.cli.listen.try_grab", return_value=fake_virt), \ + patch("PyQt6.QtWidgets.QApplication.exec", return_value=0), \ + patch("PyQt6.QtCore.QThread.start"), \ + patch("PyQt6.QtCore.QThread.wait"): + listen_mod.run(args) + + fake_virt.close.assert_called_once() + fake_dev.ungrab.assert_called_once() +``` + +- [ ] **Step 2: Run — expect the two new integration tests to FAIL** + +```bash +pytest tests/test_listen_qt_sigterm.py -v +``` + +Expected: `test_run_with_qt_installs_noop_sigterm_and_restores_it` FAILS (handler not yet replaced), `test_run_with_qt_tears_down_virt_after_sigterm` may pass or fail. + +- [ ] **Step 3: Commit failing tests** + +```bash +git add tests/test_listen_qt_sigterm.py +git commit -m "test(qt_sigterm): integration tests for _run_with_qt signal setup (failing)" +``` + +--- + +## Task 3: Implement Qt-path SIGTERM fix + +**Files:** +- Modify: `src/logitechmouse/cli/listen.py` + +- [ ] **Step 1: Add `import socket` to the top of listen.py** + +Current imports (lines 1-8): +```python +from __future__ import annotations + +import argparse +import logging +import signal +import sys +from contextlib import contextmanager +from typing import Callable +``` + +Replace with: +```python +from __future__ import annotations + +import argparse +import logging +import signal +import socket +import sys +from contextlib import contextmanager +from typing import Callable +``` + +- [ ] **Step 2: Replace the `try: app.exec()` block in `_run_with_qt`** + +Find this block in `_run_with_qt` (currently after `thread.start()`, around line 273): +```python + try: + app.exec() + thread.wait(2000) + finally: + if virt is not None: + try: + virt.close() + except Exception: + logging.exception("virt.close() failed") + try: + device.ungrab() + except OSError: + pass + + return return_code["value"] +``` + +Replace with: +```python + from PyQt6.QtCore import QSocketNotifier + + r_sock, w_sock = socket.socketpair(socket.AF_UNIX, socket.SOCK_STREAM) + r_sock.setblocking(False) + w_sock.setblocking(False) + + _wakeup_active = False + prev_wakeup_fd = -1 + prev_sigterm = signal.getsignal(signal.SIGTERM) + prev_sigint = signal.getsignal(signal.SIGINT) + notifier = None + + try: + prev_wakeup_fd = signal.set_wakeup_fd(w_sock.fileno()) + _wakeup_active = True + except ValueError as exc: + logging.warning("set_wakeup_fd unavailable: %s; SIGTERM will hard-kill the Qt path", exc) + r_sock.close() + w_sock.close() + + if _wakeup_active: + signal.signal(signal.SIGTERM, lambda s, f: None) + signal.signal(signal.SIGINT, lambda s, f: None) + notifier = QSocketNotifier(r_sock.fileno(), QSocketNotifier.Type.Read) + + def _on_signal_wakeup(): + try: + r_sock.recv(256) + except OSError: + pass + app.quit() + + notifier.activated.connect(_on_signal_wakeup) + + try: + app.exec() + thread.wait(2000) + finally: + if _wakeup_active: + notifier.setEnabled(False) + signal.set_wakeup_fd(prev_wakeup_fd) + signal.signal(signal.SIGTERM, prev_sigterm) + signal.signal(signal.SIGINT, prev_sigint) + r_sock.close() + w_sock.close() + if virt is not None: + try: + virt.close() + except Exception: + logging.exception("virt.close() failed") + try: + device.ungrab() + except OSError: + pass + + return return_code["value"] +``` + +- [ ] **Step 3: Run all Qt SIGTERM tests — expect all to pass** + +```bash +pytest tests/test_listen_qt_sigterm.py -v +``` + +Expected: all 4 tests pass (2 mechanism + 2 integration). + +- [ ] **Step 4: Run full test suite — expect no regressions** + +```bash +pytest -v +``` + +Expected: all previously passing tests still pass. + +- [ ] **Step 5: Commit** + +```bash +git add src/logitechmouse/cli/listen.py +git commit -m "feat(listen): Qt-path SIGTERM via set_wakeup_fd + QSocketNotifier" +``` + +--- + +## Task 4: install-service tests + +**Files:** +- Create: `tests/test_install_service.py` + +- [ ] **Step 1: Write all install-service tests** + +```python +# tests/test_install_service.py +from __future__ import annotations + +import argparse +from pathlib import Path +from unittest.mock import MagicMock, patch + +import pytest + + +def _args(config: Path, force: bool = False) -> argparse.Namespace: + return argparse.Namespace(config=config, force=force) + + +def _run(args, home: Path): + from logitechmouse.cli import install_service as mod + with patch("pathlib.Path.home", return_value=home): + return mod.run(args) + + +def test_happy_path_writes_unit_file(tmp_path): + config = tmp_path / "config.toml" + config.write_text("[device]\npath = '/dev/input/event0'\n") + + with patch("logitechmouse.cli.install_service.shutil.which", + return_value="/usr/local/bin/logitechmouse"), \ + patch("logitechmouse.cli.install_service.subprocess.run", + return_value=MagicMock(returncode=0)): + rc = _run(_args(config), home=tmp_path) + + assert rc == 0 + unit = tmp_path / ".config" / "systemd" / "user" / "logitechmouse.service" + assert unit.exists() + content = unit.read_text() + assert "/usr/local/bin/logitechmouse" in content + assert str(config.resolve()) in content + assert "listen --config" in content + assert "Restart=on-failure" in content + + +def test_happy_path_runs_daemon_reload(tmp_path): + config = tmp_path / "config.toml" + config.write_text("") + + mock_run = MagicMock(return_value=MagicMock(returncode=0)) + with patch("logitechmouse.cli.install_service.shutil.which", + return_value="/usr/bin/logitechmouse"), \ + patch("logitechmouse.cli.install_service.subprocess.run", mock_run): + _run(_args(config), home=tmp_path) + + mock_run.assert_called_once() + cmd = mock_run.call_args[0][0] + assert "daemon-reload" in cmd + + +def test_config_not_found_returns_1_no_unit_written(tmp_path): + args = _args(tmp_path / "missing.toml") + with patch("logitechmouse.cli.install_service.shutil.which", + return_value="/usr/bin/logitechmouse"): + rc = _run(args, home=tmp_path) + + assert rc == 1 + unit = tmp_path / ".config" / "systemd" / "user" / "logitechmouse.service" + assert not unit.exists() + + +def test_binary_not_resolvable_returns_1(tmp_path): + config = tmp_path / "config.toml" + config.write_text("") + with patch("logitechmouse.cli.install_service.shutil.which", return_value=None), \ + patch("sys.argv", []): + rc = _run(_args(config), home=tmp_path) + assert rc == 1 + + +def test_existing_unit_without_force_returns_1_and_does_not_overwrite(tmp_path): + config = tmp_path / "config.toml" + config.write_text("") + unit_dir = tmp_path / ".config" / "systemd" / "user" + unit_dir.mkdir(parents=True) + unit = unit_dir / "logitechmouse.service" + unit.write_text("old content") + + with patch("logitechmouse.cli.install_service.shutil.which", + return_value="/usr/bin/logitechmouse"): + rc = _run(_args(config, force=False), home=tmp_path) + + assert rc == 1 + assert unit.read_text() == "old content" + + +def test_force_overwrites_existing_unit(tmp_path): + config = tmp_path / "config.toml" + config.write_text("") + unit_dir = tmp_path / ".config" / "systemd" / "user" + unit_dir.mkdir(parents=True) + (unit_dir / "logitechmouse.service").write_text("old content") + + with patch("logitechmouse.cli.install_service.shutil.which", + return_value="/usr/bin/logitechmouse"), \ + patch("logitechmouse.cli.install_service.subprocess.run", + return_value=MagicMock(returncode=0)): + rc = _run(_args(config, force=True), home=tmp_path) + + assert rc == 0 + content = (unit_dir / "logitechmouse.service").read_text() + assert content != "old content" + assert "listen --config" in content + + +def test_daemon_reload_failure_warns_but_returns_0(tmp_path): + config = tmp_path / "config.toml" + config.write_text("") + + with patch("logitechmouse.cli.install_service.shutil.which", + return_value="/usr/bin/logitechmouse"), \ + patch("logitechmouse.cli.install_service.subprocess.run", + return_value=MagicMock(returncode=1)): + rc = _run(_args(config), home=tmp_path) + + assert rc == 0 # warn only, file is written + + +def test_fallback_to_argv0_when_which_returns_none(tmp_path): + config = tmp_path / "config.toml" + config.write_text("") + + with patch("logitechmouse.cli.install_service.shutil.which", return_value=None), \ + patch("sys.argv", ["/home/user/.local/bin/logitechmouse", "install-service"]), \ + patch("logitechmouse.cli.install_service.subprocess.run", + return_value=MagicMock(returncode=0)): + rc = _run(_args(config), home=tmp_path) + + assert rc == 0 + unit = tmp_path / ".config" / "systemd" / "user" / "logitechmouse.service" + assert "/home/user/.local/bin/logitechmouse" in unit.read_text() +``` + +- [ ] **Step 2: Run — expect ImportError (module doesn't exist yet)** + +```bash +pytest tests/test_install_service.py -v +``` + +Expected: `ModuleNotFoundError` or `ImportError` — `logitechmouse.cli.install_service` does not exist. + +- [ ] **Step 3: Commit failing tests** + +```bash +git add tests/test_install_service.py +git commit -m "test(install_service): all cases — happy path, errors, force, daemon-reload" +``` + +--- + +## Task 5: Implement install-service + +**Files:** +- Create: `src/logitechmouse/cli/install_service.py` + +- [ ] **Step 1: Create the module** + +```python +# src/logitechmouse/cli/install_service.py +from __future__ import annotations + +import argparse +import logging +import shutil +import subprocess +import sys +from pathlib import Path + +_UNIT_TEMPLATE = """\ +[Unit] +Description=Logitech Mouse button remapper +After=graphical-session.target + +[Service] +ExecStart={exec_start} listen --config {config_path} +Restart=on-failure +RestartSec=3 + +[Install] +WantedBy=default.target +""" + + +def run(args: argparse.Namespace) -> int: + exec_start = shutil.which("logitechmouse") or (sys.argv[0] if sys.argv else None) + if not exec_start: + logging.error("cannot resolve logitechmouse binary; check your PATH") + return 1 + + config_path = Path(args.config).resolve() + if not config_path.exists(): + logging.error("config file not found: %s", config_path) + return 1 + + unit_dir = Path.home() / ".config" / "systemd" / "user" + unit_path = unit_dir / "logitechmouse.service" + + if unit_path.exists() and not getattr(args, "force", False): + logging.error( + "service file already exists: %s — use --force to overwrite", + unit_path, + ) + return 1 + + unit_dir.mkdir(parents=True, exist_ok=True) + unit_path.write_text( + _UNIT_TEMPLATE.format(exec_start=exec_start, config_path=config_path) + ) + logging.info("wrote %s", unit_path) + + result = subprocess.run( + ["systemctl", "--user", "daemon-reload"], + capture_output=True, + ) + if result.returncode != 0: + logging.warning( + "systemctl --user daemon-reload failed (rc=%d); " + "run it manually once a systemd user session is available", + result.returncode, + ) + + print( + f"\nService file written to {unit_path}\n" + f"Enable and start with:\n" + f" systemctl --user enable --now logitechmouse\n" + ) + return 0 +``` + +- [ ] **Step 2: Run install-service tests — expect all to pass** + +```bash +pytest tests/test_install_service.py -v +``` + +Expected: 8 tests pass. + +- [ ] **Step 3: Commit** + +```bash +git add src/logitechmouse/cli/install_service.py +git commit -m "feat(cli): install-service command — writes systemd user unit file" +``` + +--- + +## Task 6: Wire install-service into main.py + +**Files:** +- Modify: `src/logitechmouse/main.py` + +- [ ] **Step 1: Add the subparser and dispatch to `main.py`** + +In `build_parser()`, after the `p_run` subparser block (around line 46), add: + +```python + p_install = sub.add_parser( + "install-service", + help="Write a systemd user unit file for logitechmouse", + ) + p_install.add_argument( + "--config", + type=Path, + required=True, + help="Path to config TOML (baked into the unit file's ExecStart)", + ) + p_install.add_argument( + "--force", + action="store_true", + help="Overwrite an existing unit file", + ) +``` + +In `main()`, add a branch in the `if/elif` chain (after the `run` branch, before the `else`): + +```python + elif args.command == "install-service": + from .cli.install_service import run as run_cmd +``` + +- [ ] **Step 2: Smoke-test the CLI wiring** + +```bash +cd /home/chrisland/projects/logitechmouse +python -m logitechmouse install-service --help +``` + +Expected output includes `--config` (required) and `--force`. + +- [ ] **Step 3: Run full test suite** + +```bash +pytest -v +``` + +Expected: all tests pass (no regressions). + +- [ ] **Step 4: Commit** + +```bash +git add src/logitechmouse/main.py +git commit -m "feat(main): wire install-service subcommand" +``` + +--- + +## Task 7: Final verification and push + +- [ ] **Step 1: Run full test suite one more time** + +```bash +pytest -v 2>&1 | tail -20 +``` + +Expected: all green, no skips other than `requires_display` / `requires_uinput`. + +- [ ] **Step 2: Check that `install-service` end-to-end produces a valid unit** + +```bash +# create a dummy config file +tmp=$(mktemp /tmp/logitechmouse-XXXX.toml) +echo '[device]' > "$tmp" +echo 'path = "/dev/input/event0"' >> "$tmp" + +python -m logitechmouse install-service --config "$tmp" +cat ~/.config/systemd/user/logitechmouse.service +rm "$tmp" +``` + +Expected: file exists, contains `ExecStart=...logitechmouse listen --config /tmp/logitechmouse-...toml`. + +- [ ] **Step 3: Push branch and open PR** + +```bash +git push origin main +``` + +Or if working on a feature branch: +```bash +git checkout -b phase6-qt-sigterm +git push -u origin phase6-qt-sigterm +gh pr create --title "feat: Phase 6 — Qt-path SIGTERM + systemd unit" \ + --body "Fixes SIGTERM on the Qt listen path via set_wakeup_fd + QSocketNotifier. Adds install-service command." +``` From 235a3975efd760f0db4e2d8443d1f14189b908ea Mon Sep 17 00:00:00 2001 From: ChristopherLandaverde Date: Mon, 27 Apr 2026 23:05:07 -0400 Subject: [PATCH 3/8] =?UTF-8?q?test(qt=5Fsigterm):=20mechanism=20unit=20te?= =?UTF-8?q?sts=20=E2=80=94=20socketpair+notifier=20and=20set=5Fwakeup=5Ffd?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- tests/test_listen_qt_sigterm.py | 83 +++++++++++++++++++++++++++++++++ 1 file changed, 83 insertions(+) create mode 100644 tests/test_listen_qt_sigterm.py diff --git a/tests/test_listen_qt_sigterm.py b/tests/test_listen_qt_sigterm.py new file mode 100644 index 0000000..36b7ba2 --- /dev/null +++ b/tests/test_listen_qt_sigterm.py @@ -0,0 +1,83 @@ +from __future__ import annotations + +import os +import signal +import socket +import sys + +import pytest + +pytest.importorskip("PyQt6") + +from PyQt6.QtCore import QSocketNotifier, QTimer +from PyQt6.QtWidgets import QApplication + + +def _app() -> QApplication: + return QApplication.instance() or QApplication(sys.argv) + + +def test_socket_notifier_fires_when_written_to(): + """Writing to the write end of a socketpair causes QSocketNotifier to fire + and call app.quit(); app.exec() returns.""" + app = _app() + r, w = socket.socketpair(socket.AF_UNIX, socket.SOCK_STREAM) + r.setblocking(False) + w.setblocking(False) + + fired = [] + notifier = QSocketNotifier(r.fileno(), QSocketNotifier.Type.Read) + + def _slot(): + try: + r.recv(256) + except OSError: + pass + fired.append(True) + app.quit() + + notifier.activated.connect(_slot) + QTimer.singleShot(50, lambda: w.send(b"\x00")) + app.exec() + + notifier.setEnabled(False) + r.close() + w.close() + + assert fired, "QSocketNotifier slot was not called" + + +def test_sigterm_via_set_wakeup_fd_triggers_notifier(): + """SIGTERM delivered to this process writes a byte via set_wakeup_fd, + which fires the QSocketNotifier and causes app.exec() to return.""" + app = _app() + r, w = socket.socketpair(socket.AF_UNIX, socket.SOCK_STREAM) + r.setblocking(False) + w.setblocking(False) + + prev_fd = signal.set_wakeup_fd(w.fileno()) + prev_sigterm = signal.getsignal(signal.SIGTERM) + signal.signal(signal.SIGTERM, lambda s, f: None) + + fired = [] + notifier = QSocketNotifier(r.fileno(), QSocketNotifier.Type.Read) + + def _slot(): + try: + r.recv(256) + except OSError: + pass + fired.append(True) + app.quit() + + notifier.activated.connect(_slot) + QTimer.singleShot(50, lambda: os.kill(os.getpid(), signal.SIGTERM)) + app.exec() + + notifier.setEnabled(False) + signal.set_wakeup_fd(prev_fd) + signal.signal(signal.SIGTERM, prev_sigterm) + r.close() + w.close() + + assert fired, "SIGTERM did not trigger QSocketNotifier via set_wakeup_fd" From 7a1a2a2e35da9b400025e833bb9dcf7c47dd5a5f Mon Sep 17 00:00:00 2001 From: ChristopherLandaverde Date: Mon, 27 Apr 2026 23:27:14 -0400 Subject: [PATCH 4/8] test(qt_sigterm): integration tests for _run_with_qt signal setup (failing) --- tests/test_listen_qt_sigterm.py | 70 +++++++++++++++++++++++++++++++++ 1 file changed, 70 insertions(+) diff --git a/tests/test_listen_qt_sigterm.py b/tests/test_listen_qt_sigterm.py index 36b7ba2..d55c4f8 100644 --- a/tests/test_listen_qt_sigterm.py +++ b/tests/test_listen_qt_sigterm.py @@ -1,9 +1,11 @@ from __future__ import annotations +import argparse import os import signal import socket import sys +from unittest.mock import MagicMock, patch import pytest @@ -12,6 +14,8 @@ from PyQt6.QtCore import QSocketNotifier, QTimer from PyQt6.QtWidgets import QApplication +from logitechmouse.cli import listen as listen_mod + def _app() -> QApplication: return QApplication.instance() or QApplication(sys.argv) @@ -81,3 +85,69 @@ def _slot(): w.close() assert fired, "SIGTERM did not trigger QSocketNotifier via set_wakeup_fd" + + +# --------------------------------------------------------------------------- +# Integration tests — _run_with_qt signal handler setup/restore +# --------------------------------------------------------------------------- + + +@pytest.fixture +def ring_cfg(tmp_path): + cfg = tmp_path / "config.toml" + cfg.write_text( + '[device]\npath = "/dev/input/event99"\n\n' + '[actions.a]\nkind = "command"\ncommand = "true"\n\n' + '[rings.r]\nsegments = [\n' + ' { action = "a", label = "A" },\n' + ' { action = "a", label = "B" },\n' + ' { action = "a", label = "C" },\n' + ']\n\n' + '[bindings.b1]\ntrigger = "BTN_BACK"\ntarget = "ring:r"\n' + ) + return cfg + + +def test_run_with_qt_installs_noop_sigterm_and_restores_it(ring_cfg): + """_run_with_qt must replace the SIGTERM handler with a no-op while + app.exec() runs, then restore the original handler on exit.""" + args = argparse.Namespace(config=ring_cfg, device=None) + fake_dev = MagicMock(path="/dev/input/event99", name="fake") + + original_sigterm = signal.getsignal(signal.SIGTERM) + captured = {} + + def fake_exec(): + captured["sigterm"] = signal.getsignal(signal.SIGTERM) + return 0 + + with patch.object(listen_mod.EvdevBackend, "resolve", return_value=fake_dev), \ + patch.object(listen_mod.EvdevBackend, "read_loop", return_value=iter([])), \ + patch("logitechmouse.cli.listen.try_grab", return_value=None), \ + patch("PyQt6.QtWidgets.QApplication.exec", side_effect=fake_exec), \ + patch("PyQt6.QtCore.QThread.start"), \ + patch("PyQt6.QtCore.QThread.wait"): + listen_mod.run(args) + + assert captured.get("sigterm") is not original_sigterm, \ + "SIGTERM handler must be replaced (no-op) during app.exec()" + assert signal.getsignal(signal.SIGTERM) is original_sigterm, \ + "SIGTERM handler must be restored after _run_with_qt returns" + + +def test_run_with_qt_tears_down_virt_after_exec_returns(ring_cfg): + """virt.close() and device.ungrab() must be called after app.exec() returns.""" + args = argparse.Namespace(config=ring_cfg, device=None) + fake_dev = MagicMock(path="/dev/input/event99", name="fake") + fake_virt = MagicMock() + + with patch.object(listen_mod.EvdevBackend, "resolve", return_value=fake_dev), \ + patch.object(listen_mod.EvdevBackend, "read_loop", return_value=iter([])), \ + patch("logitechmouse.cli.listen.try_grab", return_value=fake_virt), \ + patch("PyQt6.QtWidgets.QApplication.exec", return_value=0), \ + patch("PyQt6.QtCore.QThread.start"), \ + patch("PyQt6.QtCore.QThread.wait"): + listen_mod.run(args) + + fake_virt.close.assert_called_once() + fake_dev.ungrab.assert_called_once() From 21eaa1c74fb941847282077842a8cb1271fd2a51 Mon Sep 17 00:00:00 2001 From: ChristopherLandaverde Date: Mon, 27 Apr 2026 23:29:06 -0400 Subject: [PATCH 5/8] feat(listen): Qt-path SIGTERM via set_wakeup_fd + QSocketNotifier --- src/logitechmouse/cli/listen.py | 44 +++++++++++++++++++++++++++++++++ 1 file changed, 44 insertions(+) diff --git a/src/logitechmouse/cli/listen.py b/src/logitechmouse/cli/listen.py index 7e4f374..2442e54 100644 --- a/src/logitechmouse/cli/listen.py +++ b/src/logitechmouse/cli/listen.py @@ -3,6 +3,7 @@ import argparse import logging import signal +import socket import sys from contextlib import contextmanager from typing import Callable @@ -268,12 +269,55 @@ def _on_finished(rc: int) -> None: bridge.on_event, Qt.ConnectionType.QueuedConnection ) worker.finished.connect(_on_finished, Qt.ConnectionType.QueuedConnection) + from PyQt6.QtCore import QSocketNotifier + + r_sock, w_sock = socket.socketpair(socket.AF_UNIX, socket.SOCK_STREAM) + r_sock.setblocking(False) + w_sock.setblocking(False) + + _wakeup_active = False + prev_wakeup_fd = -1 + prev_sigterm = signal.getsignal(signal.SIGTERM) + prev_sigint = signal.getsignal(signal.SIGINT) + notifier = None + + try: + prev_wakeup_fd = signal.set_wakeup_fd(w_sock.fileno()) + _wakeup_active = True + except ValueError as exc: + logging.warning( + "set_wakeup_fd unavailable: %s; SIGTERM will hard-kill the Qt path", exc + ) + r_sock.close() + w_sock.close() + + if _wakeup_active: + signal.signal(signal.SIGTERM, lambda s, f: None) + signal.signal(signal.SIGINT, lambda s, f: None) + notifier = QSocketNotifier(r_sock.fileno(), QSocketNotifier.Type.Read) + + def _on_signal_wakeup(): + try: + r_sock.recv(256) + except OSError: + pass + app.quit() + + notifier.activated.connect(_on_signal_wakeup) + thread.start() try: app.exec() thread.wait(2000) finally: + if _wakeup_active: + notifier.setEnabled(False) + signal.set_wakeup_fd(prev_wakeup_fd) + signal.signal(signal.SIGTERM, prev_sigterm) + signal.signal(signal.SIGINT, prev_sigint) + r_sock.close() + w_sock.close() if virt is not None: try: virt.close() From 0a7ef62c00d74bbad115fc2867f010023c2221cd Mon Sep 17 00:00:00 2001 From: ChristopherLandaverde Date: Mon, 27 Apr 2026 23:30:29 -0400 Subject: [PATCH 6/8] =?UTF-8?q?test(install=5Fservice):=20all=20cases=20?= =?UTF-8?q?=E2=80=94=20happy=20path,=20errors,=20force,=20daemon-reload?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- tests/test_install_service.py | 135 ++++++++++++++++++++++++++++++++++ 1 file changed, 135 insertions(+) create mode 100644 tests/test_install_service.py diff --git a/tests/test_install_service.py b/tests/test_install_service.py new file mode 100644 index 0000000..ebd5b74 --- /dev/null +++ b/tests/test_install_service.py @@ -0,0 +1,135 @@ +from __future__ import annotations + +import argparse +from pathlib import Path +from unittest.mock import MagicMock, patch + +import pytest + + +def _args(config: Path, force: bool = False) -> argparse.Namespace: + return argparse.Namespace(config=config, force=force) + + +def _run(args, home: Path): + from logitechmouse.cli import install_service as mod + with patch("pathlib.Path.home", return_value=home): + return mod.run(args) + + +def test_happy_path_writes_unit_file(tmp_path): + config = tmp_path / "config.toml" + config.write_text("[device]\npath = '/dev/input/event0'\n") + + with patch("logitechmouse.cli.install_service.shutil.which", + return_value="/usr/local/bin/logitechmouse"), \ + patch("logitechmouse.cli.install_service.subprocess.run", + return_value=MagicMock(returncode=0)): + rc = _run(_args(config), home=tmp_path) + + assert rc == 0 + unit = tmp_path / ".config" / "systemd" / "user" / "logitechmouse.service" + assert unit.exists() + content = unit.read_text() + assert "/usr/local/bin/logitechmouse" in content + assert str(config.resolve()) in content + assert "listen --config" in content + assert "Restart=on-failure" in content + + +def test_happy_path_runs_daemon_reload(tmp_path): + config = tmp_path / "config.toml" + config.write_text("") + + mock_run = MagicMock(return_value=MagicMock(returncode=0)) + with patch("logitechmouse.cli.install_service.shutil.which", + return_value="/usr/bin/logitechmouse"), \ + patch("logitechmouse.cli.install_service.subprocess.run", mock_run): + _run(_args(config), home=tmp_path) + + mock_run.assert_called_once() + cmd = mock_run.call_args[0][0] + assert "daemon-reload" in cmd + + +def test_config_not_found_returns_1_no_unit_written(tmp_path): + args = _args(tmp_path / "missing.toml") + with patch("logitechmouse.cli.install_service.shutil.which", + return_value="/usr/bin/logitechmouse"): + rc = _run(args, home=tmp_path) + + assert rc == 1 + unit = tmp_path / ".config" / "systemd" / "user" / "logitechmouse.service" + assert not unit.exists() + + +def test_binary_not_resolvable_returns_1(tmp_path): + config = tmp_path / "config.toml" + config.write_text("") + with patch("logitechmouse.cli.install_service.shutil.which", return_value=None), \ + patch("sys.argv", []): + rc = _run(_args(config), home=tmp_path) + assert rc == 1 + + +def test_existing_unit_without_force_returns_1_and_does_not_overwrite(tmp_path): + config = tmp_path / "config.toml" + config.write_text("") + unit_dir = tmp_path / ".config" / "systemd" / "user" + unit_dir.mkdir(parents=True) + unit = unit_dir / "logitechmouse.service" + unit.write_text("old content") + + with patch("logitechmouse.cli.install_service.shutil.which", + return_value="/usr/bin/logitechmouse"): + rc = _run(_args(config, force=False), home=tmp_path) + + assert rc == 1 + assert unit.read_text() == "old content" + + +def test_force_overwrites_existing_unit(tmp_path): + config = tmp_path / "config.toml" + config.write_text("") + unit_dir = tmp_path / ".config" / "systemd" / "user" + unit_dir.mkdir(parents=True) + (unit_dir / "logitechmouse.service").write_text("old content") + + with patch("logitechmouse.cli.install_service.shutil.which", + return_value="/usr/bin/logitechmouse"), \ + patch("logitechmouse.cli.install_service.subprocess.run", + return_value=MagicMock(returncode=0)): + rc = _run(_args(config, force=True), home=tmp_path) + + assert rc == 0 + content = (unit_dir / "logitechmouse.service").read_text() + assert content != "old content" + assert "listen --config" in content + + +def test_daemon_reload_failure_warns_but_returns_0(tmp_path): + config = tmp_path / "config.toml" + config.write_text("") + + with patch("logitechmouse.cli.install_service.shutil.which", + return_value="/usr/bin/logitechmouse"), \ + patch("logitechmouse.cli.install_service.subprocess.run", + return_value=MagicMock(returncode=1)): + rc = _run(_args(config), home=tmp_path) + + assert rc == 0 + + +def test_fallback_to_argv0_when_which_returns_none(tmp_path): + config = tmp_path / "config.toml" + config.write_text("") + + with patch("logitechmouse.cli.install_service.shutil.which", return_value=None), \ + patch("sys.argv", ["/home/user/.local/bin/logitechmouse", "install-service"]), \ + patch("logitechmouse.cli.install_service.subprocess.run", + return_value=MagicMock(returncode=0)): + rc = _run(_args(config), home=tmp_path) + + assert rc == 0 + unit = tmp_path / ".config" / "systemd" / "user" / "logitechmouse.service" + assert "/home/user/.local/bin/logitechmouse" in unit.read_text() From d891612e29a99194c12898fb7c373d17c8598edd Mon Sep 17 00:00:00 2001 From: ChristopherLandaverde Date: Mon, 27 Apr 2026 23:39:08 -0400 Subject: [PATCH 7/8] =?UTF-8?q?feat(cli):=20install-service=20command=20?= =?UTF-8?q?=E2=80=94=20writes=20systemd=20user=20unit=20file?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/logitechmouse/cli/install_service.py | 65 ++++++++++++++++++++++++ 1 file changed, 65 insertions(+) create mode 100644 src/logitechmouse/cli/install_service.py diff --git a/src/logitechmouse/cli/install_service.py b/src/logitechmouse/cli/install_service.py new file mode 100644 index 0000000..8031b51 --- /dev/null +++ b/src/logitechmouse/cli/install_service.py @@ -0,0 +1,65 @@ +from __future__ import annotations + +import argparse +import logging +import shutil +import subprocess +import sys +from pathlib import Path + +_UNIT_TEMPLATE = """\n[Unit] +Description=Logitech Mouse button remapper +After=graphical-session.target + +[Service] +ExecStart={exec_start} listen --config {config_path} +Restart=on-failure +RestartSec=3 + +[Install] +WantedBy=default.target +""" + + +def run(args: argparse.Namespace) -> int: + exec_start = shutil.which("logitechmouse") or (sys.argv[0] if sys.argv else None) + if not exec_start: + logging.error("cannot resolve logitechmouse binary; check your PATH") + return 1 + + config_path = Path(args.config).resolve() + if not config_path.exists(): + logging.error("config file not found: %s", config_path) + return 1 + + unit_dir = Path.home() / ".config" / "systemd" / "user" + unit_path = unit_dir / "logitechmouse.service" + + if unit_path.exists() and not getattr(args, "force", False): + logging.error( + "service file already exists: %s — use --force to overwrite", + unit_path, + ) + return 1 + + unit_dir.mkdir(parents=True, exist_ok=True) + unit_path.write_text( + _UNIT_TEMPLATE.format(exec_start=exec_start, config_path=config_path) + ) + logging.info("wrote %s", unit_path) + + result = subprocess.run( + ["systemctl", "--user", "daemon-reload"], + capture_output=True, + ) + if result.returncode != 0: + logging.warning( + "systemctl --user daemon-reload failed (rc=%d); " + "run it manually once a systemd user session is available", + result.returncode, + ) + + print("Service file written to " + str(unit_path)) + print("Enable and start with:") + print(" systemctl --user enable --now logitechmouse") + return 0 From 77c1d565d4d73d07be0d81453985f897d4a180d7 Mon Sep 17 00:00:00 2001 From: ChristopherLandaverde Date: Mon, 27 Apr 2026 23:40:06 -0400 Subject: [PATCH 8/8] feat(main): wire install-service subcommand --- src/logitechmouse/main.py | 18 ++++++++++++++++++ 1 file changed, 18 insertions(+) diff --git a/src/logitechmouse/main.py b/src/logitechmouse/main.py index 9abc188..6c6cfa8 100644 --- a/src/logitechmouse/main.py +++ b/src/logitechmouse/main.py @@ -44,6 +44,22 @@ def build_parser() -> argparse.ArgumentParser: p_run.add_argument("name", help="Action name as defined in config") p_run.add_argument("--dry-run", action="store_true", help="Do not spawn the command") + p_install = sub.add_parser( + "install-service", + help="Write a systemd user unit file for logitechmouse", + ) + p_install.add_argument( + "--config", + type=Path, + required=True, + help="Path to config TOML (baked into ExecStart)", + ) + p_install.add_argument( + "--force", + action="store_true", + help="Overwrite an existing unit file", + ) + return parser @@ -62,6 +78,8 @@ def main() -> int: from .cli.check_config import run as run_cmd elif args.command == "run": from .cli.run import run as run_cmd + elif args.command == "install-service": + from .cli.install_service import run as run_cmd else: parser.error(f"unknown command: {args.command}")