diff --git a/aiohasupervisor/models/__init__.py b/aiohasupervisor/models/__init__.py index 1a3614b..8a9187e 100644 --- a/aiohasupervisor/models/__init__.py +++ b/aiohasupervisor/models/__init__.py @@ -62,6 +62,18 @@ ServiceState, ShutdownOptions, ) +from aiohasupervisor.models.mounts import ( + CIFSMountRequest, + CIFSMountResponse, + MountCifsVersion, + MountsInfo, + MountsOptions, + MountState, + MountType, + MountUsage, + NFSMountRequest, + NFSMountResponse, +) from aiohasupervisor.models.network import ( AccessPoint, AuthMethod, @@ -228,4 +240,14 @@ "Service", "ServiceState", "ShutdownOptions", + "CIFSMountRequest", + "CIFSMountResponse", + "MountCifsVersion", + "MountsInfo", + "MountsOptions", + "MountState", + "MountType", + "MountUsage", + "NFSMountRequest", + "NFSMountResponse", ] diff --git a/aiohasupervisor/models/mounts.py b/aiohasupervisor/models/mounts.py new file mode 100644 index 0000000..0cab3e9 --- /dev/null +++ b/aiohasupervisor/models/mounts.py @@ -0,0 +1,133 @@ +"""Models for Supervisor mounts.""" + +from abc import ABC +from dataclasses import dataclass, field +from enum import StrEnum +from pathlib import PurePath +from typing import Literal + +from .base import Request, ResponseData + +# --- ENUMS ---- + + +class MountType(StrEnum): + """MountType type.""" + + CIFS = "cifs" + NFS = "nfs" + + +class MountUsage(StrEnum): + """MountUsage type.""" + + BACKUP = "backup" + MEDIA = "media" + SHARE = "share" + + +class MountState(StrEnum): + """MountState type.""" + + ACTIVE = "active" + ACTIVATING = "activating" + DEACTIVATING = "deactivating" + FAILED = "failed" + INACTIVE = "inactive" + MAINTENANCE = "maintenance" + RELOADING = "reloading" + + +class MountCifsVersion(StrEnum): + """Mount CIFS version.""" + + LEGACY_1_0 = "1.0" + LEGACY_2_0 = "2.0" + + +# --- OBJECTS ---- + + +@dataclass(frozen=True) +class Mount(ABC): + """Mount ABC type.""" + + usage: MountUsage + server: str + port: int | None = field(kw_only=True, default=None) + + +@dataclass(frozen=True) +class CIFSMount(ABC): + """CIFSMount ABC type.""" + + share: str + version: MountCifsVersion | None = field(kw_only=True, default=None) + + +@dataclass(frozen=True) +class NFSMount(ABC): + """NFSMount ABC type.""" + + path: PurePath + + +@dataclass(frozen=True) +class MountResponse(ABC): + """MountResponse model.""" + + name: str + read_only: bool + state: MountState | None + + +@dataclass(frozen=True) +class MountRequest(ABC): # noqa: B024 + """MountRequest model.""" + + read_only: bool | None = field(kw_only=True, default=None) + + +@dataclass(frozen=True, slots=True) +class CIFSMountResponse(Mount, MountResponse, CIFSMount, ResponseData): + """CIFSMountResponse model.""" + + type: Literal[MountType.CIFS] + + +@dataclass(frozen=True, slots=True) +class NFSMountResponse(Mount, MountResponse, NFSMount, ResponseData): + """NFSMountResponse model.""" + + type: Literal[MountType.NFS] + + +@dataclass(frozen=True, slots=True) +class CIFSMountRequest(Mount, MountRequest, CIFSMount, Request): + """CIFSMountRequest model.""" + + type: Literal[MountType.CIFS] = field(init=False, default=MountType.CIFS) + username: str | None = field(kw_only=True, default=None) + password: str | None = field(kw_only=True, default=None) + + +@dataclass(frozen=True, slots=True) +class NFSMountRequest(Mount, MountRequest, NFSMount, Request): + """NFSMountRequest model.""" + + type: Literal[MountType.NFS] = field(init=False, default=MountType.NFS) + + +@dataclass(frozen=True, slots=True) +class MountsInfo(ResponseData): + """MountsInfo model.""" + + default_backup_mount: str | None + mounts: list[CIFSMountResponse | NFSMountResponse] + + +@dataclass(frozen=True, slots=True) +class MountsOptions(Request): + """MountsOptions model.""" + + default_backup_mount: str | None diff --git a/aiohasupervisor/mounts.py b/aiohasupervisor/mounts.py new file mode 100644 index 0000000..75d9725 --- /dev/null +++ b/aiohasupervisor/mounts.py @@ -0,0 +1,37 @@ +"""Mounts client for Supervisor.""" + +from .client import _SupervisorComponentClient +from .models.mounts import CIFSMountRequest, MountsInfo, MountsOptions, NFSMountRequest + + +class MountsClient(_SupervisorComponentClient): + """Handle mounts access in supervisor.""" + + async def info(self) -> MountsInfo: + """Get mounts info.""" + result = await self._client.get("mounts") + return MountsInfo.from_dict(result.data) + + async def options(self, options: MountsOptions) -> None: + """Set mounts options.""" + await self._client.post("mounts/options", json=options.to_dict()) + + async def create_mount( + self, name: str, config: CIFSMountRequest | NFSMountRequest + ) -> None: + """Create a new mount.""" + await self._client.post("mounts", json={"name": name, **config.to_dict()}) + + async def update_mount( + self, name: str, config: CIFSMountRequest | NFSMountRequest + ) -> None: + """Update an existing mount.""" + await self._client.put(f"mounts/{name}", json=config.to_dict()) + + async def delete_mount(self, name: str) -> None: + """Delete an existing mount.""" + await self._client.delete(f"mounts/{name}") + + async def reload_mount(self, name: str) -> None: + """Reload details of an existing mount.""" + await self._client.post(f"mounts/{name}/reload") diff --git a/aiohasupervisor/root.py b/aiohasupervisor/root.py index 4fbddbc..e2e7d48 100644 --- a/aiohasupervisor/root.py +++ b/aiohasupervisor/root.py @@ -11,6 +11,7 @@ from .homeassistant import HomeAssistantClient from .host import HostClient from .models.root import AvailableUpdate, AvailableUpdates, RootInfo +from .mounts import MountsClient from .network import NetworkClient from .os import OSClient from .resolution import ResolutionClient @@ -33,6 +34,7 @@ def __init__( self._os = OSClient(self._client) self._backups = BackupsClient(self._client) self._discovery = DiscoveryClient(self._client) + self._mounts = MountsClient(self._client) self._network = NetworkClient(self._client) self._host = HostClient(self._client) self._resolution = ResolutionClient(self._client) @@ -65,6 +67,11 @@ def discovery(self) -> DiscoveryClient: """Get discovery component client.""" return self._discovery + @property + def mounts(self) -> MountsClient: + """Get mounts component client.""" + return self._mounts + @property def network(self) -> NetworkClient: """Get network component client.""" diff --git a/tests/fixtures/mounts_info.json b/tests/fixtures/mounts_info.json new file mode 100644 index 0000000..27dde64 --- /dev/null +++ b/tests/fixtures/mounts_info.json @@ -0,0 +1,38 @@ +{ + "result": "ok", + "data": { + "default_backup_mount": "Test", + "mounts": [ + { + "share": "backup", + "server": "test.local", + "name": "Test", + "type": "cifs", + "usage": "backup", + "read_only": false, + "version": null, + "state": "active" + }, + { + "share": "share", + "server": "test2.local", + "name": "Test2", + "type": "cifs", + "usage": "share", + "read_only": true, + "version": "2.0", + "port": 12345, + "state": "active" + }, + { + "server": "test3.local", + "name": "Test2", + "type": "nfs", + "usage": "media", + "read_only": false, + "path": "media", + "state": "active" + } + ] + } +} diff --git a/tests/test_mounts.py b/tests/test_mounts.py new file mode 100644 index 0000000..196d9a8 --- /dev/null +++ b/tests/test_mounts.py @@ -0,0 +1,173 @@ +"""Test mounts supervisor client.""" + +from pathlib import PurePath + +from aioresponses import aioresponses +import pytest +from yarl import URL + +from aiohasupervisor import SupervisorClient +from aiohasupervisor.models import ( + CIFSMountRequest, + MountCifsVersion, + MountsOptions, + MountUsage, + NFSMountRequest, +) + +from . import load_fixture +from .const import SUPERVISOR_URL + + +async def test_mounts_info( + responses: aioresponses, supervisor_client: SupervisorClient +) -> None: + """Test mounts info API.""" + responses.get( + f"{SUPERVISOR_URL}/mounts", status=200, body=load_fixture("mounts_info.json") + ) + info = await supervisor_client.mounts.info() + assert info.default_backup_mount == "Test" + + assert info.mounts[0].name == "Test" + assert info.mounts[0].server == "test.local" + assert info.mounts[0].type == "cifs" + assert info.mounts[0].share == "backup" + assert info.mounts[0].usage == "backup" + assert info.mounts[0].read_only is False + assert info.mounts[0].version is None + assert info.mounts[0].state == "active" + + assert info.mounts[1].usage == "share" + assert info.mounts[1].read_only is True + assert info.mounts[1].version == "2.0" + assert info.mounts[1].port == 12345 + + assert info.mounts[2].type == "nfs" + assert info.mounts[2].usage == "media" + assert info.mounts[2].path.as_posix() == "media" + + +@pytest.mark.parametrize("mount_name", ["test", None]) +async def test_mounts_options( + responses: aioresponses, supervisor_client: SupervisorClient, mount_name: str | None +) -> None: + """Test mounts options API.""" + responses.post(f"{SUPERVISOR_URL}/mounts/options", status=200) + assert ( + await supervisor_client.mounts.options( + MountsOptions(default_backup_mount=mount_name) + ) + is None + ) + assert responses.requests.keys() == { + ("POST", URL(f"{SUPERVISOR_URL}/mounts/options")) + } + + +@pytest.mark.parametrize( + "mount_config", + [ + CIFSMountRequest( + server="test.local", + share="backup", + usage=MountUsage.BACKUP, + ), + CIFSMountRequest( + server="test.local", + share="media", + port=12345, + usage=MountUsage.MEDIA, + version=MountCifsVersion.LEGACY_2_0, + read_only=True, + username="test", + password="test", # noqa: S106 + ), + NFSMountRequest( + server="test.local", + path=PurePath("share"), + usage=MountUsage.SHARE, + ), + NFSMountRequest( + server="test.local", + path=PurePath("backups"), + port=12345, + read_only=False, + usage=MountUsage.BACKUP, + ), + ], +) +async def test_create_mount( + responses: aioresponses, + supervisor_client: SupervisorClient, + mount_config: CIFSMountRequest | NFSMountRequest, +) -> None: + """Test create mount API.""" + responses.post(f"{SUPERVISOR_URL}/mounts", status=200) + assert await supervisor_client.mounts.create_mount("test", mount_config) is None + assert responses.requests.keys() == {("POST", URL(f"{SUPERVISOR_URL}/mounts"))} + + +@pytest.mark.parametrize( + "mount_config", + [ + CIFSMountRequest( + server="test.local", + share="backup", + usage=MountUsage.BACKUP, + ), + CIFSMountRequest( + server="test.local", + share="media", + port=12345, + usage=MountUsage.MEDIA, + version=MountCifsVersion.LEGACY_2_0, + read_only=True, + username="test", + password="test", # noqa: S106 + ), + NFSMountRequest( + server="test.local", + path=PurePath("share"), + usage=MountUsage.SHARE, + ), + NFSMountRequest( + server="test.local", + path=PurePath("backups"), + port=12345, + read_only=False, + usage=MountUsage.BACKUP, + ), + ], +) +async def test_update_mount( + responses: aioresponses, + supervisor_client: SupervisorClient, + mount_config: CIFSMountRequest | NFSMountRequest, +) -> None: + """Test update mount API.""" + responses.put(f"{SUPERVISOR_URL}/mounts/test", status=200) + assert await supervisor_client.mounts.update_mount("test", mount_config) is None + assert responses.requests.keys() == {("PUT", URL(f"{SUPERVISOR_URL}/mounts/test"))} + + +async def test_delete_mount( + responses: aioresponses, supervisor_client: SupervisorClient +) -> None: + """Test delete mount API.""" + responses.delete(f"{SUPERVISOR_URL}/mounts/test", status=200) + assert await supervisor_client.mounts.delete_mount("test") is None + assert responses.requests.keys() == { + ("DELETE", URL(f"{SUPERVISOR_URL}/mounts/test")) + } + + +async def test_reload_mount( + responses: aioresponses, supervisor_client: SupervisorClient +) -> None: + """Test reload mount API.""" + responses.post(f"{SUPERVISOR_URL}/mounts/test/reload", status=200) + assert await supervisor_client.mounts.reload_mount("test") is None + assert responses.requests.keys() == { + ("POST", URL(f"{SUPERVISOR_URL}/mounts/test/reload")) + }