diff --git a/aiohasupervisor/models/__init__.py b/aiohasupervisor/models/__init__.py index 6d4ec72..fc13a2e 100644 --- a/aiohasupervisor/models/__init__.py +++ b/aiohasupervisor/models/__init__.py @@ -53,6 +53,25 @@ HomeAssistantStopOptions, HomeAssistantUpdateOptions, ) +from aiohasupervisor.models.network import ( + AccessPoint, + AuthMethod, + DockerNetwork, + InterfaceMethod, + InterfaceType, + IPv4, + IPv4Config, + IPv6, + IPv6Config, + NetworkInfo, + NetworkInterface, + NetworkInterfaceConfig, + Vlan, + VlanConfig, + Wifi, + WifiConfig, + WifiMode, +) from aiohasupervisor.models.os import ( BootSlot, BootSlotName, @@ -176,4 +195,21 @@ "PartialRestoreOptions", "Discovery", "DiscoveryConfig", + "AccessPoint", + "AuthMethod", + "DockerNetwork", + "InterfaceMethod", + "InterfaceType", + "IPv4", + "IPv4Config", + "IPv6", + "IPv6Config", + "NetworkInfo", + "NetworkInterface", + "NetworkInterfaceConfig", + "Vlan", + "VlanConfig", + "Wifi", + "WifiConfig", + "WifiMode", ] diff --git a/aiohasupervisor/models/network.py b/aiohasupervisor/models/network.py new file mode 100644 index 0000000..b63fac8 --- /dev/null +++ b/aiohasupervisor/models/network.py @@ -0,0 +1,198 @@ +"""Models for supervisor network.""" + +from abc import ABC +from dataclasses import dataclass +from enum import StrEnum +from ipaddress import ( + IPv4Address, + IPv4Interface, + IPv4Network, + IPv6Address, + IPv6Interface, +) + +from .base import Options, Request, ResponseData + +# --- ENUMS ---- + + +class InterfaceType(StrEnum): + """InterfaceType type.""" + + ETHERNET = "ethernet" + WIRELESS = "wireless" + VLAN = "vlan" + + +class InterfaceMethod(StrEnum): + """InterfaceMethod type.""" + + DISABLED = "disabled" + STATIC = "static" + AUTO = "auto" + + +class WifiMode(StrEnum): + """WifiMode type.""" + + INFRASTRUCTURE = "infrastructure" + MESH = "mesh" + ADHOC = "adhoc" + AP = "ap" + + +class AuthMethod(StrEnum): + """AuthMethod type.""" + + OPEN = "open" + WEP = "wep" + WPA_PSK = "wpa-psk" + + +# --- OBJECTS ---- + + +@dataclass(frozen=True) +class IpBase(ABC): + """IpBase ABC type.""" + + method: InterfaceMethod + ready: bool | None + + +@dataclass(frozen=True, slots=True) +class IPv4(IpBase, ResponseData): + """IPv4 model.""" + + address: list[IPv4Interface] + nameservers: list[IPv4Address] + gateway: IPv4Address | None + + +@dataclass(frozen=True, slots=True) +class IPv6(IpBase, ResponseData): + """IPv6 model.""" + + address: list[IPv6Interface] + nameservers: list[IPv6Address] + gateway: IPv6Address | None + + +@dataclass(frozen=True, slots=True) +class Wifi(ResponseData): + """Wifi model.""" + + mode: WifiMode + auth: AuthMethod + ssid: str + signal: int | None + + +@dataclass(frozen=True, slots=True) +class Vlan(ResponseData): + """Vlan model.""" + + id: int + interface: str + + +@dataclass(frozen=True, slots=True) +class NetworkInterface(ResponseData): + """NetworkInterface model.""" + + interface: str + type: InterfaceType + enabled: bool + connected: bool + primary: bool + mac: str + ipv4: IPv4 + ipv6: IPv6 + wifi: Wifi | None + vlan: Vlan | None + + +@dataclass(frozen=True, slots=True) +class DockerNetwork(ResponseData): + """DockerNetwork model.""" + + interface: str + address: IPv4Network + gateway: IPv4Address + dns: IPv4Address + + +@dataclass(frozen=True, slots=True) +class NetworkInfo(ResponseData): + """NetworkInfo model.""" + + interfaces: list[NetworkInterface] + docker: DockerNetwork + host_internet: bool | None + supervisor_internet: bool + + +@dataclass(frozen=True, slots=True) +class IPv4Config(Request): + """IPv4Config model.""" + + address: list[IPv4Interface] | None = None + method: InterfaceMethod | None = None + gateway: IPv4Address | None = None + nameservers: list[IPv4Address] | None = None + + +@dataclass(frozen=True, slots=True) +class IPv6Config(Request): + """IPv6Config model.""" + + address: list[IPv6Interface] | None = None + method: InterfaceMethod | None = None + gateway: IPv6Address | None = None + nameservers: list[IPv6Address] | None = None + + +@dataclass(frozen=True, slots=True) +class WifiConfig(Request): + """WifiConfig model.""" + + mode: WifiMode | None = None + method: AuthMethod | None = None + ssid: str | None = None + psk: str | None = None + + +@dataclass(frozen=True, slots=True) +class NetworkInterfaceConfig(Options): + """NetworkInterfaceConfig model.""" + + ipv4: IPv4Config | None = None + ipv6: IPv6Config | None = None + wifi: WifiConfig | None = None + enabled: bool | None = None + + +@dataclass(frozen=True, slots=True) +class AccessPoint(ResponseData): + """AccessPoint model.""" + + mode: WifiMode + ssid: str + frequency: int + signal: int + mac: str + + +@dataclass(frozen=True, slots=True) +class AccessPointList(ResponseData): + """AccessPointList model.""" + + accesspoints: list[AccessPoint] + + +@dataclass(frozen=True, slots=True) +class VlanConfig(Options): + """VlanConfig model.""" + + ipv4: IPv4Config | None = None + ipv6: IPv6Config | None = None diff --git a/aiohasupervisor/network.py b/aiohasupervisor/network.py new file mode 100644 index 0000000..bf9b464 --- /dev/null +++ b/aiohasupervisor/network.py @@ -0,0 +1,51 @@ +"""Network client for supervisor.""" + +from .client import _SupervisorComponentClient +from .models.network import ( + AccessPoint, + AccessPointList, + NetworkInfo, + NetworkInterface, + NetworkInterfaceConfig, + VlanConfig, +) + + +class NetworkClient(_SupervisorComponentClient): + """Handles network access in supervisor.""" + + async def info(self) -> NetworkInfo: + """Get network info.""" + result = await self._client.get("network/info") + return NetworkInfo.from_dict(result.data) + + async def reload(self) -> None: + """Reload network info caches.""" + await self._client.post("network/reload") + + async def interface_info(self, interface: str) -> NetworkInterface: + """Get network interface info.""" + result = await self._client.get(f"network/interface/{interface}/info") + return NetworkInterface.from_dict(result.data) + + async def update_interface( + self, interface: str, config: NetworkInterfaceConfig + ) -> None: + """Update a network interface.""" + await self._client.post( + f"network/interface/{interface}/update", json=config.to_dict() + ) + + async def access_points(self, interface: str) -> list[AccessPoint]: + """Get access points visible to a wireless interface.""" + result = await self._client.get(f"network/interface/{interface}/accesspoints") + return AccessPointList.from_dict(result.data).accesspoints + + async def save_vlan( + self, interface: str, vlan: int, config: VlanConfig | None = None + ) -> None: + """Create or update a vlan for an ethernet interface.""" + await self._client.post( + f"network/interface/{interface}/vlan/{vlan}", + json=config.to_dict() if config else None, + ) diff --git a/aiohasupervisor/root.py b/aiohasupervisor/root.py index 45052dc..5fef24a 100644 --- a/aiohasupervisor/root.py +++ b/aiohasupervisor/root.py @@ -10,6 +10,7 @@ from .discovery import DiscoveryClient from .homeassistant import HomeAssistantClient from .models.root import AvailableUpdate, AvailableUpdates, RootInfo +from .network import NetworkClient from .os import OSClient from .resolution import ResolutionClient from .store import StoreClient @@ -32,6 +33,7 @@ def __init__( self._os = OSClient(self._client) self._backups = BackupsClient(self._client) self._discovery = DiscoveryClient(self._client) + self._network = NetworkClient(self._client) self._resolution = ResolutionClient(self._client) self._store = StoreClient(self._client) self._supervisor = SupervisorManagementClient(self._client) @@ -62,6 +64,11 @@ def discovery(self) -> DiscoveryClient: """Get discovery component client.""" return self._discovery + @property + def network(self) -> NetworkClient: + """Get network component client.""" + return self._network + @property def resolution(self) -> ResolutionClient: """Get resolution center component client.""" diff --git a/tests/fixtures/network_access_points.json b/tests/fixtures/network_access_points.json new file mode 100644 index 0000000..49145a1 --- /dev/null +++ b/tests/fixtures/network_access_points.json @@ -0,0 +1,21 @@ +{ + "result": "ok", + "data": { + "accesspoints": [ + { + "mode": "infrastructure", + "ssid": "UPC4814466", + "frequency": 2462, + "signal": 47, + "mac": "AA:BB:CC:DD:EE:FF" + }, + { + "mode": "infrastructure", + "ssid": "VQ@35(55720", + "frequency": 5660, + "signal": 63, + "mac": "FF:EE:DD:CC:BB:AA" + } + ] + } +} diff --git a/tests/fixtures/network_info.json b/tests/fixtures/network_info.json new file mode 100644 index 0000000..eec208e --- /dev/null +++ b/tests/fixtures/network_info.json @@ -0,0 +1,39 @@ +{ + "result": "ok", + "data": { + "interfaces": [ + { + "interface": "end0", + "type": "ethernet", + "enabled": true, + "connected": true, + "primary": true, + "mac": "00:11:22:33:44:55", + "ipv4": { + "method": "static", + "address": ["192.168.1.2/24"], + "nameservers": ["192.168.1.1"], + "gateway": "192.168.1.1", + "ready": true + }, + "ipv6": { + "method": "disabled", + "address": ["fe80::819d:c479:d712:7a77/64"], + "nameservers": [], + "gateway": null, + "ready": true + }, + "wifi": null, + "vlan": null + } + ], + "docker": { + "interface": "hassio", + "address": "172.30.32.0/23", + "gateway": "172.30.32.1", + "dns": "172.30.32.3" + }, + "host_internet": true, + "supervisor_internet": true + } +} diff --git a/tests/fixtures/network_interface_info.json b/tests/fixtures/network_interface_info.json new file mode 100644 index 0000000..1ba70d1 --- /dev/null +++ b/tests/fixtures/network_interface_info.json @@ -0,0 +1,27 @@ +{ + "result": "ok", + "data": { + "interface": "end0", + "type": "ethernet", + "enabled": true, + "connected": true, + "primary": true, + "mac": "00:11:22:33:44:55", + "ipv4": { + "method": "static", + "address": ["192.168.1.2/24"], + "nameservers": ["192.168.1.1"], + "gateway": "192.168.1.1", + "ready": true + }, + "ipv6": { + "method": "disabled", + "address": ["fe80::819d:c479:d712:7a77/64"], + "nameservers": [], + "gateway": null, + "ready": true + }, + "wifi": null, + "vlan": null + } +} diff --git a/tests/test_network.py b/tests/test_network.py new file mode 100644 index 0000000..424d128 --- /dev/null +++ b/tests/test_network.py @@ -0,0 +1,158 @@ +"""Test network supervisor client.""" + +from ipaddress import IPv4Address, IPv4Interface + +from aioresponses import aioresponses +import pytest +from yarl import URL + +from aiohasupervisor import SupervisorClient +from aiohasupervisor.models import ( + InterfaceMethod, + IPv4Config, + NetworkInterfaceConfig, + VlanConfig, +) + +from . import load_fixture +from .const import SUPERVISOR_URL + + +async def test_network_info( + responses: aioresponses, supervisor_client: SupervisorClient +) -> None: + """Test network info API.""" + responses.get( + f"{SUPERVISOR_URL}/network/info", + status=200, + body=load_fixture("network_info.json"), + ) + result = await supervisor_client.network.info() + assert result.interfaces[0].interface == "end0" + assert result.interfaces[0].type == "ethernet" + assert result.interfaces[0].enabled is True + assert result.interfaces[0].mac == "00:11:22:33:44:55" + assert result.interfaces[0].ipv4.method == "static" + assert result.interfaces[0].ipv4.address[0].with_prefixlen == "192.168.1.2/24" + assert result.interfaces[0].ipv4.nameservers[0].compressed == "192.168.1.1" + assert result.interfaces[0].ipv4.gateway.compressed == "192.168.1.1" + assert result.interfaces[0].ipv4.ready is True + assert result.interfaces[0].ipv6.method == "disabled" + assert ( + result.interfaces[0].ipv6.address[0].with_prefixlen + == "fe80::819d:c479:d712:7a77/64" + ) + assert result.interfaces[0].ipv6.gateway is None + assert result.interfaces[0].wifi is None + assert result.interfaces[0].vlan is None + + assert result.docker.interface == "hassio" + assert result.docker.address.compressed == "172.30.32.0/23" + assert result.docker.gateway.compressed == "172.30.32.1" + assert result.docker.dns.compressed == "172.30.32.3" + assert result.host_internet is True + assert result.supervisor_internet is True + + +async def test_network_reload( + responses: aioresponses, supervisor_client: SupervisorClient +) -> None: + """Test network reload API.""" + responses.post(f"{SUPERVISOR_URL}/network/reload", status=200) + assert await supervisor_client.network.reload() is None + assert responses.requests.keys() == { + ("POST", URL(f"{SUPERVISOR_URL}/network/reload")) + } + + +async def test_network_interface_info( + responses: aioresponses, supervisor_client: SupervisorClient +) -> None: + """Test network interface info API.""" + responses.get( + f"{SUPERVISOR_URL}/network/interface/end0/info", + status=200, + body=load_fixture("network_interface_info.json"), + ) + result = await supervisor_client.network.interface_info("end0") + assert result.interface == "end0" + assert result.type == "ethernet" + assert result.enabled is True + assert result.mac == "00:11:22:33:44:55" + assert result.ipv4.method == "static" + assert result.ipv4.address[0].with_prefixlen == "192.168.1.2/24" + assert result.ipv4.nameservers[0].compressed == "192.168.1.1" + assert result.ipv4.gateway.compressed == "192.168.1.1" + assert result.ipv4.ready is True + assert result.ipv6.method == "disabled" + assert result.ipv6.address[0].with_prefixlen == "fe80::819d:c479:d712:7a77/64" + assert result.ipv6.gateway is None + assert result.wifi is None + assert result.vlan is None + + +async def test_network_update_interface( + responses: aioresponses, supervisor_client: SupervisorClient +) -> None: + """Test network interface update API.""" + responses.post(f"{SUPERVISOR_URL}/network/interface/end0/update", status=200) + config = NetworkInterfaceConfig( + ipv4=IPv4Config( + method=InterfaceMethod.STATIC, + address=[IPv4Interface("192.168.1.2/24")], + gateway=IPv4Address("192.168.1.1"), + nameservers=[IPv4Address("192.168.1.1")], + ) + ) + assert ( + await supervisor_client.network.update_interface("end0", config=config) is None + ) + assert responses.requests.keys() == { + ("POST", URL(f"{SUPERVISOR_URL}/network/interface/end0/update")) + } + + +async def test_network_access_points( + responses: aioresponses, supervisor_client: SupervisorClient +) -> None: + """Test network access points API.""" + responses.get( + f"{SUPERVISOR_URL}/network/interface/end0/accesspoints", + status=200, + body=load_fixture("network_access_points.json"), + ) + result = await supervisor_client.network.access_points("end0") + assert result[0].mode == "infrastructure" + assert result[0].ssid == "UPC4814466" + assert result[0].frequency == 2462 + assert result[0].signal == 47 + assert result[0].mac == "AA:BB:CC:DD:EE:FF" + assert result[1].ssid == "VQ@35(55720" + + +@pytest.mark.parametrize( + "config", + [None, NetworkInterfaceConfig(ipv4=IPv4Config(method=InterfaceMethod.AUTO))], +) +async def test_network_save_vlan( + responses: aioresponses, + supervisor_client: SupervisorClient, + config: NetworkInterfaceConfig | None, +) -> None: + """Test network save vlan API.""" + responses.post(f"{SUPERVISOR_URL}/network/interface/end0/vlan/1", status=200) + assert await supervisor_client.network.save_vlan("end0", 1, config=config) is None + assert responses.requests.keys() == { + ("POST", URL(f"{SUPERVISOR_URL}/network/interface/end0/vlan/1")) + } + + +async def test_network_configs_cannot_be_empty() -> None: + """Test network config instances require at least one field specified.""" + # Network interface config for update calls + with pytest.raises(ValueError, match="At least one field must have a value"): + NetworkInterfaceConfig() + + # Vlan config for save vlan calls + with pytest.raises(ValueError, match="At least one field must have a value"): + VlanConfig()