Add core APIs to client library

Author: mdegat01

aiohasupervisor/homeassistant.py

@@ -0,0 +1,61 @@
+"""Home Assistant client for supervisor."""
+
+from .client import _SupervisorComponentClient
+from .models.homeassistant import (
+    HomeAssistantInfo,
+    HomeAssistantOptions,
+    HomeAssistantRestartOptions,
+    HomeAssistantStats,
+    HomeAssistantStopOptions,
+    HomeAssistantUpdateOptions,
+)
+
+
+class HomeAssistantClient(_SupervisorComponentClient):
+    """Handles Home Assistant access in supervisor."""
+
+    async def info(self) -> HomeAssistantInfo:
+        """Get Home Assistant info."""
+        result = await self._client.get("core/info")
+        return HomeAssistantInfo.from_dict(result.data)
+
+    async def stats(self) -> HomeAssistantStats:
+        """Get Home Assistant stats."""
+        result = await self._client.get("core/stats")
+        return HomeAssistantStats.from_dict(result.data)
+
+    async def options(self, options: HomeAssistantOptions) -> None:
+        """Set Home Assistant options."""
+        await self._client.post("core/options", json=options.to_dict())
+
+    async def update(self, options: HomeAssistantUpdateOptions | None = None) -> None:
+        """Update Home Assistant."""
+        await self._client.post(
+            "core/update", json=options.to_dict() if options else None
+        )
+
+    async def restart(self, options: HomeAssistantRestartOptions | None = None) -> None:
+        """Restart Home Assistant."""
+        await self._client.post(
+            "core/restart", json=options.to_dict() if options else None
+        )
+
+    async def stop(self, options: HomeAssistantStopOptions | None = None) -> None:
+        """Stop Home Assistant."""
+        await self._client.post(
+            "core/stop", json=options.to_dict() if options else None
+        )
+
+    async def start(self) -> None:
+        """Start Home Assistant."""
+        await self._client.post("core/start")
+
+    async def check_config(self) -> None:
+        """Check Home Assistant config."""
+        await self._client.post("core/check")
+
+    async def rebuild(self, options: HomeAssistantRestartOptions | None = None) -> None:
+        """Rebuild Home Assistant."""
+        await self._client.post(
+            "core/rebuild", json=options.to_dict() if options else None
+        )

aiohasupervisor/models/__init__.py

@@ -23,6 +23,14 @@
     StoreInfo,
     SupervisorRole,
 )
+from aiohasupervisor.models.homeassistant import (
+    HomeAssistantInfo,
+    HomeAssistantOptions,
+    HomeAssistantRestartOptions,
+    HomeAssistantStats,
+    HomeAssistantStopOptions,
+    HomeAssistantUpdateOptions,
+)
 from aiohasupervisor.models.resolution import (
     Check,
     CheckOptions,
@@ -96,4 +104,10 @@
     "SupervisorOptions",
     "SupervisorStats",
     "SupervisorUpdateOptions",
+    "HomeAssistantInfo",
+    "HomeAssistantOptions",
+    "HomeAssistantRestartOptions",
+    "HomeAssistantStats",
+    "HomeAssistantStopOptions",
+    "HomeAssistantUpdateOptions",
 ]

aiohasupervisor/models/addons.py

@@ -9,7 +9,7 @@
 from mashumaro import field_options
 from mashumaro.config import TO_DICT_ADD_BY_ALIAS_FLAG, BaseConfig
 
-from .base import DEFAULT, Options, Request, RequestConfig, ResponseData
+from .base import DEFAULT, ContainerStats, Options, Request, RequestConfig, ResponseData
 
 # --- ENUMS ----
 
@@ -296,18 +296,9 @@ class AddonsSecurityOptions(Options):
 
 
 @dataclass(frozen=True, slots=True)
