From c8897f8de51e4ad857f2dd24cb8ddcaa65e47e58 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Antonio=20V=C3=A1zquez=20Blanco?= Date: Fri, 16 Oct 2026 12:40:51 +0200 Subject: [PATCH] Improve the support for keyboard events Add key_down, key_up and key_tap to send USB HID keyboard reports, with KeyCode and KeyModifier enums for the common keys and modifier bits. key_tap releases the keys even when it is cancelled. The keyboard and mouse now share one WebSocket and one check for which protocol the device speaks. Firmware older than non-Pro 2.3.2 or Pro 1.2.6 only takes one key at a time, and modifiers only with a key; any other report raises NanoKVMNotSupportedError. While the WebSocket is open, the client sends a heartbeat every 10 seconds, as the web UI does, so non-Pro 2.5.0 and newer don't close it when it's idle. --- docs/api.md | 32 ++++ nanokvm/components/hid.py | 69 +++++++++ nanokvm/components/mouse.py | 43 +----- nanokvm/components/session.py | 75 +++++++-- nanokvm/models/common.py | 136 ++++++++++++++++- tests/test_keyboard.py | 277 ++++++++++++++++++++++++++++++++++ tests/test_mouse.py | 6 +- 7 files changed, 587 insertions(+), 51 deletions(-) create mode 100644 tests/test_keyboard.py diff --git a/docs/api.md b/docs/api.md index b2ca85e..6a7a5b5 100644 --- a/docs/api.md +++ b/docs/api.md @@ -71,6 +71,7 @@ other scheme (`http` or `https`), within about half a minute. | Methods | Requires | |---|---| | `paste_text` | all | +| `key_down`, `key_up`, `key_tap` | all, see below | | `mouse_move_abs`, `mouse_move_rel`, `mouse_down`, `mouse_up`, `mouse_click`, `mouse_scroll` | all | | `get_hid_mode`, `set_hid_mode` | non-Pro 2.2.5 / Pro | | `reset_hid` | non-Pro / Pro 1.1.6 | @@ -81,6 +82,37 @@ other scheme (`http` or `https`), within about half a minute. `paste_text` accepts only the characters in `nanokvm.client.PASTE_CHAR_MAP`. +The key methods send USB HID key codes, such as `KeyCode.F12` or any `int`, +with modifiers such as `KeyModifier.LEFT_CTRL`: + +```python +from nanokvm.models import KeyCode, KeyModifier + +await client.key_tap( + KeyCode.DELETE, modifiers=KeyModifier.LEFT_CTRL | KeyModifier.LEFT_ALT +) +``` + +- `key_down` holds up to six keys and replaces the keys held before. `key_up` + releases them all. `key_tap` presses and releases, and still releases when + it is cancelled. +- Before non-Pro 2.3.2 and Pro 1.2.6, the device holds one key at a time and + applies modifiers only together with a key. Other reports raise + `NanoKVMNotSupportedError`. +- The keyboard and mouse share one WebSocket. While it is open, the client + sends a heartbeat every 10 seconds, as the web UI does, so that non-Pro 2.5.0 + and newer do not close it when idle. + +To catch a short boot-menu window, tap a key repeatedly: + +```python +loop = asyncio.get_running_loop() +end = loop.time() + 8 +while loop.time() < end: + await client.key_tap(KeyCode.F12) + await asyncio.sleep(0.4) +``` + ### Video and streaming | Methods | Requires | diff --git a/nanokvm/components/hid.py b/nanokvm/components/hid.py index c278ff9..df50f11 100644 --- a/nanokvm/components/hid.py +++ b/nanokvm/components/hid.py @@ -2,9 +2,13 @@ from __future__ import annotations +import asyncio +import math + from aiohttp import hdrs from ..compatibility import require_application_version, require_hardware +from ..exceptions import NanoKVMNotSupportedError from ..models.common import ( AddShortcutReq, DeleteShortcutReq, @@ -15,6 +19,8 @@ GetShortcutsRsp, HidMode, HWFamily, + KeyCode, + KeyModifier, MouseJigglerMode, PasteReq, SetHidModeReq, @@ -30,6 +36,9 @@ "[\\]^_`abcdefghijklmnopqrstuvwxyz{|}~" ) +_KEYBOARD_EVENT = 1 +_KEYBOARD_REPORT_KEYS = 6 + class HidController(Controller): """Implement HID operations behind the public client facade.""" @@ -148,3 +157,63 @@ async def paste_text(self, text: str) -> None: "/hid/paste", data=PasteReq(content=text), ) + + async def key_down( + self, *keys: KeyCode | int, modifiers: KeyModifier | int = 0 + ) -> None: + """Hold keys and modifiers down, replacing those held before. + + Args: + keys: Up to six USB HID key codes, such as ``KeyCode.F12``. + modifiers: Modifier keys to hold, such as ``KeyModifier.LEFT_CTRL``. + + Before non-Pro 2.3.2 and Pro 1.2.6, the device holds one key at a time + and applies modifiers only together with a key. + """ + await self._send_keyboard_report(int(modifiers), [int(key) for key in keys]) + + async def key_up(self) -> None: + """Release all keys and modifiers.""" + await self._send_keyboard_report(0, []) + + async def key_tap( + self, + *keys: KeyCode | int, + modifiers: KeyModifier | int = 0, + hold: float = 0.05, + ) -> None: + """Press keys and modifiers, and release them after ``hold`` seconds.""" + if not math.isfinite(hold) or hold < 0: + raise ValueError("hold must be a finite, non-negative number of seconds") + + await self.key_down(*keys, modifiers=modifiers) + try: + await asyncio.sleep(hold) + finally: + # Release even when cancelled, so no key stays held on the host. + await self.key_up() + + async def _send_keyboard_report(self, modifiers: int, keys: list[int]) -> None: + """Send a keyboard report in the wire format of the connected device.""" + if len(keys) > _KEYBOARD_REPORT_KEYS: + raise ValueError("A keyboard report holds at most six keys") + if not all(0 <= value <= 0xFF for value in (modifiers, *keys)): + raise ValueError("Key codes and modifiers must be between 0 and 255") + + session = self._session + if await session.uses_binary_hid_protocol(): + # Boot keyboard report: modifiers, a reserved byte and six key slots. + report = bytes((modifiers, 0, *keys)).ljust(8, bytes(1)) + await session.send_ws( + lambda ws: ws.send_bytes(bytes((_KEYBOARD_EVENT,)) + report) + ) + return + + if len(keys) > 1 or (modifiers and not any(keys)): + raise NanoKVMNotSupportedError( + "Holding several keys, or modifiers without a key, requires the " + "binary keyboard protocol" + ) + # The legacy server combines the four modifier fields into one byte. + message = [_KEYBOARD_EVENT, keys[0] if keys else 0, modifiers, 0, 0, 0] + await session.send_ws(lambda ws: ws.send_json(message)) diff --git a/nanokvm/components/mouse.py b/nanokvm/components/mouse.py index ed25207..c2ecf87 100644 --- a/nanokvm/components/mouse.py +++ b/nanokvm/components/mouse.py @@ -3,33 +3,18 @@ from __future__ import annotations import asyncio -from collections.abc import Awaitable, Callable import logging import math -import aiohttp - from ..exceptions import NanoKVMNotSupportedError from ..models.common import MouseButton from .session import Controller _LOGGER = logging.getLogger(__name__) -_BINARY_MOUSE_MIN_NON_PRO_VERSION = "2.3.2" -_BINARY_MOUSE_MIN_PRO_VERSION = "1.2.6" - class MouseController(Controller): - """Own the mouse protocol and its lazily-created WebSocket transport.""" - - async def _uses_binary_mouse_protocol(self) -> bool: - """Select the mouse wire format supported by the connected device.""" - # Without a token the WebSocket reports the authentication error, so - # the current protocol stays the default when detection is impossible. - return await self._session.application_version_at_least( - non_pro=_BINARY_MOUSE_MIN_NON_PRO_VERSION, - pro=_BINARY_MOUSE_MIN_PRO_VERSION, - ) + """Own the mouse protocol on the session's shared WebSocket.""" @staticmethod def _require_finite(*values: float) -> None: @@ -79,21 +64,9 @@ def _report_for_current_mode(self, *, wheel: int = 0) -> bytes: return self._absolute_report(wheel) return self._relative_report(wheel=wheel) - async def _send_ws( - self, - send: Callable[[aiohttp.ClientWebSocketResponse], Awaitable[None]], - ) -> None: - """Send one mouse message and invalidate the connection on failure.""" - ws = await self._session.get_ws() - try: - await send(ws) - except (aiohttp.ClientConnectionError, ConnectionError, RuntimeError): - await self._session.invalidate_ws(ws) - raise - async def _send_mouse_report(self, report: bytes) -> None: """Send a binary NanoKVM mouse event and HID report.""" - await self._send_ws(lambda ws: ws.send_bytes(bytes((2,)) + report)) + await self._session.send_ws(lambda ws: ws.send_bytes(bytes((2,)) + report)) async def _send_legacy_mouse_event( self, event_type: int, button_state: int, x: float, y: float @@ -114,7 +87,7 @@ async def _send_legacy_mouse_event( message = [2, event_type, button_state, x_value, y_value] _LOGGER.debug("Sending legacy mouse event: %s", message) - await self._send_ws(lambda ws: ws.send_json(message)) + await self._session.send_ws(lambda ws: ws.send_json(message)) async def mouse_move_abs(self, x: float, y: float) -> None: """Move mouse to absolute position. @@ -125,7 +98,7 @@ async def mouse_move_abs(self, x: float, y: float) -> None: """ self._require_finite(x, y) session = self._session - if not await self._uses_binary_mouse_protocol(): + if not await self._session.uses_binary_hid_protocol(): await self._send_legacy_mouse_event(2, 0, x, y) return @@ -142,7 +115,7 @@ async def mouse_move_rel(self, dx: float, dy: float) -> None: """ self._require_finite(dx, dy) session = self._session - if not await self._uses_binary_mouse_protocol(): + if not await self._session.uses_binary_hid_protocol(): await self._send_legacy_mouse_event(3, 0, dx, dy) return @@ -161,7 +134,7 @@ async def mouse_down(self, button: MouseButton = MouseButton.LEFT) -> None: """ session = self._session - if not await self._uses_binary_mouse_protocol(): + if not await self._session.uses_binary_hid_protocol(): if button in (MouseButton.BACK, MouseButton.FORWARD): raise NanoKVMNotSupportedError( "Back and Forward mouse buttons require the binary mouse protocol" @@ -192,7 +165,7 @@ async def mouse_up(self) -> None: The report releases all currently held buttons. """ session = self._session - if not await self._uses_binary_mouse_protocol(): + if not await self._session.uses_binary_hid_protocol(): await self._send_legacy_mouse_event(0, 0, 0.0, 0.0) return @@ -239,7 +212,7 @@ async def mouse_scroll(self, dx: float, dy: float) -> None: negative scrolls down. """ self._require_finite(dx, dy) - if not await self._uses_binary_mouse_protocol(): + if not await self._session.uses_binary_hid_protocol(): await self._send_legacy_mouse_event(4, 0, dx, dy) return diff --git a/nanokvm/components/session.py b/nanokvm/components/session.py index 035f4e4..0eec7ca 100644 --- a/nanokvm/components/session.py +++ b/nanokvm/components/session.py @@ -3,7 +3,7 @@ from __future__ import annotations import asyncio -from collections.abc import AsyncGenerator +from collections.abc import AsyncGenerator, Awaitable, Callable import contextlib from http.cookies import Morsel, SimpleCookie import json @@ -14,6 +14,7 @@ from typing import Any, Literal, NoReturn, TypeVar, overload from aiohttp import ( + ClientConnectionError, ClientResponse, ClientResponseError, ClientSession, @@ -59,6 +60,13 @@ _SESSION_COOKIE_NAME = "nano-kvm-token" _NOT_STARTED = "NanoKVMClient is not started; use 'async with NanoKVMClient(...)'" _CURRENT_PASSWORD_MIN_NON_PRO_VERSION = (2, 5, 1) +_BINARY_HID_MIN_NON_PRO_VERSION = "2.3.2" +_BINARY_HID_MIN_PRO_VERSION = "1.2.6" +_WS_SEND_ERRORS = (ClientConnectionError, ConnectionError, RuntimeError) +# Non-Pro 2.5.0 and newer close a WebSocket that sends nothing for 90 seconds. +# The web UI sends this heartbeat every 10 seconds. +_WS_HEARTBEAT = bytes((0,)) +_WS_HEARTBEAT_INTERVAL = 10.0 class SessionController: @@ -85,6 +93,7 @@ def __init__( self._request_timeout = request_timeout self._image_transfer_lock = asyncio.Lock() self._ws: ClientWebSocketResponse | None = None + self._ws_heartbeat: asyncio.Task[None] | None = None self._ws_session: ClientSession | None = None self._ws_lock = asyncio.Lock() self._mouse_buttons = 0 @@ -444,8 +453,7 @@ async def clear_local_session( self.forget_versions() self._mouse_buttons = 0 self.clear_session_cookies() - ws = self._ws - self._ws = None + ws = self._detach_ws() # Detach state atomically, then close only the old transport. Network I/O # must not block a new login or close its replacement WebSocket. if ws is not None and not ws.closed: @@ -643,8 +651,7 @@ async def get_hardware(self) -> GetHardwareRsp: async def close_ws(self) -> None: """Close and forget the current WebSocket connection.""" async with self._ws_lock: - ws = self._ws - self._ws = None + ws = self._detach_ws() if ws is not None and not ws.closed: await ws.close() @@ -652,30 +659,74 @@ async def invalidate_ws(self, ws: ClientWebSocketResponse) -> None: """Forget a failed WebSocket without closing a replacement connection.""" async with self._ws_lock: if self._ws is ws: - self._ws = None + self._detach_ws() if not ws.closed: await ws.close() + def _detach_ws(self) -> ClientWebSocketResponse | None: + """Forget the shared WebSocket and stop its heartbeat; hold _ws_lock.""" + ws = self._ws + self._ws = None + if self._ws_heartbeat is not None: + self._ws_heartbeat.cancel() + self._ws_heartbeat = None + return ws + + async def _send_heartbeats(self, ws: ClientWebSocketResponse) -> None: + """Keep the shared WebSocket open while no input is sent.""" + while True: + await asyncio.sleep(_WS_HEARTBEAT_INTERVAL) + try: + await ws.send_bytes(_WS_HEARTBEAT) + except _WS_SEND_ERRORS: + # The next input finds the broken connection and replaces it. + return + async def get_ws(self) -> ClientWebSocketResponse: - """Get or create WebSocket connection for mouse events.""" + """Get or create the WebSocket shared by keyboard and mouse events.""" generation = self._session_generation try: async with self._ws_lock: if self._ws is not None and not self._ws.closed: return self._ws - if self._ws is not None: - await self._ws.close() - self._ws = None + stale_ws = self._detach_ws() + if stale_ws is not None: + await stale_ws.close() generation = self._session_generation - self._ws = await self._connect_ws("ws") - return self._ws + ws = await self._connect_ws("ws") + self._ws = ws + self._ws_heartbeat = asyncio.create_task(self._send_heartbeats(ws)) + return ws except ClientResponseError as err: # _clear_local_session acquires _ws_lock, so do this after the # lock above has been released to avoid a self-deadlock. await self._raise_ws_error(err, "ws", generation) + async def send_ws( + self, send: Callable[[ClientWebSocketResponse], Awaitable[None]] + ) -> None: + """Send one input message on the shared WebSocket. + + A failed send forgets the connection, so the next one reconnects. + """ + ws = await self.get_ws() + try: + await send(ws) + except _WS_SEND_ERRORS: + await self.invalidate_ws(ws) + raise + + async def uses_binary_hid_protocol(self) -> bool: + """Select the keyboard and mouse wire format of the connected device.""" + # Without a token the WebSocket reports the authentication error, so + # the current protocol stays the default when detection is impossible. + return await self.application_version_at_least( + non_pro=_BINARY_HID_MIN_NON_PRO_VERSION, + pro=_BINARY_HID_MIN_PRO_VERSION, + ) + async def open_ws(self, path: str, **kwargs: Any) -> ClientWebSocketResponse: """Open a new, unshared WebSocket on ``path``, relative to the API base. diff --git a/nanokvm/models/common.py b/nanokvm/models/common.py index 71d8f01..92052d1 100644 --- a/nanokvm/models/common.py +++ b/nanokvm/models/common.py @@ -3,7 +3,7 @@ from __future__ import annotations from datetime import datetime -from enum import IntEnum, StrEnum +from enum import IntEnum, IntFlag, StrEnum from typing import Any, Generic, Self, TypeVar from pydantic import ( @@ -59,6 +59,8 @@ "ImageEnabledRsp", "ImageTransferProgress", "IsPasswordUpdatedRsp", + "KeyCode", + "KeyModifier", "LoginReq", "LoginRsp", "LoginTailscaleRsp", @@ -193,6 +195,138 @@ class MouseButton(IntEnum): FORWARD = 16 +class KeyCode(IntEnum): + """USB HID keyboard usage codes, named after the browser key codes. + + Any other usage code can be passed as an ``int``. + """ + + KEY_A = 0x04 + KEY_B = 0x05 + KEY_C = 0x06 + KEY_D = 0x07 + KEY_E = 0x08 + KEY_F = 0x09 + KEY_G = 0x0A + KEY_H = 0x0B + KEY_I = 0x0C + KEY_J = 0x0D + KEY_K = 0x0E + KEY_L = 0x0F + KEY_M = 0x10 + KEY_N = 0x11 + KEY_O = 0x12 + KEY_P = 0x13 + KEY_Q = 0x14 + KEY_R = 0x15 + KEY_S = 0x16 + KEY_T = 0x17 + KEY_U = 0x18 + KEY_V = 0x19 + KEY_W = 0x1A + KEY_X = 0x1B + KEY_Y = 0x1C + KEY_Z = 0x1D + DIGIT_1 = 0x1E + DIGIT_2 = 0x1F + DIGIT_3 = 0x20 + DIGIT_4 = 0x21 + DIGIT_5 = 0x22 + DIGIT_6 = 0x23 + DIGIT_7 = 0x24 + DIGIT_8 = 0x25 + DIGIT_9 = 0x26 + DIGIT_0 = 0x27 + ENTER = 0x28 + ESCAPE = 0x29 + BACKSPACE = 0x2A + TAB = 0x2B + SPACE = 0x2C + MINUS = 0x2D + EQUAL = 0x2E + BRACKET_LEFT = 0x2F + BRACKET_RIGHT = 0x30 + BACKSLASH = 0x31 + SEMICOLON = 0x33 + QUOTE = 0x34 + BACKQUOTE = 0x35 + COMMA = 0x36 + PERIOD = 0x37 + SLASH = 0x38 + CAPS_LOCK = 0x39 + F1 = 0x3A + F2 = 0x3B + F3 = 0x3C + F4 = 0x3D + F5 = 0x3E + F6 = 0x3F + F7 = 0x40 + F8 = 0x41 + F9 = 0x42 + F10 = 0x43 + F11 = 0x44 + F12 = 0x45 + PRINT_SCREEN = 0x46 + SCROLL_LOCK = 0x47 + PAUSE = 0x48 + INSERT = 0x49 + HOME = 0x4A + PAGE_UP = 0x4B + DELETE = 0x4C + END = 0x4D + PAGE_DOWN = 0x4E + ARROW_RIGHT = 0x4F + ARROW_LEFT = 0x50 + ARROW_DOWN = 0x51 + ARROW_UP = 0x52 + NUM_LOCK = 0x53 + NUMPAD_DIVIDE = 0x54 + NUMPAD_MULTIPLY = 0x55 + NUMPAD_SUBTRACT = 0x56 + NUMPAD_ADD = 0x57 + NUMPAD_ENTER = 0x58 + NUMPAD_1 = 0x59 + NUMPAD_2 = 0x5A + NUMPAD_3 = 0x5B + NUMPAD_4 = 0x5C + NUMPAD_5 = 0x5D + NUMPAD_6 = 0x5E + NUMPAD_7 = 0x5F + NUMPAD_8 = 0x60 + NUMPAD_9 = 0x61 + NUMPAD_0 = 0x62 + NUMPAD_DECIMAL = 0x63 + INTL_BACKSLASH = 0x64 + CONTEXT_MENU = 0x65 + POWER = 0x66 + NUMPAD_EQUAL = 0x67 + F13 = 0x68 + F14 = 0x69 + F15 = 0x6A + F16 = 0x6B + F17 = 0x6C + F18 = 0x6D + F19 = 0x6E + F20 = 0x6F + F21 = 0x70 + F22 = 0x71 + F23 = 0x72 + F24 = 0x73 + + +class KeyModifier(IntFlag): + """Modifier key bits of a keyboard report.""" + + LEFT_CTRL = 0x01 + LEFT_SHIFT = 0x02 + LEFT_ALT = 0x04 + LEFT_META = 0x08 + RIGHT_CTRL = 0x10 + RIGHT_SHIFT = 0x20 + RIGHT_ALT = 0x40 + RIGHT_META = 0x80 + + class _CaseInsensitiveStrEnum(StrEnum): """String enum with case-insensitive parsing.""" diff --git a/tests/test_keyboard.py b/tests/test_keyboard.py new file mode 100644 index 0000000..a62ef51 --- /dev/null +++ b/tests/test_keyboard.py @@ -0,0 +1,277 @@ +"""Tests for keyboard input and the shared input WebSocket.""" + +import asyncio +from collections.abc import AsyncGenerator +from typing import Any +from unittest.mock import AsyncMock, call, patch + +import pytest + +from nanokvm.client import NanoKVMClient, NanoKVMNotSupportedError +from nanokvm.models import HWVersion, KeyCode, KeyModifier + +from .common import BASE_URL, mark_detected + +_RELEASE = bytes([1, 0, 0, 0, 0, 0, 0, 0, 0]) +_HEARTBEAT = b"\x00" + + +def _mock_ws() -> AsyncMock: + ws = AsyncMock() + ws.closed = False + return ws + + +@pytest.fixture +async def ws_client( + request: pytest.FixtureRequest, +) -> AsyncGenerator[tuple[NanoKVMClient, AsyncMock], Any]: + """A client on the hardware and version given by ``request.param``.""" + hw_version, version = getattr(request, "param", (HWVersion.PCIE, "2.3.2")) + mock_ws = _mock_ws() + + with patch( + "aiohttp.ClientSession.ws_connect", new_callable=AsyncMock, return_value=mock_ws + ): + async with NanoKVMClient(BASE_URL, token="test-token") as client: + mark_detected(client, hw_version, version) + yield client, mock_ws + + +def _sent_reports(mock_ws: AsyncMock) -> list[bytes]: + return [ + call.args[0] + for call in mock_ws.send_bytes.call_args_list + if call.args[0] != _HEARTBEAT + ] + + +def _sent_events(mock_ws: AsyncMock) -> list[list[int]]: + return [call.args[0] for call in mock_ws.send_json.call_args_list] + + +async def test_key_down_and_up_send_boot_keyboard_reports( + ws_client: tuple[NanoKVMClient, AsyncMock], +) -> None: + """Binary reports carry the modifier byte, a reserved byte and six keys.""" + client, mock_ws = ws_client + + await client.key_down( + KeyCode.DELETE, modifiers=KeyModifier.LEFT_CTRL | KeyModifier.LEFT_ALT + ) + await client.key_down(*range(4, 10), modifiers=KeyModifier.RIGHT_META) + await client.key_down(modifiers=KeyModifier.LEFT_SHIFT) + await client.key_up() + + assert _sent_reports(mock_ws) == [ + bytes([1, 0x05, 0, 0x4C, 0, 0, 0, 0, 0]), + bytes([1, 0x80, 0, 4, 5, 6, 7, 8, 9]), + bytes([1, 0x02, 0, 0, 0, 0, 0, 0, 0]), + _RELEASE, + ] + mock_ws.send_json.assert_not_called() + + +@pytest.mark.parametrize( + ("keys", "modifiers"), + [(tuple(range(4, 11)), 0), ((256,), 0), ((-1,), 0), ((), 256)], +) +async def test_invalid_reports_are_rejected_before_sending( + ws_client: tuple[NanoKVMClient, AsyncMock], + keys: tuple[int, ...], + modifiers: int, +) -> None: + """More than six keys or values outside a byte never reach the device.""" + client, mock_ws = ws_client + + with pytest.raises(ValueError): + await client.key_down(*keys, modifiers=modifiers) + + mock_ws.send_bytes.assert_not_called() + + +async def test_key_tap_presses_and_releases( + ws_client: tuple[NanoKVMClient, AsyncMock], +) -> None: + """A tap holds the keys for ``hold`` seconds, then releases them.""" + client, mock_ws = ws_client + + # Spy on the real sleep: a mock that never yields would spin the heartbeat. + with patch( + "nanokvm.components.hid.asyncio.sleep", side_effect=asyncio.sleep + ) as sleep: + await client.key_tap(KeyCode.F12, modifiers=KeyModifier.LEFT_SHIFT, hold=0.01) + + assert call(0.01) in sleep.await_args_list + assert _sent_reports(mock_ws) == [ + bytes([1, 0x02, 0, 0x45, 0, 0, 0, 0, 0]), + _RELEASE, + ] + + +async def test_cancelled_key_tap_still_releases( + ws_client: tuple[NanoKVMClient, AsyncMock], +) -> None: + """Cancelling a tap while the key is held does not leave it held.""" + client, mock_ws = ws_client + + tap = asyncio.create_task(client.key_tap(KeyCode.ENTER, hold=60)) + async with asyncio.timeout(2): + while not _sent_reports(mock_ws): + await asyncio.sleep(0) + tap.cancel() + with pytest.raises(asyncio.CancelledError): + await tap + + assert _sent_reports(mock_ws) == [ + bytes([1, 0, 0, 0x28, 0, 0, 0, 0, 0]), + _RELEASE, + ] + + +@pytest.mark.parametrize("hold", [-0.1, float("nan"), float("inf")]) +async def test_key_tap_rejects_invalid_hold( + ws_client: tuple[NanoKVMClient, AsyncMock], hold: float +) -> None: + """An invalid hold time is rejected before any key is pressed.""" + client, mock_ws = ws_client + + with pytest.raises(ValueError, match="hold"): + await client.key_tap(KeyCode.KEY_A, hold=hold) + + mock_ws.send_bytes.assert_not_called() + + +@pytest.mark.parametrize("ws_client", [(HWVersion.PCIE, "2.3.1")], indirect=True) +async def test_legacy_firmware_receives_json_key_events( + ws_client: tuple[NanoKVMClient, AsyncMock], +) -> None: + """Before 2.3.2, keys are JSON events whose modifier fields are ORed.""" + client, mock_ws = ws_client + + await client.key_down( + KeyCode.KEY_A, modifiers=KeyModifier.LEFT_CTRL | KeyModifier.RIGHT_SHIFT + ) + await client.key_up() + + assert _sent_events(mock_ws) == [[1, 0x04, 0x21, 0, 0, 0], [1, 0, 0, 0, 0, 0]] + assert not _sent_reports(mock_ws) + + +@pytest.mark.parametrize("ws_client", [(HWVersion.PCIE, "2.3.1")], indirect=True) +@pytest.mark.parametrize( + ("keys", "modifiers"), + [((KeyCode.KEY_A, KeyCode.KEY_B), 0), ((), KeyModifier.LEFT_SHIFT)], +) +async def test_legacy_firmware_rejects_reports_it_cannot_express( + ws_client: tuple[NanoKVMClient, AsyncMock], + keys: tuple[KeyCode, ...], + modifiers: KeyModifier, +) -> None: + """The legacy format has one key slot and drops modifiers without a key.""" + client, mock_ws = ws_client + + with pytest.raises(NanoKVMNotSupportedError, match="binary keyboard protocol"): + await client.key_down(*keys, modifiers=modifiers) + + mock_ws.send_json.assert_not_called() + + +@pytest.mark.parametrize( + ("ws_client", "binary"), + [ + ((HWVersion.PCIE, "2.3.2"), True), + ((HWVersion.BETA, "2.3.1"), False), + ((HWVersion.PRO, "1.2.6"), True), + ((HWVersion.PRO, "1.2.5"), False), + ], + indirect=["ws_client"], +) +async def test_keyboard_protocol_follows_firmware_version( + ws_client: tuple[NanoKVMClient, AsyncMock], binary: bool +) -> None: + """Keyboard and mouse switch to binary reports at the same versions.""" + client, mock_ws = ws_client + + await client.key_up() + + assert bool(_sent_reports(mock_ws)) is binary + assert bool(_sent_events(mock_ws)) is not binary + + +async def test_shared_websocket_sends_heartbeats_until_closed( + ws_client: tuple[NanoKVMClient, AsyncMock], +) -> None: + """An idle input WebSocket sends heartbeats, which stop when it closes.""" + client, mock_ws = ws_client + + with patch("nanokvm.components.session._WS_HEARTBEAT_INTERVAL", 0): + await client.key_up() + heartbeat = client._session._ws_heartbeat + assert heartbeat is not None + async with asyncio.timeout(2): + while mock_ws.send_bytes.await_args_list[-1].args[0] != _HEARTBEAT: + await asyncio.sleep(0) + + await client._session.close_ws() + + assert client._session._ws_heartbeat is None + with pytest.raises(asyncio.CancelledError): + await heartbeat + + +async def test_logout_stops_the_heartbeat( + ws_client: tuple[NanoKVMClient, AsyncMock], +) -> None: + """Clearing the session stops the heartbeat with the WebSocket.""" + client, _ = ws_client + await client.key_up() + heartbeat = client._session._ws_heartbeat + assert heartbeat is not None + + await client._session.clear_local_session() + + assert client._session._ws_heartbeat is None + with pytest.raises(asyncio.CancelledError): + await heartbeat + + +async def test_failed_heartbeat_ends_quietly( + ws_client: tuple[NanoKVMClient, AsyncMock], +) -> None: + """A broken connection ends the heartbeat; the next input reconnects.""" + client, mock_ws = ws_client + await client.key_up() + mock_ws.send_bytes.side_effect = ConnectionResetError() + + with patch("nanokvm.components.session._WS_HEARTBEAT_INTERVAL", 0): + heartbeat = asyncio.create_task(client._session._send_heartbeats(mock_ws)) + async with asyncio.timeout(2): + await heartbeat + + assert heartbeat.result() is None + + +async def test_replacing_a_closed_websocket_restarts_the_heartbeat( + ws_client: tuple[NanoKVMClient, AsyncMock], +) -> None: + """A WebSocket closed by the device is replaced with a new heartbeat.""" + client, mock_ws = ws_client + await client.key_up() + first_heartbeat = client._session._ws_heartbeat + mock_ws.closed = True + second_ws = _mock_ws() + + with patch( + "aiohttp.ClientSession.ws_connect", + new_callable=AsyncMock, + return_value=second_ws, + ): + await client.key_up() + + assert client._session._ws is second_ws + assert client._session._ws_heartbeat not in (None, first_heartbeat) + assert _sent_reports(second_ws) == [_RELEASE] + assert first_heartbeat is not None + with pytest.raises(asyncio.CancelledError): + await first_heartbeat diff --git a/tests/test_mouse.py b/tests/test_mouse.py index a17a650..caf97d9 100644 --- a/tests/test_mouse.py +++ b/tests/test_mouse.py @@ -332,8 +332,8 @@ async def test_legacy_mouse_send_reconnects_after_transport_failure() -> None: first_ws.close.assert_awaited_once() -async def test_mouse_send_helper_invalidates_failed_websocket() -> None: - """The shared mouse transport helper invalidates the failed connection.""" +async def test_input_send_helper_invalidates_failed_websocket() -> None: + """The shared input transport helper invalidates the failed connection.""" ws = AsyncMock() ws.closed = False @@ -345,7 +345,7 @@ async def send(current_ws: Any) -> None: raise ConnectionResetError() with pytest.raises(ConnectionResetError): - await client._send_ws(send) + await client._session.send_ws(send) assert client._session._ws is None ws.close.assert_awaited_once()