Add mounts API

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",
 ]

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

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")

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."""

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"
+      }
+    ]
+  }
+}