Add supervisor APIs to client library

Author: mdegat01

aiohasupervisor/models/__init__.py

@@ -45,6 +45,12 @@
     UpdateChannel,
     UpdateType,
 )
+from aiohasupervisor.models.supervisor import (
+    SupervisorInfo,
+    SupervisorOptions,
+    SupervisorStats,
+    SupervisorUpdateOptions,
+)
 
 __all__ = [
     "HostFeature",
@@ -86,4 +92,8 @@
     "SuggestionType",
     "UnhealthyReason",
     "UnsupportedReason",
+    "SupervisorInfo",
+    "SupervisorOptions",
+    "SupervisorStats",
+    "SupervisorUpdateOptions",
 ]

aiohasupervisor/models/supervisor.py

@@ -0,0 +1,62 @@
+"""Models for supervisor component."""
+
+from dataclasses import dataclass
+
+from .base import Options, Request, ResponseData
+from .root import LogLevel, UpdateChannel
+
+
+@dataclass(frozen=True, slots=True)
+class SupervisorInfo(ResponseData):
+    """SupervisorInfo model."""
+
+    version: str
+    version_latest: str
+    update_available: bool
+    channel: UpdateChannel
+    arch: str
+    supported: bool
+    healthy: bool
+    ip_address: str
+    timezone: str | None
+    logging: LogLevel
+    debug: bool
+    debug_block: bool
+    diagnostics: bool | None
+    auto_update: bool
+
+
+@dataclass(frozen=True, slots=True)
+class SupervisorStats(ResponseData):
+    """SupervisorStats model."""
+
+    cpu_percent: float
+    memory_usage: int
+    memory_limit: int
+    memory_percent: float
+    network_rx: int
+    network_tx: int
+    blk_read: int
+    blk_write: int
+
+
+@dataclass(frozen=True, slots=True)
+class SupervisorUpdateOptions(Request):
+    """SupervisorUpdateOptions model."""
+
+    version: str
+
+
+@dataclass(frozen=True, slots=True)
+class SupervisorOptions(Options):
+    """SupervisorOptions model."""
+
+    channel: UpdateChannel | None = None
+    timezone: str | None = None
+    logging: LogLevel | None = None
+    debug: bool | None = None
+    debug_block: bool | None = None
+    diagnostics: bool | None = None
+    content_trust: bool | None = None
+    force_security: bool | None = None
+    auto_update: bool | None = None

aiohasupervisor/root.py

@@ -9,6 +9,7 @@
 from .models.root import AvailableUpdate, AvailableUpdates, RootInfo
 from .resolution import ResolutionClient
 from .store import StoreClient
+from .supervisor import SupervisorManagementClient
 
 
 class SupervisorClient:
