From da542a84aa1b2ff76e7ae7285b37c1c3b0ca47d8 Mon Sep 17 00:00:00 2001 From: Raphael Chicon Date: Fri, 24 Apr 2026 07:40:02 -0300 Subject: [PATCH 01/10] fix: respect detected password mode on change --- nanokvm/client.py | 16 +++++++++++++++- 1 file changed, 15 insertions(+), 1 deletion(-) diff --git a/nanokvm/client.py b/nanokvm/client.py index 991e230..f6c7bcf 100644 --- a/nanokvm/client.py +++ b/nanokvm/client.py @@ -445,12 +445,14 @@ async def authenticate(self, username: str, password: str) -> None: _LOGGER.debug("Auto-detecting password mode") try: await self._do_authenticate(username, obfuscate_password(password)) + self._use_password_obfuscation = True _LOGGER.info("Auto-detected obfuscated password mode") except NanoKVMAuthenticationFailure: _LOGGER.debug( "Obfuscated authentication failed, trying plain text password" ) await self._do_authenticate(username, password) + self._use_password_obfuscation = False _LOGGER.info("Auto-detected plain text password mode") await self.detect_hardware() @@ -467,12 +469,24 @@ async def logout(self) -> None: async def change_password(self, username: str, new_password: str) -> None: """Change the KVM password.""" + if self._use_password_obfuscation is None: + raise ValueError( + "Password mode is unknown. Authenticate first or set " + "use_password_obfuscation explicitly before changing the password." + ) + + password_to_send = ( + obfuscate_password(new_password) + if self._use_password_obfuscation + else new_password + ) + await self._api_request_json( hdrs.METH_POST, "/auth/password", data=ChangePasswordReq( username=username, - password=obfuscate_password(new_password), + password=password_to_send, ), ) From 72cb1e5c53bbceb82c9cb588e65f882f713b51e8 Mon Sep 17 00:00:00 2001 From: Raphael Chicon Date: Fri, 24 Apr 2026 07:40:02 -0300 Subject: [PATCH 02/10] fix: unify virtual device status parsing --- nanokvm/client.py | 12 ++-------- nanokvm/models/non_pro.py | 46 +++++++++++++++++++++++++++++++++++---- 2 files changed, 44 insertions(+), 14 deletions(-) diff --git a/nanokvm/client.py b/nanokvm/client.py index f6c7bcf..534b3b1 100644 --- a/nanokvm/client.py +++ b/nanokvm/client.py @@ -96,7 +96,6 @@ GetStaticIPRsp, GetTimeStatusRsp, GetTimeZoneRsp, - GetVirtualDeviceProRsp, RateControlMode, RefreshVirtualDeviceReq, ScanWifiRsp, @@ -604,19 +603,12 @@ async def set_oled_sleep(self, sleep_seconds: int) -> None: data=SetOledReq(sleep=sleep_seconds), ) - async def get_virtual_device_status( - self, - ) -> GetVirtualDeviceRsp | GetVirtualDeviceProRsp: + async def get_virtual_device_status(self) -> GetVirtualDeviceRsp: """Get the status of virtual devices.""" - model = ( - GetVirtualDeviceProRsp - if self._hw_version == HWVersion.PRO - else GetVirtualDeviceRsp - ) return await self._api_request_json( hdrs.METH_GET, "/vm/device/virtual", - response_model=model, + response_model=GetVirtualDeviceRsp, ) async def update_virtual_device( diff --git a/nanokvm/models/non_pro.py b/nanokvm/models/non_pro.py index 9cd16e5..2b79baa 100644 --- a/nanokvm/models/non_pro.py +++ b/nanokvm/models/non_pro.py @@ -3,8 +3,9 @@ from __future__ import annotations from enum import StrEnum +from typing import Any -from pydantic import BaseModel +from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator class ScreenSettingType(StrEnum): @@ -23,9 +24,46 @@ class SetScreenReq(BaseModel): class GetVirtualDeviceRsp(BaseModel): - network: bool - media: bool - disk: bool + model_config = ConfigDict(populate_by_name=True) + + network: bool = False + media: bool | None = None + disk: bool = False + mic: bool | None = None + is_network_enabled: bool | None = Field(default=None, alias="isNetworkEnabled") + is_mic_enabled: bool | None = Field(default=None, alias="isMicEnabled") + mounted_disk: str | None = Field(default=None, alias="mountedDisk") + is_sd_card_exist: bool | None = Field(default=None, alias="isSdCardExist") + is_emmc_exist: bool | None = Field(default=None, alias="isEmmcExist") + + @model_validator(mode="before") + @classmethod + def _normalize_virtual_device_fields(cls, value: Any) -> Any: + if not isinstance(value, dict): + return value + + data = dict(value) + + if "network" not in data and "isNetworkEnabled" in data: + data["network"] = data["isNetworkEnabled"] + + if "mic" not in data and "isMicEnabled" in data: + data["mic"] = data["isMicEnabled"] + + if "media" not in data and "mountedDisk" in data: + data["media"] = bool(data["mountedDisk"]) + + if "disk" not in data and "mountedDisk" in data: + data["disk"] = bool(data["mountedDisk"]) + + return data + + @field_validator("mounted_disk", mode="before") + @classmethod + def _normalize_mounted_disk(cls, value: Any) -> Any: + if value == "": + return None + return value class GetMemoryLimitRsp(BaseModel): From 399d6cf099c4ece25c8bc712c7b6fa6a1db78f61 Mon Sep 17 00:00:00 2001 From: Raphael Chicon Date: Fri, 24 Apr 2026 07:40:02 -0300 Subject: [PATCH 03/10] feat: add richer Pro model typing --- nanokvm/client.py | 9 +++-- nanokvm/models/common.py | 16 ++++++++- nanokvm/models/pro.py | 76 ++++++++++++++++++++++++++++++++++++---- 3 files changed, 91 insertions(+), 10 deletions(-) diff --git a/nanokvm/client.py b/nanokvm/client.py index 534b3b1..58d6075 100644 --- a/nanokvm/client.py +++ b/nanokvm/client.py @@ -84,6 +84,7 @@ from .models.pro import ( DeleteEdidReq, DiskType, + EdidValue, GetCustomEdidListRsp, GetEdidRsp, GetHdmiCaptureRsp, @@ -96,6 +97,7 @@ GetStaticIPRsp, GetTimeStatusRsp, GetTimeZoneRsp, + LcdTimeFormat, RateControlMode, RefreshVirtualDeviceReq, ScanWifiRsp, @@ -112,6 +114,7 @@ SetStreamModeReq, SetStreamQualityReq, SetTimeZoneReq, + StreamMode, SwitchEdidReq, ) from .utils import obfuscate_password @@ -757,7 +760,7 @@ async def get_lcd_time_format(self) -> GetLcdTimeFormatRsp: ) @require_hardware(HWVersion.PRO) - async def set_lcd_time_format(self, fmt: str) -> None: + async def set_lcd_time_format(self, fmt: LcdTimeFormat | str) -> None: """Set the LCD time format (12h/24h).""" await self._api_request_json( hdrs.METH_POST, @@ -811,7 +814,7 @@ async def get_edid(self) -> GetEdidRsp: ) @require_hardware(HWVersion.PRO) - async def switch_edid(self, edid: str) -> None: + async def switch_edid(self, edid: EdidValue) -> None: """Switch EDID.""" await self._api_request_json( hdrs.METH_POST, @@ -1098,7 +1101,7 @@ async def set_rate_control_mode(self, mode: RateControlMode) -> None: ) @require_hardware(HWVersion.PRO) - async def set_stream_mode(self, mode: str) -> None: + async def set_stream_mode(self, mode: StreamMode) -> None: """Set the stream mode.""" await self._api_request_json( hdrs.METH_POST, diff --git a/nanokvm/models/common.py b/nanokvm/models/common.py index 554bf4e..5190631 100644 --- a/nanokvm/models/common.py +++ b/nanokvm/models/common.py @@ -103,6 +103,13 @@ class MouseButton(IntEnum): MIDDLE = 4 +class OledType(StrEnum): + """OLED variants used by NanoKVM Pro.""" + + ATX = "ATX" + DESK = "DESK" + + # Generic Response Wrapper class ApiResponse(BaseModel, Generic[T]): """Generic API response structure.""" @@ -204,9 +211,16 @@ class UpdateVirtualDeviceReq(BaseModel): class GetOLEDRsp(BaseModel): exist: bool - type: str = "" # Pro only + type: OledType | None = None # Pro only sleep: int # Sleep timeout in seconds + @field_validator("type", mode="before") + @classmethod + def _normalize_oled_type(cls, value: Any) -> Any: + if value == "": + return None + return value + class SetOledReq(BaseModel): sleep: int # Sleep timeout in seconds diff --git a/nanokvm/models/pro.py b/nanokvm/models/pro.py index 55faed8..3b80aed 100644 --- a/nanokvm/models/pro.py +++ b/nanokvm/models/pro.py @@ -3,8 +3,9 @@ from __future__ import annotations from enum import StrEnum +from typing import Any, Self -from pydantic import BaseModel, ConfigDict, Field +from pydantic import BaseModel, ConfigDict, Field, field_validator from .common import WiFiInfo @@ -23,6 +24,59 @@ class RateControlMode(StrEnum): VBR = "vbr" +class _CaseInsensitiveStrEnum(StrEnum): + """String enum with case-insensitive parsing.""" + + @classmethod + def _missing_(cls, value: object) -> Self | None: + if isinstance(value, str): + normalized = value.lower() + for member in cls: + if member.value.lower() == normalized: + return member + return None + + +class StreamMode(StrEnum): + """Video stream modes.""" + + MJPEG = "mjpeg" + H264_WEBRTC = "h264-webrtc" + H264_DIRECT = "h264-direct" + H265_WEBRTC = "h265-webrtc" + H265_DIRECT = "h265-direct" + + +class LcdTimeFormat(_CaseInsensitiveStrEnum): + """LCD clock display formats.""" + + TWELVE_HOUR = "12h" + TWENTY_FOUR_HOUR = "24h" + + +class EdidPreset(StrEnum): + """Built-in NanoKVM Pro EDID presets.""" + + UHD_4K_30HZ = "E18-4K30FPS" + UHD_4K_39HZ = "E48-4K39FPS" + QHD_1440P_60HZ = "E56-2K60FPS" + FHD_1080P_60HZ = "E54-1080P60FPS" + WQUXGA_3840X2400_30HZ = "E58-4K16-10" + ULTRAWIDE_3440X1440_60HZ = "E63-Ultrawide" + + +EdidValue = EdidPreset | str + + +def _normalize_edid_value(value: Any) -> Any: + if isinstance(value, str): + try: + return EdidPreset(value) + except ValueError: + return value + return value + + # VM Models class GetVirtualDeviceProRsp(BaseModel): model_config = ConfigDict(populate_by_name=True) @@ -47,11 +101,21 @@ class SetLowPowerReq(BaseModel): class GetEdidRsp(BaseModel): - edid: str + edid: EdidValue + + @field_validator("edid", mode="before") + @classmethod + def _normalize_edid(cls, value: Any) -> Any: + return _normalize_edid_value(value) class SwitchEdidReq(BaseModel): - edid: str + edid: EdidValue + + @field_validator("edid", mode="before") + @classmethod + def _normalize_edid(cls, value: Any) -> Any: + return _normalize_edid_value(value) class GetCustomEdidListRsp(BaseModel): @@ -96,11 +160,11 @@ class GetTimeStatusRsp(BaseModel): class GetLcdTimeFormatRsp(BaseModel): - format: str + format: LcdTimeFormat class SetLcdTimeFormatReq(BaseModel): - format: str + format: LcdTimeFormat class GetMenuBarConfigRsp(BaseModel): @@ -154,7 +218,7 @@ class SetRateControlModeReq(BaseModel): class SetStreamModeReq(BaseModel): - mode: str + mode: StreamMode class SetStreamQualityReq(BaseModel): From b4c9924644f05343cc641d078269757dbc6724d5 Mon Sep 17 00:00:00 2001 From: Raphael Chicon Date: Fri, 24 Apr 2026 07:40:02 -0300 Subject: [PATCH 04/10] fix: normalize Pro compatibility responses --- nanokvm/models/common.py | 27 ++++++++++++++++++++++++--- nanokvm/models/pro.py | 10 ++++++++++ 2 files changed, 34 insertions(+), 3 deletions(-) diff --git a/nanokvm/models/common.py b/nanokvm/models/common.py index 5190631..8599798 100644 --- a/nanokvm/models/common.py +++ b/nanokvm/models/common.py @@ -5,7 +5,7 @@ from enum import IntEnum, StrEnum from typing import Any, Generic, TypeVar -from pydantic import BaseModel, ConfigDict, Field, field_validator +from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator T = TypeVar("T") @@ -159,8 +159,16 @@ class GetInfoRsp(BaseModel): image: str application: str device_key: str = Field(alias="deviceKey") - part_number: str = Field("", alias="pn") # Pro only - arch: str = "" # Pro only + part_number: str | None = Field(default=None, alias="pn") # Pro only + arch: str | None = None # Pro only + + @field_validator("part_number", "arch", mode="before") + @classmethod + def _normalize_optional_strings(cls, value: Any) -> Any: + if isinstance(value, str): + value = value.strip() + return value or None + return value class GetHostnameRsp(BaseModel): @@ -362,6 +370,19 @@ class GetWifiRsp(BaseModel): ssid: str = "" # Non-Pro only wifi: WiFiInfo | None = None # Pro only + @model_validator(mode="before") + @classmethod + def _normalize_wifi_fields(cls, value: Any) -> Any: + if not isinstance(value, dict): + return value + + data = dict(value) + wifi = data.get("wifi") + if data.get("ssid") in (None, "") and isinstance(wifi, dict): + data["ssid"] = wifi.get("ssid", "") + + return data + class ConnectWifiReq(BaseModel): ssid: str diff --git a/nanokvm/models/pro.py b/nanokvm/models/pro.py index 3b80aed..95f9ab1 100644 --- a/nanokvm/models/pro.py +++ b/nanokvm/models/pro.py @@ -123,6 +123,11 @@ class GetCustomEdidListRsp(BaseModel): edid_list: list[str] = Field(default_factory=list, alias="edidList") + @field_validator("edid_list", mode="before") + @classmethod + def _normalize_edid_list(cls, value: Any) -> Any: + return [] if value is None else value + class DeleteEdidReq(BaseModel): edid: str @@ -201,6 +206,11 @@ class ScanWifiRsp(BaseModel): wifi_list: list[WiFiInfo] = Field(default_factory=list, alias="wifiList") + @field_validator("wifi_list", mode="before") + @classmethod + def _normalize_wifi_list(cls, value: Any) -> Any: + return [] if value is None else value + class GetStaticIPRsp(BaseModel): enabled: bool From 9691e268c992049a775d3444089e4561dc455337 Mon Sep 17 00:00:00 2001 From: Raphael Chicon Date: Fri, 24 Apr 2026 07:40:02 -0300 Subject: [PATCH 05/10] feat: add missing helper endpoints and uploads --- nanokvm/client.py | 286 +++++++++++++++++++++++++++++++++++++-- nanokvm/models/common.py | 29 +++- nanokvm/models/pro.py | 12 +- 3 files changed, 307 insertions(+), 20 deletions(-) diff --git a/nanokvm/client.py b/nanokvm/client.py index 58d6075..0cdf75f 100644 --- a/nanokvm/client.py +++ b/nanokvm/client.py @@ -9,6 +9,8 @@ import io import json import logging +from os import PathLike +from pathlib import Path import ssl from typing import Any, TypeVar, overload @@ -26,11 +28,15 @@ import yarl from .models.common import ( + AddShortcutReq, ApiResponse, ApiResponseCode, ChangePasswordReq, ConnectWifiReq, DeleteImageReq, + DeleteMacReq, + DeleteScriptReq, + DeleteShortcutReq, DownloadImageReq, GetAccountRsp, GetGpioRsp, @@ -39,11 +45,15 @@ GetHostnameRsp, GetImagesRsp, GetInfoRsp, + GetLeaderKeyRsp, + GetMacRsp, GetMdnsStateRsp, GetMountedImageRsp, GetMouseJigglerRsp, GetOLEDRsp, GetPreviewRsp, + GetScriptsRsp, + GetShortcutsRsp, GetSSHStateRsp, GetTailscaleStatusRsp, GetVersionRsp, @@ -56,19 +66,27 @@ IsPasswordUpdatedRsp, LoginReq, LoginRsp, + LoginTailscaleRsp, MountImageReq, MouseButton, MouseJigglerMode, PasteReq, + RunScriptReq, + RunScriptRsp, + RunScriptType, SetGpioReq, SetHidModeReq, SetHostnameReq, + SetLeaderKeyReq, + SetMacNameReq, SetMouseJigglerReq, SetOledReq, SetPreviewReq, SetWebTitleReq, + ShortcutKey, StatusImageRsp, UpdateVirtualDeviceReq, + UploadScriptRsp, VirtualDevice, WakeOnLANReq, ) @@ -78,7 +96,9 @@ GetMemoryLimitRsp, GetSwapSizeRsp, GetVirtualDeviceRsp, + ScreenSettingType, SetMemoryLimitReq, + SetScreenReq, SetSwapSizeReq, ) from .models.pro import ( @@ -116,6 +136,7 @@ SetTimeZoneReq, StreamMode, SwitchEdidReq, + UploadEdidRsp, ) from .utils import obfuscate_password @@ -325,12 +346,15 @@ async def _request( assert self._session is not None assert self._ssl_config is not None + request_headers = { + hdrs.ACCEPT: "application/json", + **kwargs.pop("headers", {}), + } + async with self._session.request( method, self.url / path.lstrip("/"), - headers={ - hdrs.ACCEPT: "application/json", - }, + headers=request_headers, cookies=cookies, timeout=aiohttp.ClientTimeout(total=self._request_timeout), raise_for_status=True, @@ -383,13 +407,73 @@ async def _api_request_json( try: raw_response = await response.json(content_type=None) _LOGGER.debug("Raw JSON response data: %s", raw_response) - # Parse the outer ApiResponse structure - api_response = ApiResponse[response_model].model_validate(raw_response) # type: ignore except (json.JSONDecodeError, ValidationError) as err: raise NanoKVMInvalidResponseError( f"Invalid JSON response received: {err}" ) from err + return self._validate_api_response(raw_response, response_model) + + @overload + async def _api_request_form( + self, + method: str, + path: str, + response_model: type[T], + data: aiohttp.FormData, + **kwargs: Any, + ) -> T: ... + + @overload + async def _api_request_form( + self, + method: str, + path: str, + response_model: None = None, + data: aiohttp.FormData | None = None, + **kwargs: Any, + ) -> None: ... + + async def _api_request_form( + self, + method: str, + path: str, + response_model: type[T] | None = None, + data: aiohttp.FormData | None = None, + **kwargs: Any, + ) -> T | None: + """Make API request with multipart/form data and parse JSON response.""" + _LOGGER.debug("Making API form request: %s %s", method, path) + + async with self._request( + method, + path, + data=data, + **kwargs, + ) as response: + try: + raw_response = await response.json(content_type=None) + _LOGGER.debug("Raw JSON response data: %s", raw_response) + except (json.JSONDecodeError, ValidationError) as err: + raise NanoKVMInvalidResponseError( + f"Invalid JSON response received: {err}" + ) from err + + return self._validate_api_response(raw_response, response_model) + + def _validate_api_response( + self, + raw_response: Any, + response_model: type[T] | None = None, + ) -> T | None: + """Validate the shared NanoKVM response envelope.""" + try: + api_response = ApiResponse[response_model].model_validate(raw_response) # type: ignore + except ValidationError as err: + raise NanoKVMInvalidResponseError( + f"Invalid JSON response received: {err}" + ) from err + _LOGGER.debug("Got API response: %s", api_response) if api_response.code != ApiResponseCode.SUCCESS: @@ -402,6 +486,27 @@ async def _api_request_json( return api_response.data + async def _upload_file( + self, + path: str, + file_path: str | PathLike[str], + response_model: type[T] | None = None, + **kwargs: Any, + ) -> T | None: + """Upload a file using the NanoKVM multipart API.""" + upload_path = Path(file_path) + form = aiohttp.FormData() + + with upload_path.open("rb") as file_obj: + form.add_field("file", file_obj, filename=upload_path.name) + return await self._api_request_form( + hdrs.METH_POST, + path, + response_model=response_model, + data=form, + **kwargs, + ) + # ── Authentication ────────────────────────────────────────────────── async def _do_authenticate(self, username: str, password_to_send: str) -> None: @@ -550,6 +655,43 @@ async def get_gpio(self) -> GetGpioRsp: response_model=GetGpioRsp, ) + async def get_scripts(self) -> GetScriptsRsp: + """Get the list of uploaded scripts.""" + return await self._api_request_json( + hdrs.METH_GET, + "/vm/script", + response_model=GetScriptsRsp, + ) + + async def upload_script( + self, file_path: str | PathLike[str] + ) -> UploadScriptRsp: + """Upload a script file.""" + return await self._upload_file( + "/vm/script/upload", + file_path, + response_model=UploadScriptRsp, + ) + + async def run_script( + self, name: str, script_type: RunScriptType + ) -> RunScriptRsp: + """Run an uploaded script.""" + return await self._api_request_json( + hdrs.METH_POST, + "/vm/script/run", + response_model=RunScriptRsp, + data=RunScriptReq(name=name, type=script_type), + ) + + async def delete_script(self, name: str) -> None: + """Delete an uploaded script.""" + await self._api_request_json( + hdrs.METH_DELETE, + "/vm/script", + data=DeleteScriptReq(name=name), + ) + async def push_button(self, button: GpioType, duration_ms: int) -> None: """Simulate pushing a hardware button.""" await self._api_request_json( @@ -668,8 +810,22 @@ async def reboot_system(self) -> None: """Reboot the KVM device.""" await self._api_request_json(hdrs.METH_POST, "/vm/system/reboot") + @require_hardware(HWVersion.PRO) + async def switch_to_pikvm(self) -> None: + """Switch the system image to PiKVM.""" + await self._api_request_json(hdrs.METH_POST, "/vm/system/pikvm") + # ── VM (non-Pro only) ────────────────────────────────────────────── + @require_hardware(HWVersion.ALPHA, HWVersion.BETA, HWVersion.PCIE) + async def set_screen(self, setting: ScreenSettingType, value: int) -> None: + """Set a legacy NanoKVM screen setting.""" + await self._api_request_json( + hdrs.METH_POST, + "/vm/screen", + data=SetScreenReq(type=setting, value=value), + ) + @require_hardware(HWVersion.ALPHA, HWVersion.BETA, HWVersion.PCIE) async def get_swap_size(self) -> int: """Get Swap size.""" @@ -831,6 +987,15 @@ async def get_custom_edid_list(self) -> GetCustomEdidListRsp: response_model=GetCustomEdidListRsp, ) + @require_hardware(HWVersion.PRO) + async def upload_edid(self, file_path: str | PathLike[str]) -> UploadEdidRsp: + """Upload a custom EDID.""" + return await self._upload_file( + "/vm/edid/upload", + file_path, + response_model=UploadEdidRsp, + ) + @require_hardware(HWVersion.PRO) async def delete_edid(self, edid: str) -> None: """Delete a custom EDID.""" @@ -871,12 +1036,18 @@ async def get_led_strip(self) -> GetLedStripRsp: async def set_led_strip( self, *, - on: bool, - horizontal_count: int, - vertical_count: int, - brightness: int, + on: bool | None = None, + horizontal_count: int | None = None, + vertical_count: int | None = None, + brightness: int | None = None, ) -> None: """Set LED strip configuration.""" + if all( + value is None + for value in (on, horizontal_count, vertical_count, brightness) + ): + raise ValueError("At least one LED strip setting must be provided") + await self._api_request_json( hdrs.METH_POST, "/vm/ledstrip/set", @@ -948,6 +1119,46 @@ async def get_hid_mode(self) -> GetHidModeRsp: response_model=GetHidModeRsp, ) + async def get_shortcuts(self) -> GetShortcutsRsp: + """Get configured custom HID shortcuts.""" + return await self._api_request_json( + hdrs.METH_GET, + "/hid/shortcuts", + response_model=GetShortcutsRsp, + ) + + async def add_shortcut(self, keys: list[ShortcutKey]) -> None: + """Add a custom HID shortcut.""" + await self._api_request_json( + hdrs.METH_POST, + "/hid/shortcut", + data=AddShortcutReq(keys=keys), + ) + + async def delete_shortcut(self, shortcut_id: str) -> None: + """Delete a custom HID shortcut.""" + await self._api_request_json( + hdrs.METH_DELETE, + "/hid/shortcut", + data=DeleteShortcutReq(id=shortcut_id), + ) + + async def get_leader_key(self) -> GetLeaderKeyRsp: + """Get the configured shortcut leader key.""" + return await self._api_request_json( + hdrs.METH_GET, + "/hid/shortcut/leader-key", + response_model=GetLeaderKeyRsp, + ) + + async def set_leader_key(self, key: str = "") -> None: + """Set or clear the shortcut leader key.""" + await self._api_request_json( + hdrs.METH_POST, + "/hid/shortcut/leader-key", + data=SetLeaderKeyReq(key=key), + ) + async def set_hid_mode(self, mode: HidMode) -> None: """Set the HID mode (requires reboot).""" await self._api_request_json( @@ -1042,6 +1253,31 @@ async def connect_wifi(self, ssid: str, password: str) -> None: data=ConnectWifiReq(ssid=ssid, password=password), ) + async def connect_wifi_no_auth( + self, + ssid: str, + password: str = "", + ap_password: str | None = None, + ) -> None: + """Connect to WiFi while the device is in AP-mode setup flow.""" + headers = {"X-AP-Key": ap_password} if ap_password is not None else {} + await self._api_request_json( + hdrs.METH_POST, + "/network/wifi", + authenticate=False, + headers=headers, + data=ConnectWifiReq(ssid=ssid, password=password), + ) + + async def verify_ap_login(self, ap_password: str) -> None: + """Verify AP-mode setup credentials.""" + await self._api_request_json( + hdrs.METH_POST, + "/network/wifi/verify", + authenticate=False, + headers={"X-AP-Key": ap_password}, + ) + async def disconnect_wifi(self) -> None: """Disconnect from the current WiFi network.""" await self._api_request_json(hdrs.METH_POST, "/network/wifi/disconnect") @@ -1052,6 +1288,30 @@ async def send_wake_on_lan(self, mac: str) -> None: hdrs.METH_POST, "/network/wol", data=WakeOnLANReq(mac=mac) ) + async def get_wol_macs(self) -> GetMacRsp: + """Get saved Wake-on-LAN MAC entries.""" + return await self._api_request_json( + hdrs.METH_GET, + "/network/wol/mac", + response_model=GetMacRsp, + ) + + async def delete_wol_mac(self, mac: str) -> None: + """Delete a saved Wake-on-LAN MAC entry.""" + await self._api_request_json( + hdrs.METH_DELETE, + "/network/wol/mac", + data=DeleteMacReq(mac=mac), + ) + + 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( + hdrs.METH_POST, + "/network/wol/mac/name", + data=SetMacNameReq(mac=mac, name=name), + ) + async def get_tailscale_status(self) -> GetTailscaleStatusRsp: """Get Tailscale status.""" return await self._api_request_json( @@ -1245,9 +1505,13 @@ async def tailscale_down(self) -> None: """Bring Tailscale down.""" await self._api_request_json(hdrs.METH_POST, "/extensions/tailscale/down") - async def tailscale_login(self) -> None: + async def tailscale_login(self) -> LoginTailscaleRsp: """Log in to Tailscale.""" - await self._api_request_json(hdrs.METH_POST, "/extensions/tailscale/login") + return await self._api_request_json( + hdrs.METH_POST, + "/extensions/tailscale/login", + response_model=LoginTailscaleRsp, + ) async def tailscale_logout(self) -> None: """Log out of Tailscale.""" diff --git a/nanokvm/models/common.py b/nanokvm/models/common.py index 8599798..f6ad2fa 100644 --- a/nanokvm/models/common.py +++ b/nanokvm/models/common.py @@ -194,7 +194,16 @@ class GetGpioRsp(BaseModel): class GetScriptsRsp(BaseModel): - files: list[str] + files: list[str] = Field(default_factory=list) + + @field_validator("files", mode="before") + @classmethod + def _normalize_files(cls, value: Any) -> Any: + return [] if value is None else value + + +class UploadScriptRsp(BaseModel): + file: str class RunScriptReq(BaseModel): @@ -284,7 +293,12 @@ class Shortcut(BaseModel): class GetShortcutsRsp(BaseModel): - shortcuts: list[Shortcut] + shortcuts: list[Shortcut] = Field(default_factory=list) + + @field_validator("shortcuts", mode="before") + @classmethod + def _normalize_shortcuts(cls, value: Any) -> Any: + return [] if value is None else value class AddShortcutReq(BaseModel): @@ -339,7 +353,12 @@ class WakeOnLANReq(BaseModel): class GetMacRsp(BaseModel): - macs: list[str] + macs: list[str] = Field(default_factory=list) + + @field_validator("macs", mode="before") + @classmethod + def _normalize_macs(cls, value: Any) -> Any: + return [] if value is None else value class DeleteMacReq(BaseModel): @@ -386,7 +405,7 @@ def _normalize_wifi_fields(cls, value: Any) -> Any: class ConnectWifiReq(BaseModel): ssid: str - password: str + password: str = "" class GetTailscaleStatusRsp(BaseModel): @@ -397,7 +416,7 @@ class GetTailscaleStatusRsp(BaseModel): class LoginTailscaleRsp(BaseModel): - url: str + url: str = "" # Application Models diff --git a/nanokvm/models/pro.py b/nanokvm/models/pro.py index 95f9ab1..a6e9b12 100644 --- a/nanokvm/models/pro.py +++ b/nanokvm/models/pro.py @@ -133,6 +133,10 @@ class DeleteEdidReq(BaseModel): edid: str +class UploadEdidRsp(BaseModel): + file: str + + class GetHdmiCaptureRsp(BaseModel): enabled: bool @@ -187,10 +191,10 @@ class SetMenuBarConfigReq(BaseModel): class SetLedStripReq(BaseModel): model_config = ConfigDict(populate_by_name=True) - on: bool - horizontal_count: int = Field(alias="hor") - vertical_count: int = Field(alias="ver") - brightness: int + on: bool | None = None + horizontal_count: int | None = Field(default=None, alias="hor") + vertical_count: int | None = Field(default=None, alias="ver") + brightness: int | None = None class GetLedStripRsp(BaseModel): From 9aa1d0d42905a73a660373b80fda913da645e4e2 Mon Sep 17 00:00:00 2001 From: Raphael Chicon Date: Sat, 25 Apr 2026 15:16:54 -0300 Subject: [PATCH 06/10] fix(client): handle empty API success responses --- nanokvm/client.py | 30 +++++++++++++++++++++++++++--- tests/test_client.py | 22 +++++++++++++++++++++- 2 files changed, 48 insertions(+), 4 deletions(-) diff --git a/nanokvm/client.py b/nanokvm/client.py index 0cdf75f..de7efb2 100644 --- a/nanokvm/client.py +++ b/nanokvm/client.py @@ -468,7 +468,10 @@ def _validate_api_response( ) -> T | None: """Validate the shared NanoKVM response envelope.""" try: - api_response = ApiResponse[response_model].model_validate(raw_response) # type: ignore + if response_model is None: + api_response = ApiResponse[Any].model_validate(raw_response) + else: + api_response = ApiResponse[response_model].model_validate(raw_response) # type: ignore[valid-type] except ValidationError as err: raise NanoKVMInvalidResponseError( f"Invalid JSON response received: {err}" @@ -484,8 +487,29 @@ def _validate_api_response( data=api_response.data, ) + if response_model is None: + return None + return api_response.data + @overload + async def _upload_file( + self, + path: str, + file_path: str | PathLike[str], + response_model: type[T], + **kwargs: Any, + ) -> T: ... + + @overload + async def _upload_file( + self, + path: str, + file_path: str | PathLike[str], + response_model: None = None, + **kwargs: Any, + ) -> None: ... + async def _upload_file( self, path: str, @@ -1398,11 +1422,11 @@ async def set_fps(self, fps: int) -> None: # ── Stream (shared) ──────────────────────────────────────────────── - def _parse_jpeg_from_bytes(self, data: bytes) -> Image: + def _parse_jpeg_from_bytes(self, data: bytes) -> Image.Image: """Parse JPEG image from bytes.""" return Image.open(io.BytesIO(data), formats=["JPEG"]) - async def mjpeg_stream(self) -> AsyncIterator[Image]: + async def mjpeg_stream(self) -> AsyncIterator[Image.Image]: """Stream MJPEG frames.""" async with self._request(hdrs.METH_GET, "/stream/mjpeg") as response: reader = MultipartReader.from_response(response) diff --git a/tests/test_client.py b/tests/test_client.py index 05bf9c9..2cd7c4a 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -1,9 +1,10 @@ from aiohttp import ClientSession from aioresponses import aioresponses import pytest +import yarl from nanokvm.client import NanoKVMApiError, NanoKVMClient -from nanokvm.models import ApiResponseCode +from nanokvm.models import ApiResponseCode, VirtualDevice async def test_get_images_success() -> None: @@ -69,6 +70,25 @@ async def test_get_images_api_error() -> None: assert "failed to list images" in exc_info.value.msg +async def test_update_virtual_device_ignores_success_data() -> None: + """Test update_virtual_device handles non-Pro success payloads.""" + async with NanoKVMClient( + "http://localhost:8888/api/", token="test-token" + ) as client: + with aioresponses() as m: + m.post( + "http://localhost:8888/api/vm/device/virtual", + payload={"code": 0, "msg": "success", "data": {"on": True}}, + ) + + await client.update_virtual_device(VirtualDevice.DISK) + + calls = m.requests[ + ("POST", yarl.URL("http://localhost:8888/api/vm/device/virtual")) + ] + assert calls[0].kwargs.get("json") == {"device": "disk"} + + async def test_client_context_manager() -> None: """Test that client properly initializes and cleans up with context manager.""" async with NanoKVMClient( From 5dc38a5d515ae62c36de0d045b4d1027cba845f0 Mon Sep 17 00:00:00 2001 From: Raphael Chicon Date: Sat, 25 Apr 2026 15:18:53 -0300 Subject: [PATCH 07/10] fix(pro): preserve LED strip settings --- nanokvm/client.py | 21 ++++++++++++++++++++ nanokvm/models/pro.py | 8 ++++---- tests/test_client.py | 45 ++++++++++++++++++++++++++++++++++++++++++- 3 files changed, 69 insertions(+), 5 deletions(-) diff --git a/nanokvm/client.py b/nanokvm/client.py index de7efb2..842814e 100644 --- a/nanokvm/client.py +++ b/nanokvm/client.py @@ -1072,6 +1072,27 @@ async def set_led_strip( ): raise ValueError("At least one LED strip setting must be provided") + if any( + value is None + for value in (on, horizontal_count, vertical_count, brightness) + ): + current = await self.get_led_strip() + on = current.on if on is None else on + horizontal_count = ( + current.horizontal_count + if horizontal_count is None + else horizontal_count + ) + vertical_count = ( + current.vertical_count if vertical_count is None else vertical_count + ) + brightness = current.brightness if brightness is None else brightness + + assert on is not None + assert horizontal_count is not None + assert vertical_count is not None + assert brightness is not None + await self._api_request_json( hdrs.METH_POST, "/vm/ledstrip/set", diff --git a/nanokvm/models/pro.py b/nanokvm/models/pro.py index a6e9b12..6a8ab4c 100644 --- a/nanokvm/models/pro.py +++ b/nanokvm/models/pro.py @@ -191,10 +191,10 @@ class SetMenuBarConfigReq(BaseModel): class SetLedStripReq(BaseModel): model_config = ConfigDict(populate_by_name=True) - on: bool | None = None - horizontal_count: int | None = Field(default=None, alias="hor") - vertical_count: int | None = Field(default=None, alias="ver") - brightness: int | None = None + on: bool + horizontal_count: int = Field(alias="hor") + vertical_count: int = Field(alias="ver") + brightness: int class GetLedStripRsp(BaseModel): diff --git a/tests/test_client.py b/tests/test_client.py index 2cd7c4a..88614fa 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -4,7 +4,11 @@ import yarl from nanokvm.client import NanoKVMApiError, NanoKVMClient -from nanokvm.models import ApiResponseCode, VirtualDevice +from nanokvm.models import ( + ApiResponseCode, + HWVersion, + VirtualDevice, +) async def test_get_images_success() -> None: @@ -89,6 +93,45 @@ async def test_update_virtual_device_ignores_success_data() -> None: assert calls[0].kwargs.get("json") == {"device": "disk"} +async def test_set_led_strip_partial_preserves_current_config() -> None: + """Test partial LED updates post a complete Pro LED configuration.""" + async with NanoKVMClient( + "http://localhost:8888/api/", token="test-token" + ) as client: + client._hw_version = HWVersion.PRO + + with aioresponses() as m: + m.get( + "http://localhost:8888/api/vm/ledstrip/get", + payload={ + "code": 0, + "msg": "success", + "data": { + "on": True, + "hor": 8, + "ver": 6, + "brightness": 25, + }, + }, + ) + m.post( + "http://localhost:8888/api/vm/ledstrip/set", + payload={"code": 0, "msg": "success", "data": None}, + ) + + await client.set_led_strip(brightness=50) + + calls = m.requests[ + ("POST", yarl.URL("http://localhost:8888/api/vm/ledstrip/set")) + ] + assert calls[0].kwargs.get("json") == { + "on": True, + "hor": 8, + "ver": 6, + "brightness": 50, + } + + async def test_client_context_manager() -> None: """Test that client properly initializes and cleans up with context manager.""" async with NanoKVMClient( From 96054af1fdeaae583dde60ef1fbaf0cbb26c34a4 Mon Sep 17 00:00:00 2001 From: Raphael Chicon Date: Sat, 25 Apr 2026 15:19:03 -0300 Subject: [PATCH 08/10] fix(pro): handle endpoint-specific API errors --- nanokvm/client.py | 32 ++++++++++-- nanokvm/models/common.py | 11 +++- tests/test_client.py | 107 +++++++++++++++++++++++++++++++++++++++ 3 files changed, 144 insertions(+), 6 deletions(-) diff --git a/nanokvm/client.py b/nanokvm/client.py index 842814e..e968b13 100644 --- a/nanokvm/client.py +++ b/nanokvm/client.py @@ -70,6 +70,7 @@ MountImageReq, MouseButton, MouseJigglerMode, + OledType, PasteReq, RunScriptReq, RunScriptRsp, @@ -758,11 +759,28 @@ async def disable_mdns(self) -> None: async def get_oled_info(self) -> GetOLEDRsp: """Get OLED information.""" - return await self._api_request_json( - hdrs.METH_GET, - "/vm/oled", - response_model=GetOLEDRsp, - ) + try: + return await self._api_request_json( + hdrs.METH_GET, + "/vm/oled", + response_model=GetOLEDRsp, + ) + except NanoKVMApiError as err: + if ( + self._hw_version != HWVersion.PRO + or err.code != ApiResponseCode.ENDPOINT_ERROR_2 + or err.msg != "invalid file content" + ): + raise + + oled_type = OledType.DESK + with contextlib.suppress(NanoKVMError): + info = await self.get_info() + part_number = (info.part_number or "").casefold() + if "atx" in part_number: + oled_type = OledType.ATX + + return GetOLEDRsp(exist=True, type=oled_type, sleep=0) async def set_oled_sleep(self, sleep_seconds: int) -> None: """Set the OLED sleep timeout.""" @@ -808,6 +826,10 @@ async def set_mouse_jiggler_state( self, enabled: bool, mode: MouseJigglerMode ) -> None: """Set the mouse jiggler state.""" + current = await self.get_mouse_jiggler_state() + if current.enabled == enabled and (not enabled or current.mode == mode): + return + await self._api_request_json( hdrs.METH_POST, "/vm/mouse-jiggler/", diff --git a/nanokvm/models/common.py b/nanokvm/models/common.py index f6ad2fa..af99fbb 100644 --- a/nanokvm/models/common.py +++ b/nanokvm/models/common.py @@ -11,11 +11,20 @@ class ApiResponseCode(IntEnum): - """API Response Codes.""" + """API response codes. + + NanoKVM Pro reuses negative values from -3 through -6 with + endpoint-specific meanings. Inspect the response message for details. + """ SUCCESS = 0 FAILURE = -1 INVALID_USERNAME_OR_PASSWORD = -2 + ENDPOINT_ERROR_2 = -2 + ENDPOINT_ERROR_3 = -3 + ENDPOINT_ERROR_4 = -4 + ENDPOINT_ERROR_5 = -5 + ENDPOINT_ERROR_6 = -6 class HidMode(StrEnum): diff --git a/tests/test_client.py b/tests/test_client.py index 88614fa..797360d 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -7,6 +7,8 @@ from nanokvm.models import ( ApiResponseCode, HWVersion, + MouseJigglerMode, + OledType, VirtualDevice, ) @@ -74,6 +76,111 @@ async def test_get_images_api_error() -> None: assert "failed to list images" in exc_info.value.msg +async def test_api_error_allows_endpoint_specific_codes() -> None: + """Test endpoint-specific API error codes are surfaced as API errors.""" + async with NanoKVMClient( + "http://localhost:8888/api/", token="test-token" + ) as client: + with aioresponses() as m: + m.post( + "http://localhost:8888/api/storage/image/mount", + payload={"code": -6, "msg": "mount image failed", "data": None}, + ) + + with pytest.raises(NanoKVMApiError) as exc_info: + await client.mount_image("/data/missing.iso", read_only=True) + + assert exc_info.value.code == ApiResponseCode.ENDPOINT_ERROR_6 + assert "mount image failed" in exc_info.value.msg + + +async def test_mount_image_sends_pro_read_only_flag() -> None: + """Test mount_image sends the Pro readOnly field.""" + async with NanoKVMClient( + "http://localhost:8888/api/", token="test-token" + ) as client: + with aioresponses() as m: + m.post( + "http://localhost:8888/api/storage/image/mount", + payload={"code": 0, "msg": "success", "data": None}, + ) + + await client.mount_image("/data/test.img", read_only=True) + + calls = m.requests[ + ("POST", yarl.URL("http://localhost:8888/api/storage/image/mount")) + ] + assert calls[0].kwargs.get("json") == { + "file": "/data/test.img", + "cdrom": False, + "readOnly": True, + } + + +async def test_get_oled_info_falls_back_for_pro_invalid_file_content() -> None: + """Test Pro OLED fallback for firmware that rejects lower-case desk IDs.""" + async with NanoKVMClient( + "http://localhost:8888/api/", token="test-token" + ) as client: + client._hw_version = HWVersion.PRO + + with aioresponses() as m: + m.get( + "http://localhost:8888/api/vm/oled", + payload={ + "code": -2, + "msg": "invalid file content", + "data": None, + }, + ) + m.get( + "http://localhost:8888/api/vm/info", + payload={ + "code": 0, + "msg": "success", + "data": { + "ips": [], + "mdns": "kvm-test.local", + "image": "v1.0.14", + "application": "1.2.14", + "deviceKey": "test-device", + "pn": "unknown", + "arch": "aarch64", + }, + }, + ) + + response = await client.get_oled_info() + + assert response.exist is True + assert response.type == OledType.DESK + assert response.sleep == 0 + + +async def test_set_mouse_jiggler_state_noops_when_already_disabled() -> None: + """Test disabling mouse jiggler is idempotent.""" + async with NanoKVMClient( + "http://localhost:8888/api/", token="test-token" + ) as client: + with aioresponses() as m: + m.get( + "http://localhost:8888/api/vm/mouse-jiggler", + payload={ + "code": 0, + "msg": "success", + "data": {"enabled": False, "mode": "relative"}, + }, + ) + + await client.set_mouse_jiggler_state( + enabled=False, + mode=MouseJigglerMode.RELATIVE, + ) + + post_url = yarl.URL("http://localhost:8888/api/vm/mouse-jiggler/") + assert ("POST", post_url) not in m.requests + + async def test_update_virtual_device_ignores_success_data() -> None: """Test update_virtual_device handles non-Pro success payloads.""" async with NanoKVMClient( From f7b1fe8a83518f60530a725786dc9b172c881e08 Mon Sep 17 00:00:00 2001 From: Raphael Chicon Date: Sat, 25 Apr 2026 15:34:33 -0300 Subject: [PATCH 09/10] style: format client methods --- nanokvm/client.py | 8 ++------ 1 file changed, 2 insertions(+), 6 deletions(-) diff --git a/nanokvm/client.py b/nanokvm/client.py index e968b13..083a3e2 100644 --- a/nanokvm/client.py +++ b/nanokvm/client.py @@ -688,9 +688,7 @@ async def get_scripts(self) -> GetScriptsRsp: response_model=GetScriptsRsp, ) - async def upload_script( - self, file_path: str | PathLike[str] - ) -> UploadScriptRsp: + async def upload_script(self, file_path: str | PathLike[str]) -> UploadScriptRsp: """Upload a script file.""" return await self._upload_file( "/vm/script/upload", @@ -698,9 +696,7 @@ async def upload_script( response_model=UploadScriptRsp, ) - async def run_script( - self, name: str, script_type: RunScriptType - ) -> RunScriptRsp: + async def run_script(self, name: str, script_type: RunScriptType) -> RunScriptRsp: """Run an uploaded script.""" return await self._api_request_json( hdrs.METH_POST, From 66e519de1653f1555f78c3ef97181da401a3a9e4 Mon Sep 17 00:00:00 2001 From: Raphael Chicon Date: Sun, 26 Apr 2026 11:24:27 -0300 Subject: [PATCH 10/10] fix: harden API error handling for device quirks --- nanokvm/client.py | 42 ++++++++++++++++++------ nanokvm/models/common.py | 13 ++------ tests/test_client.py | 70 ++++++++++++++++++++++++++++++++++++++-- 3 files changed, 102 insertions(+), 23 deletions(-) diff --git a/nanokvm/client.py b/nanokvm/client.py index 083a3e2..3f1176c 100644 --- a/nanokvm/client.py +++ b/nanokvm/client.py @@ -480,7 +480,7 @@ def _validate_api_response( _LOGGER.debug("Got API response: %s", api_response) - if api_response.code != ApiResponseCode.SUCCESS: + if api_response.code != ApiResponseCode.SUCCESS.value: raise NanoKVMApiError( f"API returned error: {api_response.msg} (Code: {api_response.code})", code=api_response.code, @@ -555,7 +555,7 @@ async def _do_authenticate(self, username: str, password_to_send: str) -> None: self._token = login_response.token except NanoKVMApiError as err: - if err.code == ApiResponseCode.INVALID_USERNAME_OR_PASSWORD: + if err.code == ApiResponseCode.INVALID_USERNAME_OR_PASSWORD.value: raise NanoKVMAuthenticationFailure( "Invalid username or password" ) from err @@ -764,7 +764,7 @@ async def get_oled_info(self) -> GetOLEDRsp: except NanoKVMApiError as err: if ( self._hw_version != HWVersion.PRO - or err.code != ApiResponseCode.ENDPOINT_ERROR_2 + or err.code != ApiResponseCode.INVALID_USERNAME_OR_PASSWORD.value or err.msg != "invalid file content" ): raise @@ -888,12 +888,26 @@ async def set_swap_size(self, size_mb: int) -> None: @require_hardware(HWVersion.ALPHA, HWVersion.BETA, HWVersion.PCIE) async def enable_swap(self) -> None: """Enable swap.""" - await self._api_request_json(hdrs.METH_POST, "/vm/swap/enable") + try: + await self._api_request_json(hdrs.METH_POST, "/vm/swap/enable") + except aiohttp.ClientResponseError as err: + if err.status == 404: + raise NanoKVMNotSupportedError( + "enable_swap is unavailable on this legacy hardware/firmware" + ) from err + raise @require_hardware(HWVersion.ALPHA, HWVersion.BETA, HWVersion.PCIE) async def disable_swap(self) -> None: """Disable swap.""" - await self._api_request_json(hdrs.METH_POST, "/vm/swap/disable") + try: + await self._api_request_json(hdrs.METH_POST, "/vm/swap/disable") + except aiohttp.ClientResponseError as err: + if err.status == 404: + raise NanoKVMNotSupportedError( + "disable_swap is unavailable on this legacy hardware/firmware" + ) from err + raise @require_hardware(HWVersion.ALPHA, HWVersion.BETA, HWVersion.PCIE) async def get_memory_limit(self) -> GetMemoryLimitRsp: @@ -1353,11 +1367,19 @@ async def send_wake_on_lan(self, mac: str) -> None: async def get_wol_macs(self) -> GetMacRsp: """Get saved Wake-on-LAN MAC entries.""" - return await self._api_request_json( - hdrs.METH_GET, - "/network/wol/mac", - response_model=GetMacRsp, - ) + try: + return await self._api_request_json( + hdrs.METH_GET, + "/network/wol/mac", + response_model=GetMacRsp, + ) + except NanoKVMApiError as err: + if ( + err.code == ApiResponseCode.INVALID_USERNAME_OR_PASSWORD.value + and err.msg == "open file error" + ): + return GetMacRsp() + raise async def delete_wol_mac(self, mac: str) -> None: """Delete a saved Wake-on-LAN MAC entry.""" diff --git a/nanokvm/models/common.py b/nanokvm/models/common.py index af99fbb..9e2fa8e 100644 --- a/nanokvm/models/common.py +++ b/nanokvm/models/common.py @@ -11,20 +11,11 @@ class ApiResponseCode(IntEnum): - """API response codes. - - NanoKVM Pro reuses negative values from -3 through -6 with - endpoint-specific meanings. Inspect the response message for details. - """ + """Documented shared API response codes.""" SUCCESS = 0 FAILURE = -1 INVALID_USERNAME_OR_PASSWORD = -2 - ENDPOINT_ERROR_2 = -2 - ENDPOINT_ERROR_3 = -3 - ENDPOINT_ERROR_4 = -4 - ENDPOINT_ERROR_5 = -5 - ENDPOINT_ERROR_6 = -6 class HidMode(StrEnum): @@ -123,7 +114,7 @@ class OledType(StrEnum): class ApiResponse(BaseModel, Generic[T]): """Generic API response structure.""" - code: ApiResponseCode + code: int msg: str data: T | None = None diff --git a/tests/test_client.py b/tests/test_client.py index 797360d..b5852bf 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -3,7 +3,7 @@ import pytest import yarl -from nanokvm.client import NanoKVMApiError, NanoKVMClient +from nanokvm.client import NanoKVMApiError, NanoKVMClient, NanoKVMNotSupportedError from nanokvm.models import ( ApiResponseCode, HWVersion, @@ -90,10 +90,28 @@ async def test_api_error_allows_endpoint_specific_codes() -> None: with pytest.raises(NanoKVMApiError) as exc_info: await client.mount_image("/data/missing.iso", read_only=True) - assert exc_info.value.code == ApiResponseCode.ENDPOINT_ERROR_6 + assert exc_info.value.code == -6 assert "mount image failed" in exc_info.value.msg +async def test_none_returning_endpoint_preserves_unknown_api_code() -> None: + """Test unknown API codes from None-returning endpoints remain API errors.""" + async with NanoKVMClient( + "http://localhost:8888/api/", token="test-token" + ) as client: + with aioresponses() as m: + m.post( + "http://localhost:8888/api/vm/web-title", + payload={"code": -4, "msg": "failed to set title", "data": None}, + ) + + with pytest.raises(NanoKVMApiError) as exc_info: + await client.set_web_title("NanoKVM") + + assert exc_info.value.code == -4 + assert "failed to set title" in exc_info.value.msg + + async def test_mount_image_sends_pro_read_only_flag() -> None: """Test mount_image sends the Pro readOnly field.""" async with NanoKVMClient( @@ -239,6 +257,54 @@ async def test_set_led_strip_partial_preserves_current_config() -> None: } +async def test_get_wol_macs_returns_empty_list_for_missing_file() -> None: + """Test WOL getter normalizes the empty-file device state.""" + async with NanoKVMClient( + "http://localhost:8888/api/", token="test-token" + ) as client: + with aioresponses() as m: + m.get( + "http://localhost:8888/api/network/wol/mac", + payload={"code": -2, "msg": "open file error", "data": None}, + ) + + response = await client.get_wol_macs() + + assert response.macs == [] + + +async def test_enable_swap_404_is_not_supported() -> None: + """Test legacy swap enable 404 becomes NanoKVMNotSupportedError.""" + async with NanoKVMClient( + "http://localhost:8888/api/", token="test-token" + ) as client: + client._hw_version = HWVersion.PCIE + + with aioresponses() as m: + m.post("http://localhost:8888/api/vm/swap/enable", status=404) + + with pytest.raises(NanoKVMNotSupportedError) as exc_info: + await client.enable_swap() + + assert "enable_swap is unavailable" in str(exc_info.value) + + +async def test_disable_swap_404_is_not_supported() -> None: + """Test legacy swap disable 404 becomes NanoKVMNotSupportedError.""" + async with NanoKVMClient( + "http://localhost:8888/api/", token="test-token" + ) as client: + client._hw_version = HWVersion.PCIE + + with aioresponses() as m: + m.post("http://localhost:8888/api/vm/swap/disable", status=404) + + with pytest.raises(NanoKVMNotSupportedError) as exc_info: + await client.disable_swap() + + assert "disable_swap is unavailable" in str(exc_info.value) + + async def test_client_context_manager() -> None: """Test that client properly initializes and cleans up with context manager.""" async with NanoKVMClient(