From dda75b6c4a69206bef6e7459f6ee9dd1e9f2f6ae Mon Sep 17 00:00:00 2001 From: Raphael Chicon Date: Sun, 17 May 2026 18:49:05 -0300 Subject: [PATCH] Add application version gating --- nanokvm/client.py | 144 +++++++++++++++++++++++++++++++++++++++++++ tests/test_client.py | 106 ++++++++++++++++++++++++++++++- 2 files changed, 249 insertions(+), 1 deletion(-) diff --git a/nanokvm/client.py b/nanokvm/client.py index f713bdf..8cb8338 100644 --- a/nanokvm/client.py +++ b/nanokvm/client.py @@ -11,6 +11,7 @@ import logging from os import PathLike from pathlib import Path +import re import ssl from typing import Any, TypeVar, overload @@ -183,6 +184,29 @@ class NanoKVMNotSupportedError(NanoKVMError): F = TypeVar("F", bound=Callable[..., Coroutine[Any, Any, Any]]) +_VERSION_RE = re.compile(r"^v?(\d+(?:\.\d+)*)$") + + +def _parse_version(version: str) -> tuple[int, ...] | None: + """Parse simple semantic app versions; return None for custom/dev builds.""" + match = _VERSION_RE.fullmatch(version.strip()) + if match is None: + return None + return tuple(int(part) for part in match.group(1).split(".")) + + +def _version_at_least(version: str, minimum: str) -> bool: + """Return True when version is unknown or at least the requested version.""" + parsed_version = _parse_version(version) + parsed_minimum = _parse_version(minimum) + if parsed_version is None or parsed_minimum is None: + return True + + length = max(len(parsed_version), len(parsed_minimum)) + normalized_version = parsed_version + (0,) * (length - len(parsed_version)) + normalized_minimum = parsed_minimum + (0,) * (length - len(parsed_minimum)) + return normalized_version >= normalized_minimum + def require_hardware(*versions: HWVersion) -> Callable[[F], F]: """Decorator that restricts a method to specific hardware versions.""" @@ -208,6 +232,48 @@ async def wrapper(self: NanoKVMClient, *args: Any, **kwargs: Any) -> Any: return decorator +def require_application_version( + *, + non_pro: str | None = None, + pro: str | None = None, +) -> Callable[[F], F]: + """Decorator that restricts a method to minimum application versions.""" + + def decorator(func: F) -> F: + @functools.wraps(func) + async def wrapper(self: NanoKVMClient, *args: Any, **kwargs: Any) -> Any: + if self._hw_version is None: + if self._token is None: + return await func(self, *args, **kwargs) + await self.detect_hardware() + + minimum = pro if self._hw_version == HWVersion.PRO else non_pro + if minimum is None: + return await func(self, *args, **kwargs) + + if self._application_version is None: + if self._token is None: + return await func(self, *args, **kwargs) + await self.detect_versions() + + if self._application_version is not None and not _version_at_least( + self._application_version, minimum + ): + hardware_family = ( + "Pro" if self._hw_version == HWVersion.PRO else "non-Pro" + ) + raise NanoKVMNotSupportedError( + f"{func.__name__} requires {hardware_family} application " + f"version >= {minimum} (detected: {self._application_version})" + ) + + return await func(self, *args, **kwargs) + + return wrapper # type: ignore[return-value] + + return decorator + + class NanoKVMClient: """Async API client for the NanoKVM.""" @@ -256,6 +322,8 @@ def __init__( self._use_password_obfuscation = use_password_obfuscation self._ssl_config: ssl.SSLContext | Fingerprint | bool | None = None self._hw_version: HWVersion | None = None + self._application_version: str | None = None + self._image_version: str | None = None def _create_ssl_context(self) -> ssl.SSLContext | Fingerprint | bool: """ @@ -301,12 +369,33 @@ def hw_version(self) -> HWVersion | None: """The detected hardware version. None if not yet detected.""" return self._hw_version + @property + def application_version(self) -> str | None: + """The detected application version. None if not yet detected.""" + return self._application_version + + @property + def image_version(self) -> str | None: + """The detected image version. None if not yet detected.""" + return self._image_version + async def detect_hardware(self) -> None: """Detect and store the hardware version.""" hw = await self.get_hardware() self._hw_version = hw.version _LOGGER.info("Detected hardware: %s", hw.version) + async def detect_versions(self) -> None: + """Detect and store image and application versions.""" + info = await self.get_info() + self._application_version = info.application + self._image_version = info.image + _LOGGER.info( + "Detected versions: application=%s image=%s", + info.application, + info.image, + ) + async def __aenter__(self) -> NanoKVMClient: """Async context manager entry.""" if self._session is None and not self._external_session_provided: @@ -655,6 +744,7 @@ async def get_hardware(self) -> GetHardwareRsp: response_model=GetHardwareRsp, ) + @require_application_version(non_pro="2.2.6") async def get_hostname(self) -> GetHostnameRsp: """Get the configured hostname.""" return await self._api_request_json( @@ -663,6 +753,7 @@ async def get_hostname(self) -> GetHostnameRsp: response_model=GetHostnameRsp, ) + @require_application_version(non_pro="2.2.6") async def set_hostname(self, hostname: str) -> None: """Set the device hostname (applies after reboot).""" await self._api_request_json( @@ -736,6 +827,7 @@ async def disable_ssh(self) -> None: """Disable SSH server.""" await self._api_request_json(hdrs.METH_POST, "/vm/ssh/disable") + @require_application_version(non_pro="2.2.2") async def get_mdns_state(self) -> GetMdnsStateRsp: """Get mDNS enabled state.""" return await self._api_request_json( @@ -744,10 +836,12 @@ async def get_mdns_state(self) -> GetMdnsStateRsp: response_model=GetMdnsStateRsp, ) + @require_application_version(non_pro="2.2.2") async def enable_mdns(self) -> None: """Enable mDNS.""" await self._api_request_json(hdrs.METH_POST, "/vm/mdns/enable") + @require_application_version(non_pro="2.2.2") async def disable_mdns(self) -> None: """Disable mDNS.""" await self._api_request_json(hdrs.METH_POST, "/vm/mdns/disable") @@ -792,6 +886,7 @@ async def update_virtual_device( ), ) + @require_application_version(non_pro="2.2.6") async def get_mouse_jiggler_state(self) -> GetMouseJigglerRsp: """Get the mouse jiggler state.""" return await self._api_request_json( @@ -800,6 +895,7 @@ async def get_mouse_jiggler_state(self) -> GetMouseJigglerRsp: response_model=GetMouseJigglerRsp, ) + @require_application_version(non_pro="2.2.6") async def set_mouse_jiggler_state( self, enabled: bool, mode: MouseJigglerMode ) -> None: @@ -814,6 +910,7 @@ async def set_mouse_jiggler_state( data=SetMouseJigglerReq(enabled=enabled, mode=mode), ) + @require_application_version(non_pro="2.2.6") async def get_web_title(self) -> GetWebTitleRsp: """Get the web page title.""" return await self._api_request_json( @@ -822,6 +919,7 @@ async def get_web_title(self) -> GetWebTitleRsp: response_model=GetWebTitleRsp, ) + @require_application_version(non_pro="2.2.6") async def set_web_title(self, title: str) -> None: """Set the web page title.""" await self._api_request_json( @@ -830,6 +928,7 @@ async def set_web_title(self, title: str) -> None: data=SetWebTitleReq(title=title), ) + @require_application_version(non_pro="2.2.2") async def reboot_system(self) -> None: """Reboot the KVM device.""" await self._api_request_json(hdrs.METH_POST, "/vm/system/reboot") @@ -851,6 +950,7 @@ async def set_screen(self, setting: ScreenSettingType, value: int) -> None: ) @require_hardware(HWVersion.ALPHA, HWVersion.BETA, HWVersion.PCIE) + @require_application_version(non_pro="2.2.6") async def get_swap_size(self) -> int: """Get Swap size.""" rsp = await self._api_request_json( @@ -861,6 +961,7 @@ async def get_swap_size(self) -> int: return rsp.size @require_hardware(HWVersion.ALPHA, HWVersion.BETA, HWVersion.PCIE) + @require_application_version(non_pro="2.2.6") async def set_swap_size(self, size_mb: int) -> None: """Set the Swap size.""" await self._api_request_json( @@ -924,11 +1025,13 @@ async def reset_hdmi(self) -> None: await self._api_request_json(hdrs.METH_POST, "/vm/hdmi/reset") @require_hardware(HWVersion.ALPHA, HWVersion.BETA, HWVersion.PCIE) + @require_application_version(non_pro="2.2.8") async def enable_hdmi(self) -> None: """Enable the HDMI connection.""" await self._api_request_json(hdrs.METH_POST, "/vm/hdmi/enable") @require_hardware(HWVersion.ALPHA, HWVersion.BETA, HWVersion.PCIE) + @require_application_version(non_pro="2.2.8") async def disable_hdmi(self) -> None: """Disable the HDMI connection.""" await self._api_request_json(hdrs.METH_POST, "/vm/hdmi/disable") @@ -936,6 +1039,7 @@ async def disable_hdmi(self) -> None: # ── VM (Pro only) ────────────────────────────────────────────────── @require_hardware(HWVersion.PRO) + @require_application_version(pro="1.2.7") async def refresh_virtual_device(self, device: str) -> None: """Refresh a virtual device (e.g. emmc).""" await self._api_request_json( @@ -945,6 +1049,7 @@ async def refresh_virtual_device(self, device: str) -> None: ) @require_hardware(HWVersion.PRO) + @require_application_version(pro="1.1.6") async def get_lcd_time_format(self) -> GetLcdTimeFormatRsp: """Get the LCD time format.""" return await self._api_request_json( @@ -954,6 +1059,7 @@ async def get_lcd_time_format(self) -> GetLcdTimeFormatRsp: ) @require_hardware(HWVersion.PRO) + @require_application_version(pro="1.1.6") async def set_lcd_time_format(self, fmt: LcdTimeFormat | str) -> None: """Set the LCD time format (12h/24h).""" await self._api_request_json( @@ -1017,6 +1123,7 @@ async def switch_edid(self, edid: EdidValue) -> None: ) @require_hardware(HWVersion.PRO) + @require_application_version(pro="1.2.3") async def get_custom_edid_list(self) -> GetCustomEdidListRsp: """Get custom EDID list.""" return await self._api_request_json( @@ -1026,6 +1133,7 @@ async def get_custom_edid_list(self) -> GetCustomEdidListRsp: ) @require_hardware(HWVersion.PRO) + @require_application_version(pro="1.2.3") async def upload_edid(self, file_path: str | PathLike[str]) -> UploadEdidRsp: """Upload a custom EDID.""" return await self._upload_file( @@ -1035,6 +1143,7 @@ async def upload_edid(self, file_path: str | PathLike[str]) -> UploadEdidRsp: ) @require_hardware(HWVersion.PRO) + @require_application_version(pro="1.2.3") async def delete_edid(self, edid: str) -> None: """Delete a custom EDID.""" await self._api_request_json( @@ -1044,6 +1153,7 @@ async def delete_edid(self, edid: str) -> None: ) @require_hardware(HWVersion.PRO) + @require_application_version(pro="1.2.7") async def get_low_power(self) -> GetLowPowerRsp: """Get low power status.""" return await self._api_request_json( @@ -1053,6 +1163,7 @@ async def get_low_power(self) -> GetLowPowerRsp: ) @require_hardware(HWVersion.PRO) + @require_application_version(pro="1.2.7") async def set_low_power(self, enable: bool) -> None: """Set low power mode.""" await self._api_request_json( @@ -1137,6 +1248,7 @@ async def set_timezone(self, timezone: str) -> None: ) @require_hardware(HWVersion.PRO) + @require_application_version(pro="1.1.6") async def get_time_status(self) -> GetTimeStatusRsp: """Get time synchronization status.""" return await self._api_request_json( @@ -1146,11 +1258,13 @@ async def get_time_status(self) -> GetTimeStatusRsp: ) @require_hardware(HWVersion.PRO) + @require_application_version(pro="1.1.6") async def sync_time(self) -> None: """Synchronize time.""" await self._api_request_json(hdrs.METH_POST, "/vm/time/sync") @require_hardware(HWVersion.PRO) + @require_application_version(pro="1.2.10") async def get_menubar_config(self) -> GetMenuBarConfigRsp: """Get menu bar configuration.""" return await self._api_request_json( @@ -1160,6 +1274,7 @@ async def get_menubar_config(self) -> GetMenuBarConfigRsp: ) @require_hardware(HWVersion.PRO) + @require_application_version(pro="1.2.10") async def set_menubar_config(self, disabled_items: list[str]) -> None: """Set menu bar configuration.""" await self._api_request_json( @@ -1170,6 +1285,7 @@ async def set_menubar_config(self, disabled_items: list[str]) -> None: # ── HID ───────────────────────────────────────────────────────────── + @require_application_version(non_pro="2.2.5") async def get_hid_mode(self) -> GetHidModeRsp: """Get the current HID mode.""" return await self._api_request_json( @@ -1178,6 +1294,7 @@ async def get_hid_mode(self) -> GetHidModeRsp: response_model=GetHidModeRsp, ) + @require_application_version(non_pro="2.3.2", pro="1.2.8") async def get_shortcuts(self) -> GetShortcutsRsp: """Get configured custom HID shortcuts.""" return await self._api_request_json( @@ -1186,6 +1303,7 @@ async def get_shortcuts(self) -> GetShortcutsRsp: response_model=GetShortcutsRsp, ) + @require_application_version(non_pro="2.3.2", pro="1.2.8") async def add_shortcut(self, keys: list[ShortcutKey]) -> None: """Add a custom HID shortcut.""" await self._api_request_json( @@ -1194,6 +1312,7 @@ async def add_shortcut(self, keys: list[ShortcutKey]) -> None: data=AddShortcutReq(keys=keys), ) + @require_application_version(non_pro="2.3.2", pro="1.2.8") async def delete_shortcut(self, shortcut_id: str) -> None: """Delete a custom HID shortcut.""" await self._api_request_json( @@ -1202,6 +1321,7 @@ async def delete_shortcut(self, shortcut_id: str) -> None: data=DeleteShortcutReq(id=shortcut_id), ) + @require_application_version(non_pro="2.3.4", pro="1.2.12") async def get_leader_key(self) -> GetLeaderKeyRsp: """Get the configured shortcut leader key.""" return await self._api_request_json( @@ -1210,6 +1330,7 @@ async def get_leader_key(self) -> GetLeaderKeyRsp: response_model=GetLeaderKeyRsp, ) + @require_application_version(non_pro="2.3.4", pro="1.2.12") async def set_leader_key(self, key: str = "") -> None: """Set or clear the shortcut leader key.""" await self._api_request_json( @@ -1218,6 +1339,7 @@ async def set_leader_key(self, key: str = "") -> None: data=SetLeaderKeyReq(key=key), ) + @require_application_version(non_pro="2.2.5") async def set_hid_mode(self, mode: HidMode) -> None: """Set the HID mode (requires reboot).""" await self._api_request_json( @@ -1277,6 +1399,7 @@ async def mount_image( ), ) + @require_application_version(non_pro="2.3.0") async def delete_image(self, file: str) -> None: """Delete an image file.""" await self._api_request_json( @@ -1328,6 +1451,7 @@ async def connect_wifi_no_auth( data=ConnectWifiReq(ssid=ssid, password=password), ) + @require_application_version(non_pro="2.3.6", pro="1.2.14") async def verify_ap_login(self, ap_password: str) -> None: """Verify AP-mode setup credentials.""" await self._api_request_json( @@ -1371,6 +1495,7 @@ async def delete_wol_mac(self, mac: str) -> None: data=DeleteMacReq(mac=mac), ) + @require_application_version(non_pro="2.2.6") async def set_wol_mac_name(self, mac: str, name: str) -> None: """Set the display name for a saved Wake-on-LAN MAC entry.""" await self._api_request_json( @@ -1390,6 +1515,7 @@ async def get_tailscale_status(self) -> GetTailscaleStatusRsp: # ── Network (Pro only) ───────────────────────────────────────────── @require_hardware(HWVersion.PRO) + @require_application_version(pro="1.2.3") async def get_static_ip(self) -> GetStaticIPRsp: """Get static IP configuration.""" return await self._api_request_json( @@ -1399,6 +1525,7 @@ async def get_static_ip(self) -> GetStaticIPRsp: ) @require_hardware(HWVersion.PRO) + @require_application_version(pro="1.2.3") async def set_static_ip(self, enabled: bool, ip: str) -> None: """Set static IP configuration.""" await self._api_request_json( @@ -1408,6 +1535,7 @@ async def set_static_ip(self, enabled: bool, ip: str) -> None: ) @require_hardware(HWVersion.PRO) + @require_application_version(pro="1.1.7") async def scan_wifi(self) -> ScanWifiRsp: """Scan for available WiFi networks.""" return await self._api_request_json( @@ -1419,6 +1547,7 @@ async def scan_wifi(self) -> ScanWifiRsp: # ── Stream (Pro only) ────────────────────────────────────────────── @require_hardware(HWVersion.PRO) + @require_application_version(pro="1.2.6") async def set_rate_control_mode(self, mode: RateControlMode) -> None: """Set the stream rate control mode (CBR/VBR).""" await self._api_request_json( @@ -1456,6 +1585,7 @@ async def set_gop(self, gop: int) -> None: ) @require_hardware(HWVersion.PRO) + @require_application_version(pro="1.2.8") async def set_fps(self, fps: int) -> None: """Set the stream FPS.""" await self._api_request_json( @@ -1499,6 +1629,7 @@ async def get_application_version(self) -> GetVersionRsp: response_model=GetVersionRsp, ) + @require_application_version(non_pro="2.2.5") async def get_preview_status(self) -> GetPreviewRsp: """Check if preview updates are enabled.""" return await self._api_request_json( @@ -1507,6 +1638,7 @@ async def get_preview_status(self) -> GetPreviewRsp: response_model=GetPreviewRsp, ) + @require_application_version(non_pro="2.2.5") async def set_preview_state(self, enable: bool) -> None: """Enable or disable preview updates.""" await self._api_request_json( @@ -1521,6 +1653,7 @@ async def update_application(self) -> None: # ── Download ──────────────────────────────────────────────────────── + @require_application_version(non_pro="2.1.6") async def is_image_download_enabled(self) -> ImageEnabledRsp: """Check if the /data partition allows downloads.""" prefix = ( @@ -1532,6 +1665,7 @@ async def is_image_download_enabled(self) -> ImageEnabledRsp: response_model=ImageEnabledRsp, ) + @require_application_version(non_pro="2.1.6") async def get_image_download_status(self) -> StatusImageRsp: """Get the status of an ongoing image download.""" prefix = ( @@ -1543,6 +1677,7 @@ async def get_image_download_status(self) -> StatusImageRsp: response_model=StatusImageRsp, ) + @require_application_version(non_pro="2.1.6") async def download_image(self, url: str) -> StatusImageRsp: """Start downloading an image from a URL.""" prefix = ( @@ -1557,22 +1692,27 @@ async def download_image(self, url: str) -> StatusImageRsp: # ── Extensions (shared) ──────────────────────────────────────────── + @require_application_version(non_pro="2.1.6") async def tailscale_install(self) -> None: """Install Tailscale.""" await self._api_request_json(hdrs.METH_POST, "/extensions/tailscale/install") + @require_application_version(non_pro="2.1.6") async def tailscale_uninstall(self) -> None: """Uninstall Tailscale.""" await self._api_request_json(hdrs.METH_POST, "/extensions/tailscale/uninstall") + @require_application_version(non_pro="2.1.6") async def tailscale_up(self) -> None: """Bring Tailscale up.""" await self._api_request_json(hdrs.METH_POST, "/extensions/tailscale/up") + @require_application_version(non_pro="2.1.6") async def tailscale_down(self) -> None: """Bring Tailscale down.""" await self._api_request_json(hdrs.METH_POST, "/extensions/tailscale/down") + @require_application_version(non_pro="2.1.6") async def tailscale_login(self) -> LoginTailscaleRsp: """Log in to Tailscale.""" return await self._api_request_json( @@ -1581,18 +1721,22 @@ async def tailscale_login(self) -> LoginTailscaleRsp: response_model=LoginTailscaleRsp, ) + @require_application_version(non_pro="2.1.6") async def tailscale_logout(self) -> None: """Log out of Tailscale.""" await self._api_request_json(hdrs.METH_POST, "/extensions/tailscale/logout") + @require_application_version(non_pro="2.1.6") async def tailscale_start(self) -> None: """Start Tailscale service.""" await self._api_request_json(hdrs.METH_POST, "/extensions/tailscale/start") + @require_application_version(non_pro="2.1.6") async def tailscale_stop(self) -> None: """Stop Tailscale service.""" await self._api_request_json(hdrs.METH_POST, "/extensions/tailscale/stop") + @require_application_version(non_pro="2.1.6") async def tailscale_restart(self) -> None: """Restart Tailscale service.""" await self._api_request_json(hdrs.METH_POST, "/extensions/tailscale/restart") diff --git a/tests/test_client.py b/tests/test_client.py index e74df1b..f03ead3 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -6,7 +6,13 @@ import pytest import yarl -from nanokvm.client import NanoKVMApiError, NanoKVMClient, NanoKVMNotSupportedError +from nanokvm.client import ( + NanoKVMApiError, + NanoKVMClient, + NanoKVMNotSupportedError, + _parse_version, + _version_at_least, +) from nanokvm.models import ( ApiResponseCode, DiskType, @@ -21,6 +27,39 @@ ) +def _mark_detected( + client: NanoKVMClient, + hw_version: HWVersion = HWVersion.PCIE, + application_version: str = "9.9.9", +) -> None: + client._hw_version = hw_version + client._application_version = application_version + + +def _info_payload(application: str, image: str = "1.4.0") -> dict[str, object]: + return { + "code": 0, + "msg": "success", + "data": { + "ips": [], + "mdns": "kvm.local", + "image": image, + "application": application, + "deviceKey": "device-key", + }, + } + + +def test_version_parser() -> None: + """Test internal application version parsing and comparison.""" + assert _parse_version("v2.4.1") == (2, 4, 1) + assert _parse_version("dev") is None + assert _version_at_least("1.2.14", "1.2.9") + assert _version_at_least("2.4", "2.4.0") + assert not _version_at_least("2.4.0", "2.4.1") + assert _version_at_least("dev", "2.4.1") + + async def test_get_images_success() -> None: """Test get_images with a successful response.""" async with NanoKVMClient( @@ -84,6 +123,23 @@ async def test_get_images_api_error() -> None: assert "failed to list images" in exc_info.value.msg +async def test_detect_versions_stores_application_and_image() -> None: + """Test detect_versions caches versions from /vm/info.""" + async with NanoKVMClient( + "http://localhost:8888/api/", token="test-token" + ) as client: + with aioresponses() as m: + m.get( + "http://localhost:8888/api/vm/info", + payload=_info_payload(application="2.4.1", image="1.4.0"), + ) + + await client.detect_versions() + + assert client.application_version == "2.4.1" + assert client.image_version == "1.4.0" + + async def test_api_error_allows_endpoint_specific_codes() -> None: """Test endpoint-specific API error codes are surfaced as API errors.""" async with NanoKVMClient( @@ -107,6 +163,8 @@ async def test_none_returning_endpoint_preserves_unknown_api_code() -> None: async with NanoKVMClient( "http://localhost:8888/api/", token="test-token" ) as client: + _mark_detected(client) + with aioresponses() as m: m.post( "http://localhost:8888/api/vm/web-title", @@ -174,6 +232,8 @@ async def test_set_mouse_jiggler_state_noops_when_already_disabled() -> None: async with NanoKVMClient( "http://localhost:8888/api/", token="test-token" ) as client: + _mark_detected(client) + with aioresponses() as m: m.get( "http://localhost:8888/api/vm/mouse-jiggler", @@ -386,6 +446,46 @@ async def test_connect_wifi_no_auth_sends_ap_header() -> None: assert calls[0].kwargs.get("headers", {})["X-AP-Key"] == "setup-secret" +async def test_non_pro_application_version_gate_uses_non_pro_minimum() -> None: + """Test non-Pro devices use the non-Pro minimum application version.""" + async with NanoKVMClient( + "http://localhost:8888/api/", token="test-token" + ) as client: + _mark_detected(client, application_version="2.3.1") + + with aioresponses() as m: + with pytest.raises(NanoKVMNotSupportedError) as exc_info: + await client.get_shortcuts() + + assert "get_shortcuts requires non-Pro application version >= 2.3.2" in str( + exc_info.value + ) + shortcuts_url = yarl.URL("http://localhost:8888/api/hid/shortcuts") + assert ("GET", shortcuts_url) not in m.requests + + +async def test_pro_application_version_gate_uses_pro_minimum() -> None: + """Test Pro devices use the Pro minimum application version.""" + async with NanoKVMClient( + "http://localhost:8888/api/", token="test-token" + ) as client: + _mark_detected( + client, + hw_version=HWVersion.PRO, + application_version="1.2.7", + ) + + with aioresponses() as m: + with pytest.raises(NanoKVMNotSupportedError) as exc_info: + await client.get_shortcuts() + + assert "get_shortcuts requires Pro application version >= 1.2.8" in str( + exc_info.value + ) + shortcuts_url = yarl.URL("http://localhost:8888/api/hid/shortcuts") + assert ("GET", shortcuts_url) not in m.requests + + async def test_upload_script_uses_multipart_form(tmp_path: Path) -> None: """Test script upload uses the shared multipart helper.""" script = tmp_path / "test.sh" @@ -418,6 +518,8 @@ async def test_shortcut_methods_send_expected_payloads() -> None: async with NanoKVMClient( "http://localhost:8888/api/", token="test-token" ) as client: + _mark_detected(client, application_version="2.3.4") + with aioresponses() as m: m.post( "http://localhost:8888/api/hid/shortcut", @@ -461,6 +563,8 @@ async def test_tailscale_login_returns_url() -> None: async with NanoKVMClient( "http://localhost:8888/api/", token="test-token" ) as client: + _mark_detected(client, application_version="2.1.6") + with aioresponses() as m: m.post( "http://localhost:8888/api/extensions/tailscale/login",