@@ -26,6 +27,7 @@ def __init__(
         self._addons = AddonsClient(self._client)
         self._resolution = ResolutionClient(self._client)
         self._store = StoreClient(self._client)
+        self._supervisor = SupervisorManagementClient(self._client)
 
     @property
     def addons(self) -> AddonsClient:
@@ -42,6 +44,11 @@ def store(self) -> StoreClient:
         """Get store component client."""
         return self._store
 
+    @property
+    def supervisor(self) -> SupervisorManagementClient:
+        """Get supervisor component client."""
+        return self._supervisor
+
     async def info(self) -> RootInfo:
         """Get root info."""
         result = await self._client.get("info")

aiohasupervisor/supervisor.py

@@ -0,0 +1,55 @@
+"""Supervisor client for supervisor."""
+
+from .client import _SupervisorComponentClient
+from .const import ResponseType
+from .models.supervisor import (
+    SupervisorInfo,
+    SupervisorOptions,
+    SupervisorStats,
+    SupervisorUpdateOptions,
+)
+
+
+class SupervisorManagementClient(_SupervisorComponentClient):
+    """Handles supervisor access in supervisor."""
+
+    async def ping(self) -> None:
+        """Check connection to supervisor."""
+        await self._client.get("supervisor/ping", response_type=ResponseType.NONE)
+
+    async def info(self) -> SupervisorInfo:
+        """Get supervisor info."""
+        result = await self._client.get("supervisor/info")
+        return SupervisorInfo.from_dict(result.data)
+
+    async def stats(self) -> SupervisorStats:
+        """Get supervisor stats."""
+        result = await self._client.get("supervisor/stats")
+        return SupervisorStats.from_dict(result.data)
+
+    async def update(self, options: SupervisorUpdateOptions | None = None) -> None:
+        """Update supervisor.
+
+        Providing a target version in options only works on development systems.
+        On non-development systems this API will always update supervisor to the
+        latest version and ignore that field.
+        """
+        await self._client.post(
+            "supervisor/update", json=options.to_dict() if options else None
+        )
+
+    async def reload(self) -> None:
+        """Reload supervisor (add-ons, configuration, etc)."""
+        await self._client.post("supervisor/reload")
+
+    async def restart(self) -> None:
+        """Restart supervisor."""
+        await self._client.post("supervisor/restart")
+
+    async def options(self, options: SupervisorOptions) -> None:
+        """Set supervisor options."""
+        await self._client.post("supervisor/options", json=options.to_dict())
+
+    async def repair(self) -> None:
+        """Repair local supervisor and docker setup."""
+        await self._client.post("supervisor/repair")

tests/fixtures/supervisor_info.json

@@ -0,0 +1,49 @@
+{
+  "result": "ok",
+  "data": {
+    "version": "2024.09.1",
+    "version_latest": "2024.09.1",
+    "update_available": true,
+    "channel": "stable",
+    "arch": "aarch64",
+    "supported": true,
+    "healthy": true,
+    "ip_address": "172.30.32.2",
+    "timezone": "America/New_York",
+    "logging": "info",
+    "debug": true,
+    "debug_block": false,
+    "diagnostics": false,
+    "auto_update": true,
+    "wait_boot": 5,
+    "addons": [
+      {
+        "name": "Terminal & SSH",
+        "slug": "core_ssh",
+        "version": "9.14.0",
+        "version_latest": "9.14.0",
+        "update_available": false,
+        "state": "started",
+        "repository": "core",
+        "icon": true
+      },
+      {
+        "name": "Mosquitto broker",
+        "slug": "core_mosquitto",
+        "version": "6.4.1",
+        "version_latest": "6.4.1",
+        "update_available": false,
+        "state": "started",
+        "repository": "core",
+        "icon": true
+      }
+    ],
+    "addons_repositories": [
+      { "name": "Local add-ons", "slug": "local" },
+      { "name": "Music Assistant", "slug": "d5369777" },
+      { "name": "Official add-ons", "slug": "core" },
+      { "name": "ESPHome", "slug": "5c53de3b" },
+      { "name": "Home Assistant Community Add-ons", "slug": "a0d7b954" }
+    ]
+  }
+}

tests/fixtures/supervisor_stats.json

@@ -0,0 +1,13 @@
+{
+  "result": "ok",
+  "data": {
+    "cpu_percent": 0.04,
+    "memory_usage": 243982336,
+    "memory_limit": 3899138048,
+    "memory_percent": 6.26,
+    "network_rx": 176623,
+    "network_tx": 114204,
+    "blk_read": 0,
+    "blk_write": 0
+  }
+}

tests/test_supervisor.py

@@ -0,0 +1,117 @@
+"""Test for supervisor management client."""
+
+from aioresponses import aioresponses
+from yarl import URL
+
+from aiohasupervisor import SupervisorClient
+from aiohasupervisor.models import SupervisorOptions
+
+from . import load_fixture
+from .const import SUPERVISOR_URL
+
+
+async def test_supervisor_ping(
+    responses: aioresponses, supervisor_client: SupervisorClient
+) -> None:
+    """Test supervisor ping API."""
+    responses.get(f"{SUPERVISOR_URL}/supervisor/ping", status=200)
+    assert await supervisor_client.supervisor.ping() is None
+    assert responses.requests.keys() == {
+        ("GET", URL(f"{SUPERVISOR_URL}/supervisor/ping"))
+    }
+
+
+async def test_supervisor_info(
+    responses: aioresponses, supervisor_client: SupervisorClient
+) -> None:
+    """Test supervisor info API."""
+    responses.get(
+        f"{SUPERVISOR_URL}/supervisor/info",
+        status=200,
+        body=load_fixture("supervisor_info.json"),
+    )
+    info = await supervisor_client.supervisor.info()
+
+    assert info.version == "2024.09.1"
+    assert info.channel == "stable"
+    assert info.arch == "aarch64"
+    assert info.supported is True
+    assert info.healthy is True
+    assert info.logging == "info"
+
+
+async def test_supervisor_stats(
+    responses: aioresponses, supervisor_client: SupervisorClient
+) -> None:
+    """Test supervisor stats API."""
+    responses.get(
+        f"{SUPERVISOR_URL}/supervisor/stats",
+        status=200,
+        body=load_fixture("supervisor_stats.json"),
+    )
+    stats = await supervisor_client.supervisor.stats()
+
+    assert stats.cpu_percent == 0.04
+    assert stats.memory_usage == 243982336
+    assert stats.memory_limit == 3899138048
+    assert stats.memory_percent == 6.26
+
+
+async def test_supervisor_update(
+    responses: aioresponses, supervisor_client: SupervisorClient
+) -> None:
+    """Test supervisor update API."""
+    responses.post(f"{SUPERVISOR_URL}/supervisor/update", status=200)
+    assert await supervisor_client.supervisor.update() is None
+    assert responses.requests.keys() == {
+        ("POST", URL(f"{SUPERVISOR_URL}/supervisor/update"))
+    }
+
+
+async def test_supervisor_reload(
+    responses: aioresponses, supervisor_client: SupervisorClient
+) -> None:
+    """Test supervisor reload API."""
+    responses.post(f"{SUPERVISOR_URL}/supervisor/reload", status=200)
+    assert await supervisor_client.supervisor.reload() is None
+    assert responses.requests.keys() == {
+        ("POST", URL(f"{SUPERVISOR_URL}/supervisor/reload"))
+    }
+
+
+async def test_supervisor_restart(
+    responses: aioresponses, supervisor_client: SupervisorClient
+) -> None:
+    """Test supervisor restart API."""
+    responses.post(f"{SUPERVISOR_URL}/supervisor/restart", status=200)
+    assert await supervisor_client.supervisor.restart() is None
+    assert responses.requests.keys() == {
+        ("POST", URL(f"{SUPERVISOR_URL}/supervisor/restart"))
+    }
+
+
+async def test_supervisor_options(
+    responses: aioresponses, supervisor_client: SupervisorClient
+) -> None:
+    """Test supervisor options API."""
+    responses.post(f"{SUPERVISOR_URL}/supervisor/options", status=200)
+    assert (
+        await supervisor_client.supervisor.options(
+            SupervisorOptions(debug=True, debug_block=True)
+        )
+        is None
+    )
+    assert responses.requests.keys() == {
+        ("POST", URL(f"{SUPERVISOR_URL}/supervisor/options"))
+    }
+
+
+async def test_supervisor_repair(
+    responses: aioresponses, supervisor_client: SupervisorClient
+) -> None:
+    """Test supervisor repair API."""
+    responses.post(f"{SUPERVISOR_URL}/supervisor/repair", status=200)
+    assert await supervisor_client.supervisor.repair() is None
+    assert responses.requests.keys() == {
+        ("POST", URL(f"{SUPERVISOR_URL}/supervisor/repair"))
+    }