Add Host APIs from supervisor

Author: mdegat01

aiohasupervisor/host.py

@@ -0,0 +1,40 @@
+"""Host client for supervisor."""
+
+from .client import _SupervisorComponentClient
+from .models.host import HostInfo, HostOptions, Service, ServiceList, ShutdownOptions
+
+
+class HostClient(_SupervisorComponentClient):
+    """Handles host access in supervisor."""
+
+    async def info(self) -> HostInfo:
+        """Get host info."""
+        result = await self._client.get("host/info")
+        return HostInfo.from_dict(result.data)
+
+    async def reboot(self, options: ShutdownOptions | None = None) -> None:
+        """Reboot host."""
+        await self._client.post(
+            "host/reboot", json=options.to_dict() if options else None
+        )
+
+    async def shutdown(self, options: ShutdownOptions | None = None) -> None:
+        """Shutdown host."""
+        await self._client.post(
+            "host/shutdown", json=options.to_dict() if options else None
+        )
+
+    async def reload(self) -> None:
+        """Reload host info cache."""
+        await self._client.post("host/reload")
+
+    async def options(self, options: HostOptions) -> None:
+        """Set host options."""
+        await self._client.post("host/options", json=options.to_dict())
+
+    async def services(self) -> list[Service]:
+        """Get list of available services on host."""
+        result = await self._client.get("host/services")
+        return ServiceList.from_dict(result.data).services
+
+    # Omitted for now - Log endpoints

aiohasupervisor/models/__init__.py

@@ -23,6 +23,13 @@
     StoreInfo,
     SupervisorRole,
 )
+from aiohasupervisor.models.host import (
+    HostInfo,
+    HostOptions,
+    Service,
+    ServiceState,
+    ShutdownOptions,
+)
 from aiohasupervisor.models.resolution import (
     Check,
     CheckOptions,
@@ -86,4 +93,9 @@
     "SuggestionType",
     "UnhealthyReason",
     "UnsupportedReason",
+    "HostInfo",
+    "HostOptions",
+    "Service",
+    "ServiceState",
+    "ShutdownOptions",
 ]

aiohasupervisor/models/host.py

@@ -0,0 +1,91 @@
+"""Models for host APIs."""
+
+from dataclasses import dataclass
+from datetime import datetime
+from enum import StrEnum
+
+from .base import Request, ResponseData
+from .root import HostFeature
+
+# --- ENUMS ----
+
+
+class ServiceState(StrEnum):
+    """ServiceState type.
+
+    The service state is determined by systemd, not supervisor. The list below
+    is pulled from `systemctl --state=help`. It may be incomplete and it may
+    change based on the host. Therefore within a list of services there may be
+    some with a state not in this list parsed as string. If you find this
+    please create an issue or pr to get the state added.
+    """
+
+    ACTIVE = "active"
+    RELOADING = "reloading"
+    INACTIVE = "inactive"
+    FAILED = "failed"
+    ACTIVATING = "activating"
+    DEACTIVATING = "deactivating"
+    MAINTENANCE = "maintenance"
+
+
+# --- OBJECTS ----
+
+
+@dataclass(frozen=True, slots=True)
+class HostInfo(ResponseData):
+    """HostInfo model."""
+
+    agent_version: str | None
+    apparmor_version: str | None
+    chassis: str | None
+    virtualization: str | None
+    cpe: str | None
+    deployment: str | None
+    disk_free: float
+    disk_total: float
+    disk_used: float
+    disk_life_time: float
+    features: list[HostFeature]
+    hostname: str | None
+    llmnr_hostname: str | None
+    kernel: str | None
+    operating_system: str | None
+    timezone: str | None
+    dt_utc: datetime | None
+    dt_synchronized: bool | None
+    use_ntp: bool | None
+    startup_time: float | None
+    boot_timestamp: int | None
+    broadcast_llmnr: bool | None
+    broadcast_mdns: bool | None
+
+
+@dataclass(frozen=True, slots=True)
+class ShutdownOptions(Request):
+    """ShutdownOptions model."""
+
+    force: bool
+
+
+@dataclass(frozen=True, slots=True)
+class HostOptions(Request):
+    """HostOptions model."""
+
+    hostname: str
+
+
+@dataclass(frozen=True, slots=True)
+class Service(ResponseData):
+    """Service model."""
+
+    name: str
+    description: str
+    state: ServiceState | str
+
+
+@dataclass(frozen=True, slots=True)
+class ServiceList(ResponseData):
+    """ServiceList model."""
+
+    services: list[Service]

