From 9b88812f5e507b33b1c49800181033b8f407258a Mon Sep 17 00:00:00 2001 From: Hironobu Sano <7529745+sanohiro@users.noreply.github.com> Date: Sat, 26 Sep 2026 12:58:31 +0900 Subject: [PATCH] fix: capture composed frames and add SSH input test helper --- .github/workflows/ci.yml | 5 + HANDOFF.md | 123 +++++---- src/main.rs | 46 ++-- tests/README.md | 4 + tests/automation/README.md | 243 ++++++++++++++++ tests/automation/bcon-test-input | 34 +++ tests/automation/input_helper.py | 382 ++++++++++++++++++++++++++ tests/automation/probe.py | 82 ++++++ tests/automation/test_input_helper.py | 243 ++++++++++++++++ 9 files changed, 1085 insertions(+), 77 deletions(-) create mode 100644 tests/automation/README.md create mode 100755 tests/automation/bcon-test-input create mode 100644 tests/automation/input_helper.py create mode 100644 tests/automation/probe.py create mode 100644 tests/automation/test_input_helper.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 70c2347..bb8f1f8 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -13,6 +13,11 @@ jobs: steps: - uses: actions/checkout@v4 + - name: Test input helper + run: | + bash -n tests/automation/bcon-test-input + python3 -B -W error::ResourceWarning -m unittest discover -s tests/automation -p 'test_*.py' -v + - uses: dtolnay/rust-toolchain@stable - name: Install dependencies diff --git a/HANDOFF.md b/HANDOFF.md index bbd4d58..094500b 100644 --- a/HANDOFF.md +++ b/HANDOFF.md @@ -1,58 +1,69 @@ # Agent Handoff -Status: Active — review complete; v1.5.0 merge and release pending. - -## Goal - -Merge the reviewed display rotation (#17), NumLock-off keypad navigation (#18), -and related rendering/config fixes. The user authorized merging and necessary -documentation updates. The user also explicitly authorized a release. Use v1.5.0 because rotation adds -a feature. All commit, PR, release, and issue messages must be English. -Leave issue #19 unchanged. -After release, reply to #17 and #18 in English, briefly apologizing for the slow reply -(as requested by the user). Do not send those replies before release. - -## Completed - -- Reviewed all application, documentation, and test-tool changes. -- Fixed review findings: relative-path config watch filtering, grayscale mode - selection and R8-to-RGB atlas upload, and out-of-bounds absolute pointer input - after rotation. Added regression coverage. -- Fixed final-target handling in the interactive pointer check and NumLock - restoration in the keypad check. -- Updated English/Japanese configuration, keybind, and installation docs; - added an Unreleased changelog section and tests/README.md. -- Earlier systemd visual smoke run: 12 screenshots showed correct text, colors, - Sixel, both Kitty transfers, partial update cleanup, and alternate-screen - restoration. Service had zero restarts during that run. This predates the - additional code-review fixes; screenshots do not verify all rotation angles - or VT switching animations. - -## Verification - -- `cargo test --locked`: 65 passed, 1 hardware-dependent seatd test ignored. -- `cargo test --locked --no-default-features`: passed. -- `cargo build --release --locked`: passed after the review fixes. -- Shell/Python parsing and both C test utilities' strict syntax checks passed. -- Relative-config atomic-save regression failed before the fix and passed after. -- User reports overnight stress did not reproduce issue #19; not proof of a fix. - -## Next steps - -- Publish the reviewed branch/PR, verify CI, merge, tag v1.5.0, and verify the - release artifacts. Then reply to issues #17 and #18 in English with a brief - apology for the slow reply. Update this handoff once complete. -- Update this file after merge so future agents do not repeat stale work. - -## Uncommitted files and blockers - -- Changes are being prepared on a dedicated review branch. Check git status and - the PR before resuming. No application restart is needed for GitHub operations. -- The development share is CIFS and reports every file as executable. Preserve - existing Git modes; ignore mode-only differences when reviewing/staging. -- Privileged deployment requires the user's sudo interaction. The user offered - to execute commands themselves; never request their password. - -- CIFS rejected writing loose Git objects during staging. A local review clone - is recorded in /tmp/bcon-review-path; it has the same reviewed file contents - and preserves repository file modes. Use it for publication if needed. +Status: No active task. Input helper and focused screenshot fix verified locally. + +## User direction + +- Prioritize stability and focused bug reviews with reproducible tests. Defer + issues #5/#6; leave #19 unchanged unless instructed otherwise. +- Operate via Codex started over SSH on the Linux test machine, whether Parallels + or physical hardware. Use actual bcon input/captures for feature/app checks. +- All commit, PR, release and issue messages must be English. +- Never push/tag without an explicit instruction. No new publication requested. + +## Completed in this working tree + +- tests/automation/probe.py: read-only environment inventory. +- input_helper.py: sudo bootstrap opens uinput and pins the target bcon PID; + drops UID/GID, then serves bounded keyboard requests on a private Unix socket. + Peer UID checks, release-on-interruption, VT/process guard, and cleanup. +- bcon-test-input shell launcher, symlinked into /home/hiro/.local/bin; works + from any directory. No-argument invocation starts helper via sudo for tty2. +- tests/automation/test_input_helper.py: 18 protocol/lifecycle tests passed. +- src/main.rs: moved PNG capture from the content FBO to after screen composition, + just before buffer swap, within presentation guards. Success flash starts after + capture. No other product source changed. +- cargo test --locked: 65 passed, 1 hardware-dependent seatd test ignored. +- cargo build --release --locked: passed (CIFS cache hard-link warning in tests). +- User installed binary and restarted bcon: PID 20247, start 12:45:23 JST. + Installed SHA256 55308c4bf690505d26d0fbc0ab314022360dcc6ec308f0ebc1bc5b97323f41eb. +- Live helper test confirms new PNGs contain tab bar, divider, terminal cursor, + and 42% progress overlay. Focus, zoom/unzoom and test-tab closure exercised. + PID stayed 20247 and restart count stayed 0. Original login/shell tab restored. + +## Environment and evidence + +- Helper remains running in the user's separate SSH session; query status before + using it and reserve the seat exclusively. It pins the PID, so restart helper + after bcon restarts. User stops it with Ctrl+C. +- User added [paths] screenshot_dir to /etc/bcon/config.toml. Do not append again. + Directory /home/hiro/.local/state/bcon-test/screenshots: hiro:hiro, 0700; + PNG files root:root, 0644, readable by hiro but directory excludes other users. +- Evidence/report: composed-frame-regression.md in that screenshot directory. + The earlier first-helper-check.md describes the pre-fix run. +- Both focus states have tab-background RGBA (31,31,38,255) at (20,920), + divider RGBA (76,76,89,255) at (512,300). A fleeting visual suspicion of a + missing tab bar was disproved by PNG pixel checks; it is not a known defect. + +## Limits and possible future work + +- CI runs the helper protocol/lifecycle tests and shell syntax check. +- No complete application scenario runner or automatic PNG collector yet. +- Hardware pointer planes are outside screenshots. Rotation, sync-update + deferral, VT recovery and host-display flicker were not live-tested here. +- Earlier lines move out of view on split/unzoom and reappear on zoom. Observed + before and after the screenshot fix; separate resize/reflow review candidate. +- Earlier pane-close logs reported SIGHUP/SIGKILL timeouts; not diagnosed here. +- Do not silently expand into rendering refactors or mark these candidates fixed. + +## Working tree / publication + +- Change includes src/main.rs, tests/README.md, tests/automation/, HANDOFF.md, + and CI coverage for the input helper. No release or tag is requested. +- CIFS reports all files executable. Use git -c core.filemode=false status/diff + and preserve tracked modes when staging. The shell launcher needs executable mode. +- Main baseline is v1.5.0, e3751738f4361cea94b081a2ae7b0e3483321d0c. +- Earlier PR #20 merged and v1.5.0 released; artifact checksums verified. +- Issue replies already posted; do not repeat: + https://github.com/sanohiro/bcon/issues/17#issuecomment-5842086768 + https://github.com/sanohiro/bcon/issues/18#issuecomment-5842088706 diff --git a/src/main.rs b/src/main.rs index e86d3c1..0da8c3c 100644 --- a/src/main.rs +++ b/src/main.rs @@ -6280,27 +6280,6 @@ Make sure seatd/logind is running and you're on an active VT." text_renderer.flush(gl, &glyph_atlas, screen_w, screen_h); } - // Bell flash is drawn after FBO blit (outside FBO cache) - - // === Screenshot === - if take_screenshot { - take_screenshot = false; - let user_home = term.user_home_dir(); - if let Err(e) = save_screenshot( - gl, - display_config.width, - display_config.height, - &cfg.paths.screenshot_dir, - user_home.as_deref(), - ) { - log::warn!("Screenshot save failed: {}", e); - } else { - // Flash on success - bell_flash_until = - Some(std::time::Instant::now() + Duration::from_millis(BELL_FLASH_DURATION_MS)); - } - } - // === Pane dividers (drawn in FBO, after all pane content) === // Only draw divider lines between panes (not around outer edges) if !pane_dividers.is_empty() && any_content_dirty { @@ -6957,6 +6936,31 @@ Make sure seatd/logind is running and you're on an active VT." } } + // Capture the composed default framebuffer, including the tab bar, + // terminal cursor and overlays, before swapping it out. Keep the + // request pending while synchronized updates or VT state defer + // presentation. DRM hardware cursor planes are not part of this PNG. + if take_screenshot { + take_screenshot = false; + let user_home = term.user_home_dir(); + if let Err(e) = save_screenshot( + gl, + display_config.width, + display_config.height, + &cfg.paths.screenshot_dir, + user_home.as_deref(), + ) { + log::warn!("Screenshot save failed: {}", e); + } else { + // Acknowledge on the next frame so this capture does not + // include its own success flash. + bell_flash_until = Some( + std::time::Instant::now() + + Duration::from_millis(BELL_FLASH_DURATION_MS), + ); + } + } + egl_context.swap_buffers()?; // Get front buffer and display to DRM diff --git a/tests/README.md b/tests/README.md index c68da00..8cfaa4c 100644 --- a/tests/README.md +++ b/tests/README.md @@ -1,5 +1,9 @@ # Interactive checks +For SSH-driven application testing, see [automation/README.md](automation/README.md). +It includes a read-only environment probe and a temporary sudo-started keyboard +helper for use from a separate SSH session. + Run these inside bcon on a text VT, including a session started through `bcon@tty2.service`. A terminal window or SSH session alone cannot verify bcon's rendering or local keyboard handling. Stop a test before switching VTs when it diff --git a/tests/automation/README.md b/tests/automation/README.md new file mode 100644 index 0000000..a79fe31 --- /dev/null +++ b/tests/automation/README.md @@ -0,0 +1,243 @@ +# Real-VT application testing + +Status: environment probe and temporary keyboard bridge implemented. The full +application runner and screenshot collection are not implemented yet. +This infrastructure lives outside bcon's renderer. The goal is repeatable +application checks using real input and rendered frames on Linux, either in a +Parallels VM or on physical hardware. + +## Operating model + +The user connects to the Linux test machine over SSH and starts Codex there. +Codex runs the controller locally through that SSH session. The tested apps +run inside bcon on a real foreground VT on the same machine. Screenshots and +results are local files that Codex can inspect. No macOS automation or Parallels +API is required for the initial workflow. + +The intended loop is: inspect state, launch a fixture scenario in bcon, send +keys, capture the rendered frame, inspect it, and record a verdict with evidence. +This loop is not implemented yet. SSH PTY output alone is not visual evidence +about bcon. Headless machines without a usable DRM display need separate setup. + +Initial setup may require the user to prepare a reserved bcon session and grant +the input helper appropriate access. Routine runs should then need no manual +key presses or visual judgments from the user. The controller must stop if the +test session is no longer reserved or the expected application has exited. + +## Inventory + +Run from the agent's shell or over SSH, outside the tested terminal: + +```sh +python3 tests/automation/probe.py --tty tty2 > /tmp/bcon-test-environment.json +``` + +Python 3 standard library only. This is read-only: no sudo, key injection, +package installation, service restart, or VT switch. Missing tools and access +denials are reported. It does not certify that the environment is ready. + +## Temporary input helper + +Requires Linux with pidfd support (kernel 5.3+), Python 3.9+, libevdev.so.2 +(Ubuntu/Debian package `libevdev2`), and /dev/uinput. No build step is needed. +Reserve the foreground bcon session exclusively and log in as the test user. +Do not type concurrently or use this against a root shell. + +In a separate SSH session, from the repository root: + +```sh +sudo /usr/bin/python3 -I -B tests/automation/input_helper.py serve --user hiro --tty tty2 +``` + +For a command usable from any directory, install the shell launcher once as +your regular user (from the repository root): + +```sh +mkdir -p ~/.local/bin +ln -s "$(pwd)/tests/automation/bcon-test-input" ~/.local/bin/bcon-test-input +``` + +Ensure `~/.local/bin` is on PATH. The launcher resolves its symlink to locate +the Python helper, so keep this checkout available. On the current development +machine it is already installed for `hiro` and that directory is on PATH. + +```sh +bcon-test-input # sudo prompt, current user, tty2 +bcon-test-input start --tty tty3 # choose another VT +``` + +Client commands need no sudo: `bcon-test-input status`, +`bcon-test-input screenshot`, and `bcon-test-input tap CTRL+SHIFT+ENTER`. +Stop the foreground helper with Ctrl+C. `bcon-test-input --help` lists usage. + +Replace `hiro` with the actual test account. `-I` isolates Python's import path +and environment. The helper looks up the MainPID of `bcon@tty2.service`; for a +manually launched bcon use `--pid PID`. The process must have tty2 as its +controlling terminal and tty2 must be foreground. It does not switch VTs, +log in, restart services, or reload systemd units. + +The helper opens uinput, pins the target process with a pidfd, clears supplementary +groups, and drops to the specified user's UID/GID before creating the virtual +keyboard and listening. Wait for the JSON `"ready": true` message. It waits two +seconds for device hotplug; this does not guarantee bcon received the device. + +In the Codex SSH session, as that same user: + +```sh +python3 -I -B tests/automation/input_helper.py status +python3 -I -B tests/automation/input_helper.py screenshot +``` + +`screenshot` taps PrintScreen. Its successful response confirms event submission, +not PNG creation or display correctness. Check the configured screenshot binding +and directory, then inspect the new image separately. Do not assume a default +key binding if bcon's configuration was customized. + +After establishing the expected screen, examples of deliberate test operations: + +```sh +# Default split-right binding (confirm the actual configuration first). +python3 -I -B tests/automation/input_helper.py tap CTRL+SHIFT+ENTER +# Type without submitting; text requires US layout, CapsLock off and IME off. +python3 -I -B tests/automation/input_helper.py text 'echo bcon-test' +python3 -I -B tests/automation/input_helper.py tap ENTER +``` + +Text accepts up to 256 printable ASCII characters; it does not append Enter. +Key names include A-Z, 0-9, F1-F12, arrows, HOME, END, PAGEUP, PAGEDOWN, +BACKSPACE, DELETE, TAB, ENTER, ESC, PRINTSCREEN, and keypad keys. Modifiers +are CTRL, SHIFT, ALT, and SUPER. Each request is a complete tap, with modifiers +released in reverse order; persistent key-down requests are not supported. + +The socket defaults to `/tmp/bcon-input-UID.sock`, mode 0600. Both ends verify +peer UID using SO_PEERCRED. An alternate path may be set with `--socket PATH` +**before** the subcommand, on both server and client. Existing paths are never +overwritten. If SIGKILL leaves a stale socket, first confirm no helper remains, +then remove only that socket as its owner before restarting. + +The helper latches a disabled state if it observes a VT change or target-process +exit. Start a new helper to rearm. Requests are bounded; disconnects cancel +remaining taps, and I/O errors terminate the helper and destroy its device. +Ctrl+C, SIGTERM, or SSH hangup also clean up. A failed/aborted request may have +partially typed text: inspect the screen rather than retrying automatically. +VT-switch, Ctrl+Alt+Delete, and Alt+PrintScreen chords are rejected. + +Input is seat-wide, not directed to a particular process. Foreground checks +reduce accidental input but cannot make key delivery atomic with VT changes or +concurrent physical keyboard activity. Do not use other sessions on the same +seat during a test. The helper cannot detect whether an app has exited to a shell; +that belongs to the scenario controller. Ctrl+C in the helper SSH session is the +manual stop mechanism. + +To run protocol/lifecycle tests without root or real key injection: + +```sh +python3 -B -m unittest discover -s tests/automation -p 'test_*.py' -v +``` + +These tests use a fake keyboard and real Unix sockets. They do not validate +privilege dropping, libinput hotplug, actual key delivery, or PNG capture. + +## Proposed runner + +1. Reserve a test VM/session and fixture directory. Fix display dimensions, + font, locale, keyboard layout, and bcon configuration. Record actual bcon + and application versions, not just executable paths. +2. Run a user-owned test worker inside bcon. It launches only the chosen scenario + with a timeout, isolated application configuration, and disposable files. + Login remains an explicit setup step; do not type passwords automatically. +3. From the controlling shell, send a bounded sequence of virtual-keyboard + events through uinput. The repository's keypad check is an existing example. + Verify the expected foreground VT and bcon process before input, release all + held keys on cancellation, and stop on unexpected state. VT checks cannot + eliminate races with concurrent human input: reserve the session exclusively. +4. Trigger bcon's screenshot key and wait for a new, complete PNG. Copy it to a + unique scenario artifact immediately. Existing screenshot names have only + second precision; serialize captures to avoid collisions. Confirm configured + screenshot key/path rather than assuming defaults. +5. Check expected application behavior, inspect images, and record evidence. + Save before/after service counters and relevant logs; report timeouts, missing + dependencies, and uncertain visual results separately from passes. +6. End the scenario, verify terminal restoration, and remove the virtual input + device. Do not automatically restart the user's normal bcon service. + +The input helper uses the temporary sudo startup above; do not make /dev/uinput +world-writable or run test applications as root. A dedicated VM is preferable; +a reserved test window in the existing VM is an alternative. + +## Visual verdicts + +### Composed-frame screenshot regression + +On a reserved bcon session running the revised binary: + +1. Capture a single-pane shell. Check visible terminal text and the terminal + cursor (take multiple captures if it is blinking). +2. Open a second tab, split it horizontally, and type different labels in each + pane. Capture after input settles. Expect both labels, the pane divider and + the bottom tab bar with the active tab indicated. +3. Move focus, zoom, then unzoom. Capture each settled state and check the cursor + and tab indicator against the active pane/tab. Treat changed text visibility + on resize as a separate finding, not automatically a screenshot bug. +4. Capture a visible notification/progress overlay when a fixture is available. + Expect the overlay in the PNG. Wait for the previous screenshot success flash + to expire between captures; a screenshot must not contain its own success flash. +5. Close only the test tab and confirm return to the original session. Compare + service PID/restart counters before and after. + +Before the fix, capture ran against the content FBO before screen composition: +tab bars, terminal cursors and later overlays were absent. Capture now belongs +after composition and before buffer swap, within the presentation guard. Requests +remain pending while presentation is deferred by synchronized updates or VT state. +The hardware mouse cursor is a separate DRM plane and is not captured. A PNG +still cannot prove successful scanout, absence of flicker, or VT recovery. + +Actual post-fix display verification requires deploying the revised binary; +unit tests and a successful build alone do not establish this regression pass. + +- Start with explicit assertions: text present, selection moves, borders align, + CJK/emoji do not overlap, scrolling clears old content, and exit restores the + terminal. Store the action sequence and expected result with each image. +- Agent image review can find visible defects but can also miss them. Preserve + images and reasons; use `needs-review` for ambiguous results. +- Establish reviewed baselines before using image differences. Mask dynamic + regions such as clocks/cursors; a matching baseline alone does not prove + correctness, and different terminals need not render identical pixels. +- With the screenshot timing fix, bcon screenshots sample the composed default + framebuffer before buffer swap. They do not independently prove successful + display scanout or absence of flicker. Host-side VM captures or + video are a later complementary check, especially for VT recovery. +- Publish results as "tested version/scenario/environment", not blanket support + for every feature of an application or every GPU. + +## First scenarios + +Begin with one deterministic input/capture round trip. Then add fzf selection +with a fixed list, Neovim editing/search of a disposable Unicode fixture, and +yazi navigation/preview of a fixture directory. Add btop and lazygit after this +works; their dynamic content needs tailored assertions. These are candidates, +not compatibility claims. Do not install/update applications implicitly. + +## Inspection on 2026-09-26 + +- Parallels, Linux aarch64; bcon@tty2 active with tty2 foreground. +- /dev/uinput exists but is not writable by the agent user. +- bcon has PNG screenshots and existing uinput tests; no new product control API + is needed for the initial prototype. +- systemd reports NeedDaemonReload=yes. Confirm the intended unit/binary before + a run; do not silently reload/restart the active session. +- NRestarts=2 at inspection is only a baseline, not evidence of a new crash. +- Found nvim, vim, yazi, btop, fzf, lazygit, and tmux executable paths; their + versions and compatibility have not been verified by this inventory. + +## Live verification on 2026-09-26 + +The composed-frame fix was deployed and tested through the input helper at +1024x936, rotation 0 on the Parallels VM. Captures contain the terminal cursor, +tab bar, divider and an OSC 9;4 progress overlay (42%). Focus changes, zoom/unzoom +and test-tab cleanup were exercised. Service PID stayed 20247 with zero restarts. +PNG pixel checks confirmed the tab background and divider in both focus states. +Local evidence: ~/.local/state/bcon-test/screenshots/composed-frame-regression.md. +Hardware cursor planes, rotation, sync-update deferral and VT recovery are not +covered by this check. Text visibility changes on resize remain a separate +review candidate; this change does not claim to fix reflow. diff --git a/tests/automation/bcon-test-input b/tests/automation/bcon-test-input new file mode 100755 index 0000000..fa8bee0 --- /dev/null +++ b/tests/automation/bcon-test-input @@ -0,0 +1,34 @@ +#!/usr/bin/env bash +# Launch the temporary input helper from any working directory or symlink. +set -euo pipefail + +script_path=$(readlink -f -- "${BASH_SOURCE[0]}") +helper_path="$(dirname -- "$script_path")/input_helper.py" + +if [[ ${1:-} == --help || ${1:-} == -h ]]; then + cat <<'EOF' +Usage: bcon-test-input [start [--tty tty2] [--user USER] [--pid PID]] + bcon-test-input status + bcon-test-input screenshot + bcon-test-input tap CTRL+SHIFT+ENTER + bcon-test-input text 'echo bcon-test' + +With no arguments, start the helper on tty2 for the current user. +Start requests sudo when necessary and stays in the foreground; stop with Ctrl+C. +Other commands run without sudo. Reserve the bcon session before sending input. +For the underlying client's options: bcon-test-input status --help +EOF + exit 0 +fi + +if [[ $# == 0 || ${1:-} == start || ${1:-} == serve ]]; then + if [[ $# != 0 ]]; then shift; fi + target_user=${SUDO_USER:-$(id -un)} + if [[ $EUID == 0 ]]; then + exec /usr/bin/python3 -I -B "$helper_path" serve --user "$target_user" "$@" + else + exec /usr/bin/sudo /usr/bin/python3 -I -B "$helper_path" serve --user "$target_user" "$@" + fi +fi + +exec /usr/bin/python3 -I -B "$helper_path" "$@" diff --git a/tests/automation/input_helper.py b/tests/automation/input_helper.py new file mode 100644 index 0000000..b75a85e --- /dev/null +++ b/tests/automation/input_helper.py @@ -0,0 +1,382 @@ +#!/usr/bin/env python3 +"""Temporary Linux uinput bridge for an exclusively reserved bcon VT.""" + +import argparse +import ctypes +import json +import os +from pathlib import Path +import pwd +import re +import select +import signal +import socket +import stat +import struct +import subprocess +import sys +import time + +MAX_PACKET = 4096 +MODIFIERS = {"LEFTCTRL", "LEFTSHIFT", "LEFTALT", "LEFTMETA"} +ALIASES = {"CTRL": "LEFTCTRL", "SHIFT": "LEFTSHIFT", "ALT": "LEFTALT", + "SUPER": "LEFTMETA", "ESCAPE": "ESC", "RETURN": "ENTER", + "PRINTSCREEN": "SYSRQ", "PAGEUP": "PAGEUP", "PAGEDOWN": "PAGEDOWN"} +KEYS = set("ABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789") | MODIFIERS | { + "ENTER", "ESC", "BACKSPACE", "TAB", "SPACE", "MINUS", "EQUAL", "LEFTBRACE", + "RIGHTBRACE", "BACKSLASH", "SEMICOLON", "APOSTROPHE", "GRAVE", "COMMA", + "DOT", "SLASH", "CAPSLOCK", "NUMLOCK", "SCROLLLOCK", "SYSRQ", "PAUSE", + "HOME", "END", "INSERT", "DELETE", "UP", "DOWN", "LEFT", "RIGHT", + "PAGEUP", "PAGEDOWN", "KPDOT", "KPPLUS", "KPMINUS", "KPASTERISK", + "KPSLASH", "KPENTER", +} | {"F" + str(i) for i in range(1, 13)} | {"KP" + str(i) for i in range(10)} + + +def chord(value): + if not isinstance(value, str): + raise ValueError("keys must be a chord string, e.g. CTRL+SHIFT+ENTER") + names = [ALIASES.get(k.upper(), k.upper()) for k in value.split("+")] + if not 1 <= len(names) <= 5 or len(set(names)) != len(names): + raise ValueError("use 1-5 distinct keys") + if any(k not in KEYS for k in names): + raise ValueError("unknown/unsupported key") + normal = [k for k in names if k not in MODIFIERS] + if len(normal) != 1: + raise ValueError("a tap requires exactly one non-modifier key") + # Do not expose the kernel's VT-switch or magic SysRq chords. + if "LEFTALT" in names and ("SYSRQ" in names or + ("LEFTCTRL" in names and + (re.fullmatch(r"F[0-9]+", normal[0]) or normal[0] == "DELETE"))): + raise ValueError("VT-switch, reboot, and Alt+PrintScreen chords are disabled") + return [k for k in names if k in MODIFIERS] + normal + + +def text_chords(text): + if not isinstance(text, str) or not 1 <= len(text) <= 256: + raise ValueError("text must contain 1-256 printable ASCII characters") + plain = dict(zip("`-=[]\\;',./ ", ["GRAVE", "MINUS", "EQUAL", "LEFTBRACE", + "RIGHTBRACE", "BACKSLASH", "SEMICOLON", "APOSTROPHE", + "COMMA", "DOT", "SLASH", "SPACE"])) + shifted = dict(zip('~_+{}|:"<>?', ["GRAVE", "MINUS", "EQUAL", "LEFTBRACE", + "RIGHTBRACE", "BACKSLASH", "SEMICOLON", "APOSTROPHE", + "COMMA", "DOT", "SLASH"])) + shifted.update(dict(zip("!@#$%^&*()", "1234567890"))) + result = [] + for c in text: + if "a" <= c <= "z" or "0" <= c <= "9": + result.append([c.upper()]) + elif "A" <= c <= "Z": + result.append(["LEFTSHIFT", c]) + elif c in plain: + result.append([plain[c]]) + elif c in shifted: + result.append(["LEFTSHIFT", shifted[c]]) + else: + raise ValueError("text requires a US layout, printable ASCII only; send ENTER separately") + return result + + +class Keyboard: + """libevdev handles native uinput ABI details on both arm64 and x86_64.""" + def __init__(self, fd): + self.lib = ctypes.CDLL("libevdev.so.2") + signatures = { + "libevdev_new": (ctypes.c_void_p, []), + "libevdev_free": (None, [ctypes.c_void_p]), + "libevdev_set_name": (None, [ctypes.c_void_p, ctypes.c_char_p]), + "libevdev_event_code_from_name": (ctypes.c_int, [ctypes.c_uint, ctypes.c_char_p]), + "libevdev_enable_event_code": (ctypes.c_int, [ctypes.c_void_p, ctypes.c_uint, + ctypes.c_uint, ctypes.c_void_p]), + "libevdev_uinput_create_from_device": (ctypes.c_int, [ctypes.c_void_p, ctypes.c_int, + ctypes.POINTER(ctypes.c_void_p)]), + "libevdev_uinput_write_event": (ctypes.c_int, [ctypes.c_void_p, ctypes.c_uint, + ctypes.c_uint, ctypes.c_int]), + "libevdev_uinput_destroy": (None, [ctypes.c_void_p]), + } + for name, (result, args) in signatures.items(): + fn = getattr(self.lib, name) + fn.restype, fn.argtypes = result, args + self.device = ctypes.c_void_p() + self.held = [] + desc = self.lib.libevdev_new() + if not desc: + raise MemoryError("libevdev_new") + try: + self.lib.libevdev_set_name(desc, b"bcon SSH test keyboard") + self.codes = {} + for name in sorted(KEYS): + code = self.lib.libevdev_event_code_from_name(1, ("KEY_" + name).encode()) + if code < 0: + raise ValueError("libevdev does not know " + name) + self.codes[name] = code + self.check(self.lib.libevdev_enable_event_code(desc, 1, code, None)) + self.check(self.lib.libevdev_uinput_create_from_device(desc, fd, + ctypes.byref(self.device))) + finally: + self.lib.libevdev_free(desc) + + @staticmethod + def check(result): + if result < 0: + raise OSError(-result, os.strerror(-result)) + + def event(self, code, value): + self.check(self.lib.libevdev_uinput_write_event(self.device, 1, code, value)) + self.check(self.lib.libevdev_uinput_write_event(self.device, 0, 0, 0)) + + def release(self): + # Attempt every release even if one write fails. Device destruction is + # the final fallback; no held state is allowed across requests. + failed = None + while self.held: + try: + self.event(self.held.pop(), 0) + except OSError as error: + failed = error + if failed: + raise failed + + def tap(self, names, guard): + try: + for name in names: + guard.check() + code = self.codes[name] + self.held.append(code) + self.event(code, 1) + time.sleep(0.03) + finally: + self.release() + time.sleep(0.02) + + def close(self): + try: + self.release() + finally: + self.lib.libevdev_uinput_destroy(self.device) + + +class Guard: + def __init__(self, tty, pidfd, active_path="/sys/class/tty/tty0/active"): + self.tty, self.pidfd, self.active_path = tty, pidfd, active_path + self.reason = None + + def check(self): + if self.reason is None: + try: + if select.select([self.pidfd], [], [], 0)[0]: + self.reason = "target process exited; restart helper" + elif Path(self.active_path).read_text().strip() != self.tty: + self.reason = "foreground VT changed; restart helper" + except OSError as error: + self.reason = "cannot verify foreground VT: " + str(error) + if self.reason: + raise ValueError(self.reason) + + +def peer_uid(connection): + return struct.unpack("3i", connection.getsockopt(socket.SOL_SOCKET, socket.SO_PEERCRED, 12))[1] + + +def handle(connection, uid, keyboard, guard): + if peer_uid(connection) != uid: + return {"ok": False, "error": "peer UID is not authorized"} + packet, _, flags, _ = connection.recvmsg(MAX_PACKET) + if flags & socket.MSG_TRUNC: + raise ValueError("request too large") + request = json.loads(packet) + if not isinstance(request, dict): + raise ValueError("request must be an object") + op = request.get("op") + if op == "status": + try: + guard.check() + except ValueError: + pass + return {"ok": True, "armed": guard.reason is None, "reason": guard.reason, + "tty": guard.tty, "uid": os.getuid()} + if op == "tap": + sequence = [chord(request.get("keys"))] + elif op == "text": + sequence = text_chords(request.get("text")) + else: + raise ValueError("supported operations: status, tap, text") + # Validate the complete request before injecting anything. + for names in sequence: + # Detect an abandoned client between taps and individual presses. + class RequestGuard: + def check(self): + guard.check() + if select.select([connection], [], [], 0)[0]: + raise ValueError("client disconnected or sent additional data") + keyboard.tap(names, RequestGuard()) + return {"ok": True, "taps": len(sequence)} + + +def serve(args): + if os.geteuid() != 0: + raise ValueError("start serve using sudo in a separate SSH session") + account = pwd.getpwnam(args.user) + if account.pw_uid == 0: + raise ValueError("--user must be a non-root account") + if not re.fullmatch(r"tty[1-9][0-9]*", args.tty): + raise ValueError("--tty must name a text VT") + pid = args.pid + if pid is None: + result = subprocess.run(["/usr/bin/systemctl", "show", "bcon@" + args.tty + ".service", + "--property=MainPID", "--value"], check=True, + capture_output=True, text=True, timeout=5, + env={"PATH": "/usr/bin:/bin", "LC_ALL": "C"}) + pid = int(result.stdout.strip()) + if pid <= 1: + raise ValueError("no running target bcon process") + pidfd = os.pidfd_open(pid) + try: + # Pin identity before dropping privileges; root-owned bcon /proc entries + # are not readable by the client user. pidfd prevents PID-reuse confusion. + executable = os.readlink(f"/proc/{pid}/exe").removesuffix(" (deleted)") + fields = Path(f"/proc/{pid}/stat").read_text().rsplit(")", 1)[1].split() + if Path(executable).name != "bcon" or int(fields[4]) != os.stat("/dev/" + args.tty).st_rdev: + raise ValueError("target must be bcon with the specified controlling VT") + guard = Guard(args.tty, pidfd) + guard.check() + fd = os.open("/dev/uinput", os.O_RDWR | os.O_NONBLOCK | os.O_CLOEXEC) + try: + # Retain only the already-open uinput and pidfd. Never serve as root. + os.setgroups([]) + os.setgid(account.pw_gid) + os.setuid(account.pw_uid) + os.umask(0o077) + if os.getuid() != account.pw_uid or os.geteuid() != account.pw_uid: + raise RuntimeError("privilege drop failed") + run_server(args.socket or default_socket(account.pw_uid), account.pw_uid, + fd, guard, pid) + finally: + os.close(fd) + finally: + os.close(pidfd) + + +def run_server(path, uid, fd, guard, pid): + listener = socket.socket(socket.AF_UNIX, socket.SOCK_SEQPACKET) + bound = False + keyboard = None + try: + # No unlink-before-bind: never replace another helper or follow a stale + # socket/symlink. Filesystem work takes place after dropping root. + listener.bind(path) + bound = True + identity = os.lstat(path) + os.chmod(path, 0o600) + listener.listen(1) + listener.settimeout(0.2) + keyboard = Keyboard(fd) + # Let libinput discover the hotplugged keyboard; first real capture is + # still required to demonstrate that the target consumed the events. + time.sleep(2) + guard.check() + print(json.dumps({"ready": True, "socket": path, "uid": os.getuid(), + "tty": guard.tty, "bcon_pid": pid}), flush=True) + while True: + try: + guard.check() + except ValueError: + pass # Latch disarmed state; allow status requests only. + try: + connection, _ = listener.accept() + except socket.timeout: + continue + with connection: + connection.settimeout(2) + fatal = None + try: + response = handle(connection, uid, keyboard, guard) + except (ValueError, UnicodeError, OSError) as error: + response = {"ok": False, "error": str(error), + "note": "input may be partial; do not retry automatically"} + if isinstance(error, OSError) and not isinstance(error, socket.timeout): + fatal = error + try: + connection.sendall(json.dumps(response).encode()) + except OSError: + pass + if fatal is not None: + # A failed release cannot leave the virtual device alive. + # Destroy it and require an explicit new helper session. + raise fatal + finally: + try: + if keyboard is not None: + keyboard.close() + finally: + listener.close() + if bound: + try: + current = os.lstat(path) + if current.st_ino == identity.st_ino and current.st_dev == identity.st_dev: + os.unlink(path) + except FileNotFoundError: + pass + + +def default_socket(uid=None): + return f"/tmp/bcon-input-{os.getuid() if uid is None else uid}.sock" + + +def client(path, request): + info = os.lstat(path) + if not stat.S_ISSOCK(info.st_mode) or info.st_uid != os.getuid() or stat.S_IMODE(info.st_mode) != 0o600: + raise ValueError("expected a mode-0600 socket owned by the current user") + with socket.socket(socket.AF_UNIX, socket.SOCK_SEQPACKET) as connection: + connection.settimeout(20) + connection.connect(path) + if peer_uid(connection) != os.getuid(): + raise ValueError("helper UID does not match client UID") + connection.sendall(json.dumps(request).encode()) + return json.loads(connection.recv(MAX_PACKET)) + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--socket", help="default: /tmp/bcon-input-UID.sock") + commands = parser.add_subparsers(dest="command", required=True) + server = commands.add_parser("serve", help="run in the foreground via sudo") + server.add_argument("--user", required=True) + server.add_argument("--tty", default="tty2") + server.add_argument("--pid", type=int, help="manual bcon PID; default: service MainPID") + commands.add_parser("status") + tap = commands.add_parser("tap") + tap.add_argument("keys", help="e.g. CTRL+SHIFT+ENTER or PRINTSCREEN") + text = commands.add_parser("text", help="US layout ASCII; does not append Enter") + text.add_argument("text") + commands.add_parser("screenshot", help="tap PrintScreen; does not verify PNG creation") + args = parser.parse_args() + if args.command == "serve": + def stop(signum, frame): + raise KeyboardInterrupt + for signum in (signal.SIGINT, signal.SIGTERM, signal.SIGHUP): + signal.signal(signum, stop) + serve(args) + return 0 + request = {"op": args.command} + if args.command == "tap": + chord(args.keys) + request["keys"] = args.keys + elif args.command == "text": + text_chords(args.text) + request["text"] = args.text + elif args.command == "screenshot": + request = {"op": "tap", "keys": "PRINTSCREEN"} + response = client(args.socket or default_socket(), request) + print(json.dumps(response, ensure_ascii=False)) + return 0 if response.get("ok") else 1 + + +if __name__ == "__main__": + try: + sys.exit(main()) + except KeyboardInterrupt: + print("Input helper stopped.", file=sys.stderr) + sys.exit(0) + except (ValueError, KeyError, OSError, subprocess.SubprocessError) as error: + print(f"Error: {error}", file=sys.stderr) + sys.exit(1) diff --git a/tests/automation/probe.py b/tests/automation/probe.py new file mode 100644 index 0000000..5515e18 --- /dev/null +++ b/tests/automation/probe.py @@ -0,0 +1,82 @@ +#!/usr/bin/env python3 +"""Read-only inventory for real-VT automation; never injects input or uses sudo.""" + +import argparse +import datetime +import json +import os +from pathlib import Path +import platform +import re +import shutil +import subprocess + + +def command(argv): + try: + result = subprocess.run(argv, capture_output=True, text=True, timeout=10) + return {"argv": argv, "exit_code": result.returncode, + "stdout": result.stdout.strip(), "stderr": result.stderr.strip()} + except (OSError, subprocess.TimeoutExpired) as error: + return {"argv": argv, "error": str(error)} + + +def read(path): + try: + return Path(path).read_text().strip() + except OSError as error: + return {"error": str(error)} + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--tty", default="tty2") + args = parser.parse_args() + if not re.fullmatch(r"tty[1-9][0-9]*", args.tty): + parser.error("--tty must be a text VT, e.g. tty2") + service = "bcon@" + args.tty + ".service" + state = command(["systemctl", "show", service, + "--property=ActiveState,SubState,MainPID,NRestarts,Result," + "NeedDaemonReload,ExecStart,ExecMainStartTimestamp"]) + props = dict(line.split("=", 1) for line in state.get("stdout", "").splitlines() + if "=" in line) + pid = props.get("MainPID", "0") + executable = None + if pid.isdigit() and pid != "0": + try: + executable = os.readlink("/proc/" + pid + "/exe") + except OSError as error: + executable = {"error": str(error)} + warnings = [] + if props.get("NeedDaemonReload") == "yes": + warnings.append("systemd's loaded unit differs from disk; reconcile before testing") + if props.get("ActiveState") != "active": + warnings.append("target bcon service is not confirmed active") + active_vt = read("/sys/class/tty/tty0/active") + if active_vt != args.tty: + warnings.append("target VT is not confirmed foreground; do not inject keys") + writable = os.access("/dev/uinput", os.W_OK) + if not writable: + warnings.append("current user cannot write /dev/uinput; input setup is required") + report = { + "captured_at": datetime.datetime.now(datetime.timezone.utc).isoformat(), + "platform": platform.platform(), "uid": os.getuid(), + "virtualization": command(["systemd-detect-virt"]), + "target_service": service, "active_vt": active_vt, + "service": state, "running_executable": executable, + "uinput": {"exists": Path("/dev/uinput").exists(), "writable": writable}, + "drm_devices": sorted(str(p) for p in Path("/dev/dri").glob("*")), + "tools": {name: shutil.which(name) for name in + ("bcon", "python3", "cc", "nvim", "vim", "hx", "yazi", + "btop", "fzf", "lazygit", "tmux", "img2sixel")}, + "warnings": warnings, + "notes": ["Inventory only: no application compatibility result.", + "Tool paths do not establish versions or the running bcon binary.", + "NRestarts is a baseline counter, not a diagnosis.", + "No keys sent, screenshots captured, or services changed."], + } + print(json.dumps(report, indent=2, ensure_ascii=False)) + + +if __name__ == "__main__": + main() diff --git a/tests/automation/test_input_helper.py b/tests/automation/test_input_helper.py new file mode 100644 index 0000000..fd7a3ec --- /dev/null +++ b/tests/automation/test_input_helper.py @@ -0,0 +1,243 @@ +"""Protocol/lifecycle tests without root, uinput, or touching the active VT.""" +import importlib.util +import json +import multiprocessing +import os +from pathlib import Path +import socket +import tempfile +import time +import unittest +from unittest.mock import patch + +spec = importlib.util.spec_from_file_location("input_helper", Path(__file__).with_name("input_helper.py")) +h = importlib.util.module_from_spec(spec) +spec.loader.exec_module(h) + + +class FakeGuard: + reason = None + tty = "tty2" + + def check(self): + if self.reason: + raise ValueError(self.reason) + + +class FakeKeyboard: + def __init__(self): + self.calls = [] + + def tap(self, names, guard): + guard.check() + self.calls.append(names) + + def close(self): + pass + + +class ProtocolTests(unittest.TestCase): + def request(self, payload, guard=None, uid=None): + keyboard = FakeKeyboard() + a, b = socket.socketpair(socket.AF_UNIX, socket.SOCK_SEQPACKET) + with a, b: + a.sendall(json.dumps(payload).encode()) + response = h.handle(b, os.getuid() if uid is None else uid, + keyboard, guard or FakeGuard()) + return response, keyboard.calls + + def test_chord_and_modifier_order(self): + response, calls = self.request({"op": "tap", "keys": "enter+ctrl+shift"}) + self.assertEqual(calls, [["LEFTCTRL", "LEFTSHIFT", "ENTER"]]) + self.assertTrue(response["ok"]) + + def test_printscreen(self): + self.assertEqual(h.chord("PRINTSCREEN"), ["SYSRQ"]) + + def test_reject_dangerous_or_ambiguous_chords(self): + for value in ("CTRL", "CTRL+CTRL+A", "A+B", "ALT+PRINTSCREEN", + "CTRL+ALT+F2", "CTRL+ALT+DELETE", "POWER", "", None): + with self.subTest(value=value), self.assertRaises(ValueError): + h.chord(value) + + def test_text_does_not_append_enter(self): + response, calls = self.request({"op": "text", "text": "aA +!"}) + self.assertEqual(calls, [["A"], ["LEFTSHIFT", "A"], ["SPACE"], + ["LEFTSHIFT", "EQUAL"], ["LEFTSHIFT", "1"]]) + self.assertEqual(response["taps"], 5) + + def test_full_printable_ascii(self): + self.assertEqual(len(h.text_chords(''.join(chr(i) for i in range(32, 127)))), 95) + + def test_invalid_suffix_cannot_inject_prefix(self): + keyboard = FakeKeyboard() + a, b = socket.socketpair(socket.AF_UNIX, socket.SOCK_SEQPACKET) + with a, b: + a.sendall(json.dumps({"op": "text", "text": "abc\n"}).encode()) + with self.assertRaises(ValueError): + h.handle(b, os.getuid(), keyboard, FakeGuard()) + self.assertEqual(keyboard.calls, []) + + def test_wrong_uid(self): + response, calls = self.request({"op": "tap", "keys": "ENTER"}, uid=os.getuid() + 1) + self.assertFalse(response["ok"]) + self.assertEqual(calls, []) + + def test_disarmed_rejects_tap_but_status_works(self): + guard = FakeGuard() + guard.reason = "VT changed" + with self.assertRaises(ValueError): + self.request({"op": "tap", "keys": "ENTER"}, guard) + response, calls = self.request({"op": "status"}, guard) + self.assertFalse(response["armed"]) + self.assertEqual(calls, []) + + def test_large_packet(self): + a, b = socket.socketpair(socket.AF_UNIX, socket.SOCK_SEQPACKET) + with a, b: + a.sendall(b"x" * (h.MAX_PACKET + 1)) + with self.assertRaisesRegex(ValueError, "too large"): + h.handle(b, os.getuid(), FakeKeyboard(), FakeGuard()) + + def test_disconnected_client_sends_no_keys(self): + a, b = socket.socketpair(socket.AF_UNIX, socket.SOCK_SEQPACKET) + a.sendall(json.dumps({"op": "tap", "keys": "ENTER"}).encode()) + a.close() + keyboard = FakeKeyboard() + with b, self.assertRaisesRegex(ValueError, "disconnected"): + h.handle(b, os.getuid(), keyboard, FakeGuard()) + self.assertEqual(keyboard.calls, []) + + def test_unknown_operation(self): + with self.assertRaises(ValueError): + self.request({"op": "exec", "command": "anything"}) + + +class LifecycleTests(unittest.TestCase): + def test_real_socket_client_server_and_shutdown_cleanup(self): + context = multiprocessing.get_context("fork") + stop = context.Event() + with tempfile.TemporaryDirectory() as directory: + path = str(Path(directory) / "helper.sock") + + def server(): + guard = FakeGuard() + + def check(): + if stop.is_set(): + raise KeyboardInterrupt + + guard.check = check + try: + with patch.object(h, "Keyboard", return_value=FakeKeyboard()): + h.run_server(path, os.getuid(), -1, guard, os.getpid()) + except KeyboardInterrupt: + pass + + process = context.Process(target=server) + process.start() + try: + deadline = time.monotonic() + 5 + while not Path(path).exists(): + if time.monotonic() >= deadline or not process.is_alive(): + self.fail("server did not bind") + time.sleep(0.01) + self.assertTrue(h.client(path, {"op": "status"})["armed"]) + self.assertEqual(h.client(path, {"op": "tap", "keys": "PRINTSCREEN"}), + {"ok": True, "taps": 1}) + self.assertFalse(h.client(path, {"op": "unknown"})["ok"]) + finally: + stop.set() + process.join(5) + if process.is_alive(): + process.kill() + process.join() + self.assertEqual(process.exitcode, 0) + self.assertFalse(Path(path).exists()) + + def test_client_rejects_world_accessible_socket(self): + with tempfile.TemporaryDirectory() as directory: + path = str(Path(directory) / "helper.sock") + with socket.socket(socket.AF_UNIX, socket.SOCK_SEQPACKET) as server: + server.bind(path) + os.chmod(path, 0o666) + with self.assertRaisesRegex(ValueError, "0600"): + h.client(path, {"op": "status"}) + + def test_process_exit_disarms_guard(self): + pidfd = os.pidfd_open(os.getpid()) + try: + guard = h.Guard("tty2", pidfd) + with patch.object(h.select, "select", return_value=([pidfd], [], [])): + with self.assertRaisesRegex(ValueError, "exited"): + guard.check() + finally: + os.close(pidfd) + + def test_vt_change_latches_even_if_restored(self): + with tempfile.TemporaryDirectory() as directory: + path = Path(directory) / "active" + path.write_text("tty2\n") + pidfd = os.pidfd_open(os.getpid()) + try: + guard = h.Guard("tty2", pidfd, path) + guard.check() + path.write_text("tty1\n") + with self.assertRaises(ValueError): + guard.check() + path.write_text("tty2\n") + with self.assertRaises(ValueError): + guard.check() + finally: + os.close(pidfd) + + def test_keys_released_when_guard_fails_mid_chord(self): + keyboard = object.__new__(h.Keyboard) + keyboard.codes = {"LEFTCTRL": 29, "ENTER": 28} + keyboard.held = [] + events = [] + keyboard.event = lambda code, value: events.append((code, value)) + guard = FakeGuard() + calls = 0 + + def check(): + nonlocal calls + calls += 1 + if calls == 2: + raise ValueError("VT switched") + + guard.check = check + with self.assertRaises(ValueError): + keyboard.tap(["LEFTCTRL", "ENTER"], guard) + self.assertEqual(events, [(29, 1), (29, 0)]) + self.assertEqual(keyboard.held, []) + + def test_release_attempts_remaining_keys_after_io_error(self): + keyboard = object.__new__(h.Keyboard) + keyboard.held = [29, 28] + events = [] + + def event(code, value): + events.append((code, value)) + if code == 28: + raise OSError("write failed") + + keyboard.event = event + with self.assertRaises(OSError): + keyboard.release() + self.assertEqual(events, [(28, 0), (29, 0)]) + self.assertEqual(keyboard.held, []) + + def test_existing_socket_is_never_replaced(self): + with tempfile.TemporaryDirectory() as directory: + path = str(Path(directory) / "helper.sock") + with socket.socket(socket.AF_UNIX, socket.SOCK_SEQPACKET) as existing: + existing.bind(path) + inode = os.stat(path).st_ino + with self.assertRaises(OSError): + h.run_server(path, os.getuid(), -1, FakeGuard(), 1) + self.assertEqual(os.stat(path).st_ino, inode) + + +if __name__ == "__main__": + unittest.main()