From 09c54df84cf2eedee3cf43d55d41099f7dc80fa7 Mon Sep 17 00:00:00 2001 From: Raphael Chicon Date: Sun, 17 May 2026 20:03:28 -0300 Subject: [PATCH] Add non-Pro DNS management API --- nanokvm/client.py | 25 ++++++ nanokvm/models/non_pro.py | 50 +++++++++++- tests/test_client.py | 166 ++++++++++++++++++++++++++++++++++++++ 3 files changed, 240 insertions(+), 1 deletion(-) diff --git a/nanokvm/client.py b/nanokvm/client.py index 8cb8338..10a304a 100644 --- a/nanokvm/client.py +++ b/nanokvm/client.py @@ -93,11 +93,14 @@ WakeOnLANReq, ) from .models.non_pro import ( + DNSMode, GetCdRomRsp, + GetDNSRsp, GetHdmiStateRsp, GetMemoryLimitRsp, GetSwapSizeRsp, ScreenSettingType, + SetDNSReq, SetMemoryLimitReq, SetScreenReq, SetSwapSizeReq, @@ -1504,6 +1507,28 @@ async def set_wol_mac_name(self, mac: str, name: str) -> None: data=SetMacNameReq(mac=mac, name=name), ) + @require_hardware(HWVersion.ALPHA, HWVersion.BETA, HWVersion.PCIE) + @require_application_version(non_pro="2.4.1") + async def get_dns(self) -> GetDNSRsp: + """Get DNS configuration.""" + return await self._api_request_json( + hdrs.METH_GET, + "/network/dns", + response_model=GetDNSRsp, + ) + + @require_hardware(HWVersion.ALPHA, HWVersion.BETA, HWVersion.PCIE) + @require_application_version(non_pro="2.4.1") + async def set_dns( + self, mode: DNSMode | str, servers: list[str] | None = None + ) -> None: + """Set DNS configuration.""" + await self._api_request_json( + hdrs.METH_POST, + "/network/dns", + data=SetDNSReq(mode=DNSMode(mode), servers=servers or []), + ) + async def get_tailscale_status(self) -> GetTailscaleStatusRsp: """Get Tailscale status.""" return await self._api_request_json( diff --git a/nanokvm/models/non_pro.py b/nanokvm/models/non_pro.py index 5c712b2..27f41da 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 class ScreenSettingType(StrEnum): @@ -15,6 +16,19 @@ class ScreenSettingType(StrEnum): QUALITY = "quality" +class DNSMode(StrEnum): + """DNS configuration modes.""" + + MANUAL = "manual" + DHCP = "dhcp" + + +def _normalize_string_list(value: Any) -> Any: + if value is None: + return [] + return value + + class SetScreenReq(BaseModel): """Pro uses separate stream endpoints instead.""" @@ -46,3 +60,37 @@ class GetHdmiStateRsp(BaseModel): class GetCdRomRsp(BaseModel): cdrom: int + + +class DNSInfo(BaseModel): + model_config = ConfigDict(populate_by_name=True) + + interface: str = "" + type: str = "" + address: str = "" + subnet_mask: str = Field("", alias="subnetMask") + gateway: str = "" + search_domains: list[str] = Field(default_factory=list, alias="searchDomains") + + @field_validator("search_domains", mode="before") + @classmethod + def _normalize_search_domains(cls, value: Any) -> Any: + return _normalize_string_list(value) + + +class GetDNSRsp(BaseModel): + mode: DNSMode + servers: list[str] = Field(default_factory=list) + effective: list[str] = Field(default_factory=list) + dhcp: list[str] = Field(default_factory=list) + info: DNSInfo + + @field_validator("servers", "effective", "dhcp", mode="before") + @classmethod + def _normalize_dns_lists(cls, value: Any) -> Any: + return _normalize_string_list(value) + + +class SetDNSReq(BaseModel): + mode: DNSMode + servers: list[str] = Field(default_factory=list) diff --git a/tests/test_client.py b/tests/test_client.py index f03ead3..46fac09 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -16,6 +16,7 @@ from nanokvm.models import ( ApiResponseCode, DiskType, + DNSMode, GetMacRsp, GetOLEDRsp, HWVersion, @@ -446,6 +447,171 @@ 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_get_dns_parses_configuration() -> None: + """Test DNS configuration response parsing.""" + async with NanoKVMClient( + "http://localhost:8888/api/", token="test-token" + ) as client: + _mark_detected(client, application_version="2.4.1") + + with aioresponses() as m: + m.get( + "http://localhost:8888/api/network/dns", + payload={ + "code": 0, + "msg": "success", + "data": { + "mode": "manual", + "servers": ["1.1.1.1", "2606:4700:4700::1111"], + "effective": ["1.1.1.1"], + "dhcp": ["192.168.1.1"], + "info": { + "interface": "eth0", + "type": "Wired", + "address": "192.168.1.20/24", + "subnetMask": "255.255.255.0", + "gateway": "192.168.1.1", + "searchDomains": ["lan"], + }, + }, + }, + ) + + response = await client.get_dns() + + assert response.mode is DNSMode.MANUAL + assert response.servers == ["1.1.1.1", "2606:4700:4700::1111"] + assert response.effective == ["1.1.1.1"] + assert response.dhcp == ["192.168.1.1"] + assert response.info.interface == "eth0" + assert response.info.type == "Wired" + assert response.info.address == "192.168.1.20/24" + assert response.info.subnet_mask == "255.255.255.0" + assert response.info.gateway == "192.168.1.1" + assert response.info.search_domains == ["lan"] + + +async def test_set_dns_manual_sends_servers() -> None: + """Test manual DNS mode sends the configured servers.""" + async with NanoKVMClient( + "http://localhost:8888/api/", token="test-token" + ) as client: + _mark_detected(client, application_version="2.4.1") + + with aioresponses() as m: + m.post( + "http://localhost:8888/api/network/dns", + payload={"code": 0, "msg": "success", "data": None}, + ) + + await client.set_dns( + DNSMode.MANUAL, + ["1.1.1.1", "2606:4700:4700::1111"], + ) + + calls = m.requests[ + ("POST", yarl.URL("http://localhost:8888/api/network/dns")) + ] + assert calls[0].kwargs.get("json") == { + "mode": "manual", + "servers": ["1.1.1.1", "2606:4700:4700::1111"], + } + + +async def test_set_dns_dhcp_string_sends_empty_servers() -> None: + """Test DHCP DNS mode accepts string mode and sends empty servers.""" + async with NanoKVMClient( + "http://localhost:8888/api/", token="test-token" + ) as client: + _mark_detected(client, application_version="2.4.1") + + with aioresponses() as m: + m.post( + "http://localhost:8888/api/network/dns", + payload={"code": 0, "msg": "success", "data": None}, + ) + + await client.set_dns("dhcp") + + calls = m.requests[ + ("POST", yarl.URL("http://localhost:8888/api/network/dns")) + ] + assert calls[0].kwargs.get("json") == { + "mode": "dhcp", + "servers": [], + } + + +async def test_get_dns_pro_is_not_supported() -> None: + """Test DNS management is limited to non-Pro hardware.""" + async with NanoKVMClient( + "http://localhost:8888/api/", token="test-token" + ) as client: + client._hw_version = HWVersion.PRO + + with aioresponses() as m: + with pytest.raises(NanoKVMNotSupportedError) as exc_info: + await client.get_dns() + + assert "get_dns requires hardware: Alpha, Beta, PCIE" in str(exc_info.value) + assert not m.requests + + +async def test_get_dns_old_application_version_is_not_supported() -> None: + """Test old non-Pro application versions reject DNS before endpoint call.""" + async with NanoKVMClient( + "http://localhost:8888/api/", token="test-token" + ) as client: + client._hw_version = HWVersion.PCIE + + with aioresponses() as m: + m.get( + "http://localhost:8888/api/vm/info", + payload=_info_payload(application="2.4.0"), + ) + + with pytest.raises(NanoKVMNotSupportedError) as exc_info: + await client.get_dns() + + assert "get_dns requires non-Pro application version >= 2.4.1" in str( + exc_info.value + ) + dns_url = yarl.URL("http://localhost:8888/api/network/dns") + assert ("GET", dns_url) not in m.requests + + +async def test_get_dns_exact_application_version_is_supported() -> None: + """Test DNS is allowed at the first upstream version that introduced it.""" + async with NanoKVMClient( + "http://localhost:8888/api/", token="test-token" + ) as client: + client._hw_version = HWVersion.PCIE + + with aioresponses() as m: + m.get( + "http://localhost:8888/api/vm/info", + payload=_info_payload(application="2.4.1"), + ) + m.get( + "http://localhost:8888/api/network/dns", + payload={ + "code": 0, + "msg": "success", + "data": { + "mode": "dhcp", + "servers": [], + "effective": [], + "dhcp": [], + "info": {}, + }, + }, + ) + + response = await client.get_dns() + + assert response.mode is DNSMode.DHCP + + 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(