Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
25 changes: 25 additions & 0 deletions nanokvm/client.py
Original file line number Diff line number Diff line change
Expand Up @@ -93,11 +93,14 @@
WakeOnLANReq,
)
from .models.non_pro import (
DNSMode,
GetCdRomRsp,
GetDNSRsp,
GetHdmiStateRsp,
GetMemoryLimitRsp,
GetSwapSizeRsp,
ScreenSettingType,
SetDNSReq,
SetMemoryLimitReq,
SetScreenReq,
SetSwapSizeReq,
Expand Down Expand Up @@ -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(
Expand Down
50 changes: 49 additions & 1 deletion nanokvm/models/non_pro.py
Original file line number Diff line number Diff line change
Expand Up @@ -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):
Expand All @@ -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."""

Expand Down Expand Up @@ -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)
166 changes: 166 additions & 0 deletions tests/test_client.py
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@
from nanokvm.models import (
ApiResponseCode,
DiskType,
DNSMode,
GetMacRsp,
GetOLEDRsp,
HWVersion,
Expand Down Expand Up @@ -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(
Expand Down
Loading