aiohasupervisor/root.py

@@ -6,6 +6,7 @@
 
 from .addons import AddonsClient
 from .client import _SupervisorClient
+from .host import HostClient
 from .models.root import AvailableUpdate, AvailableUpdates, RootInfo
 from .resolution import ResolutionClient
 from .store import StoreClient
@@ -24,6 +25,7 @@ def __init__(
         """Initialize client."""
         self._client = _SupervisorClient(api_host, token, request_timeout, session)
         self._addons = AddonsClient(self._client)
+        self._host = HostClient(self._client)
         self._resolution = ResolutionClient(self._client)
         self._store = StoreClient(self._client)
 
@@ -32,6 +34,11 @@ def addons(self) -> AddonsClient:
         """Get addons component client."""
         return self._addons
 
+    @property
+    def host(self) -> HostClient:
+        """Get host component client."""
+        return self._host
+
     @property
     def resolution(self) -> ResolutionClient:
         """Get resolution center component client."""

tests/fixtures/host_info.json

@@ -0,0 +1,41 @@
+{
+  "result": "ok",
+  "data": {
+    "agent_version": "1.6.0",
+    "apparmor_version": "3.1.2",
+    "chassis": "embedded",
+    "virtualization": "",
+    "cpe": "cpe:2.3:o:home-assistant:haos:12.4.dev20240527:*:development:*:*:*:odroid-n2:*",
+    "deployment": "development",
+    "disk_free": 20.1,
+    "disk_total": 27.9,
+    "disk_used": 6.7,
+    "disk_life_time": 10.0,
+    "features": [
+      "reboot",
+      "shutdown",
+      "services",
+      "network",
+      "hostname",
+      "timedate",
+      "os_agent",
+      "haos",
+      "resolved",
+      "journal",
+      "disk",
+      "mount"
+    ],
+    "hostname": "homeassistant",
+    "llmnr_hostname": "homeassistant3",
+    "kernel": "6.6.32-haos",
+    "operating_system": "Home Assistant OS 12.4.dev20240527",
+    "timezone": "Etc/UTC",
+    "dt_utc": "2024-10-03T00:00:00.000000+00:00",
+    "dt_synchronized": true,
+    "use_ntp": true,
+    "startup_time": 1.966311,
+    "boot_timestamp": 1716927644219811,
+    "broadcast_llmnr": true,
+    "broadcast_mdns": true
+  }
+}

tests/fixtures/host_services.json

@@ -0,0 +1,47 @@
+{
+  "result": "ok",
+  "data": {
+    "services": [
+      {
+        "name": "emergency.service",
+        "description": "Emergency Shell",
+        "state": "inactive"
+      },
+      {
+        "name": "bluetooth.service",
+        "description": "Bluetooth service",
+        "state": "inactive"
+      },
+      {
+        "name": "haos-swapfile.service",
+        "description": "HAOS swap",
+        "state": "inactive"
+      },
+      {
+        "name": "hassos-config.service",
+        "description": "HassOS Configuration Manager",
+        "state": "inactive"
+      },
+      {
+        "name": "dropbear.service",
+        "description": "Dropbear SSH daemon",
+        "state": "active"
+      },
+      {
+        "name": "systemd-time-wait-sync.service",
+        "description": "Wait Until Kernel Time Synchronized",
+        "state": "active"
+      },
+      {
+        "name": "systemd-journald.service",
+        "description": "Journal Service",
+        "state": "active"
+      },
+      {
+        "name": "systemd-resolved.service",
+        "description": "Network Name Resolution",
+        "state": "active"
+      }
+    ]
+  }
+}

tests/test_host.py

@@ -0,0 +1,104 @@
+"""Test host supervisor client."""
+
+from datetime import UTC, datetime
+
+from aioresponses import aioresponses
+from yarl import URL
+
+from aiohasupervisor import SupervisorClient
+from aiohasupervisor.models import HostOptions
+
+from . import load_fixture
+from .const import SUPERVISOR_URL
+
+
+async def test_host_info(
+    responses: aioresponses, supervisor_client: SupervisorClient
+) -> None:
+    """Test host info API."""
+    responses.get(
+        f"{SUPERVISOR_URL}/host/info", status=200, body=load_fixture("host_info.json")
+    )
+    result = await supervisor_client.host.info()
+    assert result.agent_version == "1.6.0"
+    assert result.chassis == "embedded"
+    assert result.virtualization == ""
+    assert result.disk_total == 27.9
+    assert result.disk_life_time == 10
+    assert result.features == [
+        "reboot",
+        "shutdown",
+        "services",
+        "network",
+        "hostname",
+        "timedate",
+        "os_agent",
+        "haos",
+        "resolved",
+        "journal",
+        "disk",
+        "mount",
+    ]
+    assert result.hostname == "homeassistant"
+    assert result.llmnr_hostname == "homeassistant3"
+    assert result.dt_utc == datetime(2024, 10, 3, 0, 0, 0, 0, UTC)
+    assert result.dt_synchronized is True
+    assert result.startup_time == 1.966311
+
+
+async def test_host_reboot(
+    responses: aioresponses, supervisor_client: SupervisorClient
+) -> None:
+    """Test host reboot API."""
+    responses.post(f"{SUPERVISOR_URL}/host/reboot", status=200)
+    assert await supervisor_client.host.reboot() is None
+    assert responses.requests.keys() == {("POST", URL(f"{SUPERVISOR_URL}/host/reboot"))}
+
+
+async def test_host_shutdown(
+    responses: aioresponses, supervisor_client: SupervisorClient
+) -> None:
+    """Test host shutdown API."""
+    responses.post(f"{SUPERVISOR_URL}/host/shutdown", status=200)
+    assert await supervisor_client.host.shutdown() is None
+    assert responses.requests.keys() == {
+        ("POST", URL(f"{SUPERVISOR_URL}/host/shutdown"))
+    }
+
+
+async def test_host_reload(
+    responses: aioresponses, supervisor_client: SupervisorClient
+) -> None:
+    """Test host reload API."""
+    responses.post(f"{SUPERVISOR_URL}/host/reload", status=200)
+    assert await supervisor_client.host.reload() is None
+    assert responses.requests.keys() == {("POST", URL(f"{SUPERVISOR_URL}/host/reload"))}
+
+
+async def test_host_options(
+    responses: aioresponses, supervisor_client: SupervisorClient
+) -> None:
+    """Test host options API."""
+    responses.post(f"{SUPERVISOR_URL}/host/options", status=200)
+    assert await supervisor_client.host.options(HostOptions(hostname="test")) is None
+    assert responses.requests.keys() == {
+        ("POST", URL(f"{SUPERVISOR_URL}/host/options"))
+    }
+
+
+async def test_host_services(
+    responses: aioresponses, supervisor_client: SupervisorClient
+) -> None:
+    """Test host services API."""
+    responses.get(
+        f"{SUPERVISOR_URL}/host/services",
+        status=200,
+        body=load_fixture("host_services.json"),
+    )
+    result = await supervisor_client.host.services()
+    assert result[0].name == "emergency.service"
+    assert result[0].description == "Emergency Shell"
+    assert result[0].state == "inactive"
+    assert result[-1].name == "systemd-resolved.service"
+    assert result[-1].description == "Network Name Resolution"
+    assert result[-1].state == "active"