diff --git a/aiohasupervisor/models/__init__.py b/aiohasupervisor/models/__init__.py index 2133aee..396a1b2 100644 --- a/aiohasupervisor/models/__init__.py +++ b/aiohasupervisor/models/__init__.py @@ -32,6 +32,20 @@ HomeAssistantStopOptions, HomeAssistantUpdateOptions, ) +from aiohasupervisor.models.os import ( + BootSlot, + BootSlotName, + DataDisk, + GreenInfo, + GreenOptions, + MigrateDataOptions, + OSInfo, + OSUpdate, + RaucState, + SetBootSlotOptions, + YellowInfo, + YellowOptions, +) from aiohasupervisor.models.resolution import ( Check, CheckOptions, @@ -112,4 +126,16 @@ "HomeAssistantStats", "HomeAssistantStopOptions", "HomeAssistantUpdateOptions", + "RaucState", + "BootSlotName", + "BootSlot", + "OSInfo", + "OSUpdate", + "MigrateDataOptions", + "DataDisk", + "SetBootSlotOptions", + "GreenInfo", + "GreenOptions", + "YellowInfo", + "YellowOptions", ] diff --git a/aiohasupervisor/models/os.py b/aiohasupervisor/models/os.py new file mode 100644 index 0000000..82ed337 --- /dev/null +++ b/aiohasupervisor/models/os.py @@ -0,0 +1,126 @@ +"""Models for OS APIs.""" + +from dataclasses import dataclass +from enum import StrEnum +from pathlib import PurePath + +from .base import Options, Request, ResponseData + +# --- ENUMS ---- + + +class RaucState(StrEnum): + """RaucState type.""" + + GOOD = "good" + BAD = "bad" + ACTIVE = "active" + + +class BootSlotName(StrEnum): + """BootSlotName type.""" + + A = "A" + B = "B" + + +# --- OBJECTS ---- + + +@dataclass(frozen=True, slots=True) +class BootSlot(ResponseData): + """BootSlot model.""" + + state: str + status: RaucState | None + version: str | None + + +@dataclass(frozen=True, slots=True) +class OSInfo(ResponseData): + """OSInfo model.""" + + version: str | None + version_latest: str | None + update_available: bool + board: str | None + boot: str | None + data_disk: str | None + boot_slots: dict[str, BootSlot] + + +@dataclass(frozen=True, slots=True) +class OSUpdate(Request): + """OSUpdate model.""" + + version: str | None = None + + +@dataclass(frozen=True, slots=True) +class MigrateDataOptions(Request): + """MigrateDataOptions model.""" + + device: str + + +@dataclass(frozen=True, slots=True) +class DataDisk(ResponseData): + """DataDisk model.""" + + name: str + vendor: str + model: str + serial: str + size: int + id: str + dev_path: PurePath + + +@dataclass(frozen=True, slots=True) +class DataDiskList(ResponseData): + """DataDiskList model.""" + + disks: list[DataDisk] + + +@dataclass(frozen=True, slots=True) +class SetBootSlotOptions(Request): + """SetBootSlotOptions model.""" + + boot_slot: BootSlotName + + +@dataclass(frozen=True, slots=True) +class GreenInfo(ResponseData): + """GreenInfo model.""" + + activity_led: bool + power_led: bool + system_health_led: bool + + +@dataclass(frozen=True, slots=True) +class GreenOptions(Options): + """GreenOptions model.""" + + activity_led: bool | None = None + power_led: bool | None = None + system_health_led: bool | None = None + + +@dataclass(frozen=True, slots=True) +class YellowInfo(ResponseData): + """YellowInfo model.""" + + disk_led: bool + heartbeat_led: bool + power_led: bool + + +@dataclass(frozen=True, slots=True) +class YellowOptions(Options): + """YellowOptions model.""" + + disk_led: bool | None = None + heartbeat_led: bool | None = None + power_led: bool | None = None diff --git a/aiohasupervisor/os.py b/aiohasupervisor/os.py new file mode 100644 index 0000000..593c76e --- /dev/null +++ b/aiohasupervisor/os.py @@ -0,0 +1,69 @@ +"""OS client for supervisor.""" + +from .client import _SupervisorComponentClient +from .models.os import ( + DataDisk, + DataDiskList, + GreenInfo, + GreenOptions, + MigrateDataOptions, + OSInfo, + OSUpdate, + SetBootSlotOptions, + YellowInfo, + YellowOptions, +) + + +class OSClient(_SupervisorComponentClient): + """Handles OS access in supervisor.""" + + async def info(self) -> OSInfo: + """Get OS info.""" + result = await self._client.get("os/info") + return OSInfo.from_dict(result.data) + + async def update(self, options: OSUpdate | None = None) -> None: + """Update OS.""" + await self._client.post( + "os/update", json=options.to_dict() if options else None + ) + + async def config_sync(self) -> None: + """Trigger config reload on OS.""" + await self._client.post("os/config/sync") + + async def migrate_data(self, options: MigrateDataOptions) -> None: + """Migrate data to new data disk and reboot.""" + await self._client.post("os/datadisk/move", json=options.to_dict()) + + async def list_data_disks(self) -> list[DataDisk]: + """Get all data disks.""" + result = await self._client.get("os/datadisk/list") + return DataDiskList.from_dict(result.data).disks + + async def wipe_data(self) -> None: + """Trigger data disk wipe on host and reboot.""" + await self._client.post("os/datadisk/wipe") + + async def set_boot_slot(self, options: SetBootSlotOptions) -> None: + """Change active boot slot on host and reboot.""" + await self._client.post("os/boot-slot", json=options.to_dict()) + + async def green_info(self) -> GreenInfo: + """Get info for green board (if in use).""" + result = await self._client.get("os/boards/green") + return GreenInfo.from_dict(result.data) + + async def green_options(self, options: GreenOptions) -> None: + """Set options for green board (if in use).""" + await self._client.post("os/boards/green", json=options.to_dict()) + + async def yellow_info(self) -> YellowInfo: + """Get info for yellow board (if in use).""" + result = await self._client.get("os/boards/yellow") + return YellowInfo.from_dict(result.data) + + async def yellow_options(self, options: YellowOptions) -> None: + """Set options for yellow board (if in use).""" + await self._client.post("os/boards/yellow", json=options.to_dict()) diff --git a/aiohasupervisor/root.py b/aiohasupervisor/root.py index 84d944c..acf8d5b 100644 --- a/aiohasupervisor/root.py +++ b/aiohasupervisor/root.py @@ -8,6 +8,7 @@ from .client import _SupervisorClient from .homeassistant import HomeAssistantClient from .models.root import AvailableUpdate, AvailableUpdates, RootInfo +from .os import OSClient from .resolution import ResolutionClient from .store import StoreClient from .supervisor import SupervisorManagementClient @@ -26,6 +27,7 @@ def __init__( """Initialize client.""" self._client = _SupervisorClient(api_host, token, request_timeout, session) self._addons = AddonsClient(self._client) + self._os = OSClient(self._client) self._resolution = ResolutionClient(self._client) self._store = StoreClient(self._client) self._supervisor = SupervisorManagementClient(self._client) @@ -41,6 +43,11 @@ def homeassistant(self) -> HomeAssistantClient: """Get Home Assistant component client.""" return self._homeassistant + @property + def os(self) -> OSClient: + """Get OS component client.""" + return self._os + @property def resolution(self) -> ResolutionClient: """Get resolution center component client.""" diff --git a/tests/fixtures/os_datadisk_list.json b/tests/fixtures/os_datadisk_list.json new file mode 100644 index 0000000..6111fb5 --- /dev/null +++ b/tests/fixtures/os_datadisk_list.json @@ -0,0 +1,17 @@ +{ + "result": "ok", + "data": { + "devices": ["SSK-SSK-Storage-DF123"], + "disks": [ + { + "name": "SSK SSK Storage (DF123)", + "vendor": "SSK", + "model": "SSK Storage", + "serial": "DF123", + "size": 250059350016, + "id": "SSK-SSK-Storage-DF123", + "dev_path": "/dev/sda" + } + ] + } +} diff --git a/tests/fixtures/os_green_info.json b/tests/fixtures/os_green_info.json new file mode 100644 index 0000000..5110778 --- /dev/null +++ b/tests/fixtures/os_green_info.json @@ -0,0 +1,4 @@ +{ + "result": "ok", + "data": { "activity_led": true, "power_led": true, "system_health_led": true } +} diff --git a/tests/fixtures/os_info.json b/tests/fixtures/os_info.json new file mode 100644 index 0000000..5731e86 --- /dev/null +++ b/tests/fixtures/os_info.json @@ -0,0 +1,19 @@ +{ + "result": "ok", + "data": { + "version": "13.0", + "version_latest": "13.1", + "update_available": true, + "board": "odroid-n2", + "boot": "B", + "data_disk": "BJTD4R-0xaabbccdd", + "boot_slots": { + "A": { "state": "inactive", "status": "good", "version": null }, + "B": { + "state": "booted", + "status": "good", + "version": "13.0" + } + } + } +} diff --git a/tests/fixtures/os_yellow_info.json b/tests/fixtures/os_yellow_info.json new file mode 100644 index 0000000..e596e04 --- /dev/null +++ b/tests/fixtures/os_yellow_info.json @@ -0,0 +1,4 @@ +{ + "result": "ok", + "data": { "disk_led": true, "heartbeat_led": true, "power_led": true } +} diff --git a/tests/test_os.py b/tests/test_os.py new file mode 100644 index 0000000..ba0dc9f --- /dev/null +++ b/tests/test_os.py @@ -0,0 +1,178 @@ +"""Test OS supervisor client.""" + +from pathlib import PurePath + +from aioresponses import aioresponses +import pytest +from yarl import URL + +from aiohasupervisor import SupervisorClient +from aiohasupervisor.models import ( + BootSlotName, + GreenOptions, + MigrateDataOptions, + OSUpdate, + SetBootSlotOptions, + YellowOptions, +) + +from . import load_fixture +from .const import SUPERVISOR_URL + + +async def test_os_info( + responses: aioresponses, supervisor_client: SupervisorClient +) -> None: + """Test OS info API.""" + responses.get( + f"{SUPERVISOR_URL}/os/info", + status=200, + body=load_fixture("os_info.json"), + ) + info = await supervisor_client.os.info() + assert info.version == "13.0" + assert info.version_latest == "13.1" + assert info.update_available is True + assert info.boot_slots["A"].state == "inactive" + assert info.boot_slots["B"].state == "booted" + assert info.boot_slots["B"].status == "good" + assert info.boot_slots["B"].version == "13.0" + + +@pytest.mark.parametrize("options", [None, OSUpdate(version="13.0")]) +async def test_os_update( + responses: aioresponses, + supervisor_client: SupervisorClient, + options: OSUpdate | None, +) -> None: + """Test OS update API.""" + responses.post(f"{SUPERVISOR_URL}/os/update", status=200) + assert await supervisor_client.os.update(options) is None + assert responses.requests.keys() == {("POST", URL(f"{SUPERVISOR_URL}/os/update"))} + + +async def test_os_config_sync( + responses: aioresponses, supervisor_client: SupervisorClient +) -> None: + """Test OS config sync API.""" + responses.post(f"{SUPERVISOR_URL}/os/config/sync", status=200) + assert await supervisor_client.os.config_sync() is None + assert responses.requests.keys() == { + ("POST", URL(f"{SUPERVISOR_URL}/os/config/sync")) + } + + +async def test_os_migrate_data( + responses: aioresponses, supervisor_client: SupervisorClient +) -> None: + """Test OS migrate data API.""" + responses.post(f"{SUPERVISOR_URL}/os/datadisk/move", status=200) + assert ( + await supervisor_client.os.migrate_data(MigrateDataOptions(device="/dev/test")) + is None + ) + assert responses.requests.keys() == { + ("POST", URL(f"{SUPERVISOR_URL}/os/datadisk/move")) + } + + +async def test_os_list_data_disks( + responses: aioresponses, supervisor_client: SupervisorClient +) -> None: + """Test OS datadisk list API.""" + responses.get( + f"{SUPERVISOR_URL}/os/datadisk/list", + status=200, + body=load_fixture("os_datadisk_list.json"), + ) + datadisks = await supervisor_client.os.list_data_disks() + assert datadisks[0].vendor == "SSK" + assert datadisks[0].model == "SSK Storage" + assert datadisks[0].serial == "DF123" + assert datadisks[0].name == "SSK SSK Storage (DF123)" + assert datadisks[0].dev_path == PurePath("/dev/sda") + + +async def test_os_wipe_data( + responses: aioresponses, supervisor_client: SupervisorClient +) -> None: + """Test OS wipe data API.""" + responses.post(f"{SUPERVISOR_URL}/os/datadisk/wipe", status=200) + assert await supervisor_client.os.wipe_data() is None + assert responses.requests.keys() == { + ("POST", URL(f"{SUPERVISOR_URL}/os/datadisk/wipe")) + } + + +async def test_os_set_boot_slot( + responses: aioresponses, supervisor_client: SupervisorClient +) -> None: + """Test OS set boot slot API.""" + responses.post(f"{SUPERVISOR_URL}/os/boot-slot", status=200) + assert ( + await supervisor_client.os.set_boot_slot( + SetBootSlotOptions(boot_slot=BootSlotName.B) + ) + is None + ) + assert responses.requests.keys() == { + ("POST", URL(f"{SUPERVISOR_URL}/os/boot-slot")) + } + + +async def test_os_green_info( + responses: aioresponses, supervisor_client: SupervisorClient +) -> None: + """Test OS green board info API.""" + responses.get( + f"{SUPERVISOR_URL}/os/boards/green", + status=200, + body=load_fixture("os_green_info.json"), + ) + info = await supervisor_client.os.green_info() + assert info.activity_led is True + assert info.power_led is True + assert info.system_health_led is True + + +async def test_os_green_options( + responses: aioresponses, supervisor_client: SupervisorClient +) -> None: + """Test OS green board options API.""" + responses.post(f"{SUPERVISOR_URL}/os/boards/green", status=200) + assert ( + await supervisor_client.os.green_options(GreenOptions(activity_led=False)) + is None + ) + assert responses.requests.keys() == { + ("POST", URL(f"{SUPERVISOR_URL}/os/boards/green")) + } + + +async def test_os_yellow_info( + responses: aioresponses, supervisor_client: SupervisorClient +) -> None: + """Test OS yellow board info API.""" + responses.get( + f"{SUPERVISOR_URL}/os/boards/yellow", + status=200, + body=load_fixture("os_yellow_info.json"), + ) + info = await supervisor_client.os.yellow_info() + assert info.disk_led is True + assert info.heartbeat_led is True + assert info.power_led is True + + +async def test_os_yellow_options( + responses: aioresponses, supervisor_client: SupervisorClient +) -> None: + """Test OS yellow board options API.""" + responses.post(f"{SUPERVISOR_URL}/os/boards/yellow", status=200) + assert ( + await supervisor_client.os.yellow_options(YellowOptions(heartbeat_led=False)) + is None + ) + assert responses.requests.keys() == { + ("POST", URL(f"{SUPERVISOR_URL}/os/boards/yellow")) + }