diff --git a/nanokvm/client.py b/nanokvm/client.py index 991e230..3f1176c 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,28 @@ IsPasswordUpdatedRsp, LoginReq, LoginRsp, + LoginTailscaleRsp, MountImageReq, MouseButton, MouseJigglerMode, + OledType, PasteReq, + RunScriptReq, + RunScriptRsp, + RunScriptType, SetGpioReq, SetHidModeReq, SetHostnameReq, + SetLeaderKeyReq, + SetMacNameReq, SetMouseJigglerReq, SetOledReq, SetPreviewReq, SetWebTitleReq, + ShortcutKey, StatusImageRsp, UpdateVirtualDeviceReq, + UploadScriptRsp, VirtualDevice, WakeOnLANReq, ) @@ -78,12 +97,15 @@ GetMemoryLimitRsp, GetSwapSizeRsp, GetVirtualDeviceRsp, + ScreenSettingType, SetMemoryLimitReq, + SetScreenReq, SetSwapSizeReq, ) from .models.pro import ( DeleteEdidReq, DiskType, + EdidValue, GetCustomEdidListRsp, GetEdidRsp, GetHdmiCaptureRsp, @@ -96,7 +118,7 @@ GetStaticIPRsp, GetTimeStatusRsp, GetTimeZoneRsp, - GetVirtualDeviceProRsp, + LcdTimeFormat, RateControlMode, RefreshVirtualDeviceReq, ScanWifiRsp, @@ -113,7 +135,9 @@ SetStreamModeReq, SetStreamQualityReq, SetTimeZoneReq, + StreamMode, SwitchEdidReq, + UploadEdidRsp, ) from .utils import obfuscate_password @@ -323,12 +347,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, @@ -381,16 +408,79 @@ 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: + 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}" + ) from err + _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, @@ -398,8 +488,50 @@ async def _api_request_json( 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, + 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: @@ -423,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 @@ -445,12 +577,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 +601,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, ), ) @@ -534,6 +680,39 @@ 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( @@ -576,11 +755,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.INVALID_USERNAME_OR_PASSWORD.value + 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.""" @@ -590,19 +786,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( @@ -633,6 +822,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/", @@ -659,8 +852,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.""" @@ -681,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: @@ -751,7 +972,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, @@ -805,7 +1026,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, @@ -822,6 +1043,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.""" @@ -862,12 +1092,39 @@ 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") + + 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", @@ -939,6 +1196,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( @@ -1033,6 +1330,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") @@ -1043,6 +1365,38 @@ 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.""" + 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.""" + 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( @@ -1092,7 +1446,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, @@ -1129,11 +1483,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) @@ -1236,9 +1590,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 554bf4e..9e2fa8e 100644 --- a/nanokvm/models/common.py +++ b/nanokvm/models/common.py @@ -5,13 +5,13 @@ 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") class ApiResponseCode(IntEnum): - """API Response Codes.""" + """Documented shared API response codes.""" SUCCESS = 0 FAILURE = -1 @@ -103,11 +103,18 @@ 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.""" - code: ApiResponseCode + code: int msg: str data: T | None = None @@ -152,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): @@ -179,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): @@ -204,9 +228,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 @@ -262,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): @@ -317,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): @@ -348,10 +389,23 @@ 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 - password: str + password: str = "" class GetTailscaleStatusRsp(BaseModel): @@ -362,7 +416,7 @@ class GetTailscaleStatusRsp(BaseModel): class LoginTailscaleRsp(BaseModel): - url: str + url: str = "" # Application Models 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): diff --git a/nanokvm/models/pro.py b/nanokvm/models/pro.py index 55faed8..6a8ab4c 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): @@ -59,11 +123,20 @@ 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 +class UploadEdidRsp(BaseModel): + file: str + + class GetHdmiCaptureRsp(BaseModel): enabled: bool @@ -96,11 +169,11 @@ class GetTimeStatusRsp(BaseModel): class GetLcdTimeFormatRsp(BaseModel): - format: str + format: LcdTimeFormat class SetLcdTimeFormatReq(BaseModel): - format: str + format: LcdTimeFormat class GetMenuBarConfigRsp(BaseModel): @@ -137,6 +210,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 @@ -154,7 +232,7 @@ class SetRateControlModeReq(BaseModel): class SetStreamModeReq(BaseModel): - mode: str + mode: StreamMode class SetStreamQualityReq(BaseModel): diff --git a/tests/test_client.py b/tests/test_client.py index 05bf9c9..b5852bf 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -1,9 +1,16 @@ 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.client import NanoKVMApiError, NanoKVMClient, NanoKVMNotSupportedError +from nanokvm.models import ( + ApiResponseCode, + HWVersion, + MouseJigglerMode, + OledType, + VirtualDevice, +) async def test_get_images_success() -> None: @@ -69,6 +76,235 @@ 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 == -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( + "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( + "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_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_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(