-class AddonsStats(ResponseData):
+class AddonsStats(ContainerStats):
     """AddonsStats 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 AddonsUninstall(Request):

aiohasupervisor/models/homeassistant.py

@@ -0,0 +1,70 @@
+"""Models for Home Assistant."""
+
+from dataclasses import dataclass
+
+from .base import DEFAULT, ContainerStats, Options, Request, ResponseData
+
+# --- OBJECTS ----
+
+
+@dataclass(frozen=True, slots=True)
+class HomeAssistantInfo(ResponseData):
+    """HomeAssistantInfo model."""
+
+    version: str | None
+    version_latest: str | None
+    update_available: bool
+    machine: str
+    ip_address: str
+    arch: str
+    image: str
+    boot: bool
+    port: int
+    ssl: bool
+    watchdog: bool
+    audio_input: str
+    audio_output: str
+    backups_exclude_database: bool
+
+
+@dataclass(frozen=True, slots=True)
+class HomeAssistantStats(ContainerStats):
+    """HomeAssistantStats model."""
+
+
+@dataclass(frozen=True, slots=True)
+class HomeAssistantOptions(Options):
+    """HomeAssistantOptions model."""
+
+    boot: bool | None = None
+    image: str | None = DEFAULT  # type: ignore[assignment]
+    port: int | None = None
+    ssl: bool | None = None
+    watchdog: bool | None = None
+    refresh_token: str | None = DEFAULT  # type: ignore[assignment]
+    audio_input: str | None = DEFAULT  # type: ignore[assignment]
+    audio_output: str | None = DEFAULT  # type: ignore[assignment]
+    backups_exclude_database: bool | None = None
+
+
+@dataclass(frozen=True, slots=True)
+class HomeAssistantUpdateOptions(Options):
+    """HomeAssistantUpdateOptions model."""
+
+    version: str | None = None
+    backup: bool | None = None
+
+
+@dataclass(frozen=True, slots=True)
+class HomeAssistantRestartOptions(Options):
+    """HomeAssistantRestartOptions model."""
+
+    safe_mode: bool | None = None
+    force: bool | None = None
+
+
+@dataclass(frozen=True, slots=True)
+class HomeAssistantStopOptions(Request):
+    """HomeAssistantStopOptions model."""
+
+    force: bool

aiohasupervisor/root.py

@@ -6,6 +6,7 @@
 
 from .addons import AddonsClient
 from .client import _SupervisorClient
+from .homeassistant import HomeAssistantClient
 from .models.root import AvailableUpdate, AvailableUpdates, RootInfo
 from .resolution import ResolutionClient
 from .store import StoreClient
@@ -28,12 +29,18 @@ def __init__(
         self._resolution = ResolutionClient(self._client)
         self._store = StoreClient(self._client)
         self._supervisor = SupervisorManagementClient(self._client)
+        self._homeassistant = HomeAssistantClient(self._client)
 
     @property
     def addons(self) -> AddonsClient:
         """Get addons component client."""
         return self._addons
 
+    @property
+    def homeassistant(self) -> HomeAssistantClient:
+        """Get Home Assistant component client."""
+        return self._homeassistant
+
     @property
     def resolution(self) -> ResolutionClient:
         """Get resolution center component client."""

tests/fixtures/homeassistant_info.json

@@ -0,0 +1,19 @@
+{
+  "result": "ok",
+  "data": {
+    "version": "2024.9.0",
+    "version_latest": "2024.9.0",
+    "update_available": false,
+    "machine": "odroid-n2",
+    "ip_address": "172.30.32.1",
+    "arch": "aarch64",
+    "image": "ghcr.io/home-assistant/odroid-n2-homeassistant",
+    "boot": true,
+    "port": 8123,
+    "ssl": false,
+    "watchdog": true,
+    "audio_input": null,
+    "audio_output": null,
+    "backups_exclude_database": false
+  }
+}

tests/fixtures/homeassistant_stats.json

@@ -0,0 +1,13 @@
+{
+  "result": "ok",
+  "data": {
+    "cpu_percent": 0.01,
+    "memory_usage": 678883328,
+    "memory_limit": 3899138048,
+    "memory_percent": 17.41,
+    "network_rx": 0,
+    "network_tx": 0,
+    "blk_read": 0,
+    "blk_write": 0
+  }
+}

tests/test_homeassistant.py

@@ -0,0 +1,119 @@
+"""Test Home Assistant supervisor client."""
+
+from aioresponses import aioresponses
+from yarl import URL
+
+from aiohasupervisor import SupervisorClient
+from aiohasupervisor.models import HomeAssistantOptions
+
+from . import load_fixture
+from .const import SUPERVISOR_URL
+
+
+async def test_homeassistant_info(
+    responses: aioresponses, supervisor_client: SupervisorClient
+) -> None:
+    """Test Home Assistant info API."""
+    responses.get(
+        f"{SUPERVISOR_URL}/core/info",
+        status=200,
+        body=load_fixture("homeassistant_info.json"),
+    )
+    info = await supervisor_client.homeassistant.info()
+
+    assert info.version == "2024.9.0"
+    assert info.update_available is False
+    assert info.arch == "aarch64"
+    assert info.ssl is False
+    assert info.port == 8123
+    assert info.audio_output is None
+
+
+async def test_homeassistant_stats(
+    responses: aioresponses, supervisor_client: SupervisorClient
+) -> None:
+    """Test Home Assistant stats API."""
+    responses.get(
+        f"{SUPERVISOR_URL}/core/stats",
+        status=200,
+        body=load_fixture("homeassistant_stats.json"),
+    )
+    stats = await supervisor_client.homeassistant.stats()
+
+    assert stats.cpu_percent == 0.01
+    assert stats.memory_usage == 678883328
+    assert stats.memory_percent == 17.41
+
+
+async def test_homeassistant_options(
+    responses: aioresponses, supervisor_client: SupervisorClient
+) -> None:
+    """Test Home Assistant options API."""
+    responses.post(f"{SUPERVISOR_URL}/core/options", status=200)
+    assert (
+        await supervisor_client.homeassistant.options(
+            HomeAssistantOptions(watchdog=False, backups_exclude_database=True)
+        )
+        is None
+    )
+    assert responses.requests.keys() == {
+        ("POST", URL(f"{SUPERVISOR_URL}/core/options"))
+    }
+
+
+async def test_homeassistant_update(
+    responses: aioresponses, supervisor_client: SupervisorClient
+) -> None:
+    """Test Home Assistant update API."""
+    responses.post(f"{SUPERVISOR_URL}/core/update", status=200)
+    assert await supervisor_client.homeassistant.update() is None
+    assert responses.requests.keys() == {("POST", URL(f"{SUPERVISOR_URL}/core/update"))}
+
+
+async def test_homeassistant_restart(
+    responses: aioresponses, supervisor_client: SupervisorClient
+) -> None:
+    """Test Home Assistant restart API."""
+    responses.post(f"{SUPERVISOR_URL}/core/restart", status=200)
+    assert await supervisor_client.homeassistant.restart() is None
+    assert responses.requests.keys() == {
+        ("POST", URL(f"{SUPERVISOR_URL}/core/restart"))
+    }
+
+
+async def test_homeassistant_stop(
+    responses: aioresponses, supervisor_client: SupervisorClient
+) -> None:
+    """Test Home Assistant stop API."""
+    responses.post(f"{SUPERVISOR_URL}/core/stop", status=200)
+    assert await supervisor_client.homeassistant.stop() is None
+    assert responses.requests.keys() == {("POST", URL(f"{SUPERVISOR_URL}/core/stop"))}
+
+
+async def test_homeassistant_start(
+    responses: aioresponses, supervisor_client: SupervisorClient
+) -> None:
+    """Test Home Assistant start API."""
+    responses.post(f"{SUPERVISOR_URL}/core/start", status=200)
+    assert await supervisor_client.homeassistant.start() is None
+    assert responses.requests.keys() == {("POST", URL(f"{SUPERVISOR_URL}/core/start"))}
+
+
+async def test_homeassistant_check_config(
+    responses: aioresponses, supervisor_client: SupervisorClient
+) -> None:
+    """Test Home Assistant check config API."""
+    responses.post(f"{SUPERVISOR_URL}/core/check", status=200)
+    assert await supervisor_client.homeassistant.check_config() is None
+    assert responses.requests.keys() == {("POST", URL(f"{SUPERVISOR_URL}/core/check"))}
+
+
+async def test_homeassistant_rebuild(
+    responses: aioresponses, supervisor_client: SupervisorClient
+) -> None:
+    """Test Home Assistant rebuild API."""
+    responses.post(f"{SUPERVISOR_URL}/core/rebuild", status=200)
+    assert await supervisor_client.homeassistant.rebuild() is None
+    assert responses.requests.keys() == {
+        ("POST", URL(f"{SUPERVISOR_URL}/core/rebuild"))
+    }