Add addon availability API (#172)

* Add addon availability API

* Add support for keyed errors

* Remove unused regexes

* Remove message_template

Author: mdegat01

aiohasupervisor/__init__.py

@@ -1,6 +1,10 @@
 """Init file for aiohasupervisor."""
 
 from aiohasupervisor.exceptions import (
+    AddonNotSupportedArchitectureError,
+    AddonNotSupportedError,
+    AddonNotSupportedHomeAssistantVersionError,
+    AddonNotSupportedMachineTypeError,
     SupervisorAuthenticationError,
     SupervisorBadRequestError,
     SupervisorConnectionError,
@@ -14,14 +18,18 @@
 from aiohasupervisor.root import SupervisorClient
 
 __all__ = [
-    "SupervisorError",
-    "SupervisorConnectionError",
+    "AddonNotSupportedArchitectureError",
+    "AddonNotSupportedError",
+    "AddonNotSupportedHomeAssistantVersionError",
+    "AddonNotSupportedMachineTypeError",
     "SupervisorAuthenticationError",
     "SupervisorBadRequestError",
+    "SupervisorClient",
+    "SupervisorConnectionError",
+    "SupervisorError",
     "SupervisorForbiddenError",
     "SupervisorNotFoundError",
     "SupervisorResponseError",
     "SupervisorServiceUnavailableError",
     "SupervisorTimeoutError",
-    "SupervisorClient",
 ]

aiohasupervisor/client.py

@@ -17,6 +17,7 @@
 
 from .const import DEFAULT_TIMEOUT, ResponseType
 from .exceptions import (
+    ERROR_KEYS,
     SupervisorAuthenticationError,
     SupervisorBadRequestError,
     SupervisorConnectionError,
@@ -73,7 +74,13 @@ async def _raise_on_status(self, response: ClientResponse) -> None:
 
             if is_json(response):
                 result = Response.from_json(await response.text())
-                raise exc_type(result.message, result.job_id)
+                if result.error_key in ERROR_KEYS:
+                    exc_type = ERROR_KEYS[result.error_key]
+                raise exc_type(
+                    result.message,
+                    result.extra_fields,
+                    result.job_id,
+                )
             raise exc_type()
 
     async def _request(

aiohasupervisor/exceptions.py

@@ -1,17 +1,45 @@
 """Exceptions from supervisor client."""
 
+from abc import ABC
+from collections.abc import Callable
+from typing import Any
+
 
 class SupervisorError(Exception):
     """Generic exception."""
 
-    def __init__(self, message: str | None = None, job_id: str | None = None) -> None:
+    error_key: str | None = None
+
+    def __init__(
+        self,
+        message: str | None = None,
+        extra_fields: dict[str, Any] | None = None,
+        job_id: str | None = None,
+    ) -> None:
         """Initialize exception."""
         if message is not None:
             super().__init__(message)
         else:
             super().__init__()
 
-        self.job_id: str | None = job_id
+        self.job_id = job_id
+        self.extra_fields = extra_fields
+
+
+ERROR_KEYS: dict[str, type[SupervisorError]] = {}
+
+
+def error_key(
+    key: str,
+) -> Callable[[type[SupervisorError]], type[SupervisorError]]:
+    """Store exception in keyed error map."""
+
+    def wrap(cls: type[SupervisorError]) -> type[SupervisorError]:
+        ERROR_KEYS[key] = cls
+        cls.error_key = key
+        return cls
+
+    return wrap
 
 
 class SupervisorConnectionError(SupervisorError, ConnectionError):
@@ -44,3 +72,22 @@ class SupervisorServiceUnavailableError(SupervisorError):
 
 class SupervisorResponseError(SupervisorError):
     """Unusable response received from Supervisor with the wrong type or encoding."""
+
+
+class AddonNotSupportedError(SupervisorError, ABC):
+    """Addon is not supported on this system."""
+
+
+@error_key("addon_not_supported_architecture_error")
+class AddonNotSupportedArchitectureError(AddonNotSupportedError):
+    """Addon is not supported on this system due to its architecture."""
+
+
+@error_key("addon_not_supported_machine_type_error")
+class AddonNotSupportedMachineTypeError(AddonNotSupportedError):
+    """Addon is not supported on this system due to its machine type."""
+
+
+@error_key("addon_not_supported_home_assistant_version_error")
+class AddonNotSupportedHomeAssistantVersionError(AddonNotSupportedError):
+    """Addon is not supported on this system due to its version of Home Assistant."""

aiohasupervisor/models/base.py

@@ -84,6 +84,8 @@ class Response(DataClassORJSONMixin):
     data: Any | None = None
     message: str | None = None
     job_id: str | None = None
+    error_key: str | None = None
+    extra_fields: dict[str, Any] | None = None
 
 
 @dataclass(frozen=True, slots=True)

aiohasupervisor/store.py

@@ -48,10 +48,31 @@ async def addon_documentation(self, addon: str) -> str:
         )
         return result.data
 
+    async def addon_availability(self, addon: str) -> None:
+        """Determine if latest version of addon can be installed on this system.
+
+        No return means it can be. If not, raises one of the following errors:
+        - AddonUnavailableHomeAssistantVersionError
+        - AddonUnavailableArchitectureError
+        - AddonUnavailableMachineTypeError
+
+        If Supervisor adds a new reason an add-on can be restricted from being
+        installed on some systems in the future, older versions of this client
+        will raise the generic SupervisorBadRequestError for that reason.
+        """
+        await self._client.get(
+            f"store/addons/{addon}/availability", response_type=ResponseType.NONE
+        )
+
     async def install_addon(
         self, addon: str, options: StoreAddonInstall | None = None
     ) -> None:
-        """Install an addon."""
+        """Install an addon.
+
+        Supervisor does an availability check before install. If the addon
+        cannot be installed on this system it will raise one of those errors
+        shown in the `addon_availability` method.
+        """
         # Must disable timeout if API call waits for install to complete
         kwargs: dict[str, Any] = {}
         if not options or not options.background:
@@ -66,7 +87,12 @@ async def install_addon(
     async def update_addon(
         self, addon: str, options: StoreAddonUpdate | None = None
     ) -> None:
-        """Update an addon to latest version."""
+        """Update an addon to latest version.
+
+        Supervisor does an availability check before update. If the new version
+        cannot be installed on this system it will raise one of those errors
+        shown in the `addon_availability` method.
+        """
         # Must disable timeout if API call waits for update to complete
         kwargs: dict[str, Any] = {}
         if not options or not options.background:

tests/fixtures/store_addon_availability_error_architecture.json

@@ -0,0 +1,9 @@
+{
+  "result": "error",
+  "message": "Add-on core_mosquitto not supported on this platform, supported architectures: i386",
+  "error_key": "addon_not_supported_architecture_error",
+  "extra_fields": {
+    "slug": "core_mosquitto",
+    "architectures": "i386"
+  }
+}

tests/fixtures/store_addon_availability_error_home_assistant.json

@@ -0,0 +1,9 @@
+{
+  "result": "error",
+  "message": "Add-on core_mosquitto not supported on this system, requires Home Assistant version 2023.1.1 or greater",
+  "error_key": "addon_not_supported_home_assistant_version_error",
+  "extra_fields": {
+    "slug": "core_mosquitto",
+    "version": "2023.1.1"
+  }
+}

tests/fixtures/store_addon_availability_error_machine.json

@@ -0,0 +1,9 @@
+{
+  "result": "error",
+  "message": "Add-on core_mosquitto not supported on this machine, supported machine types: odroid-n2",
+  "error_key": "addon_not_supported_machine_type_error",
+  "extra_fields": {
+    "slug": "core_mosquitto",
+    "machine_types": "odroid-n2"
+  }
+}

tests/fixtures/store_addon_availability_error_other.json

@@ -0,0 +1,9 @@
+{
+  "result": "error",
+  "message": "Add-on core_mosquitto not supported on this system, requires <something new> to be <something else>",
+  "error_key": "addon_not_supported_other_error",
+  "extra_fields": {
+    "slug": "core_mosquitto",
+    "requirement": "<something else>"
+  }
+}

tests/test_store.py

@@ -1,10 +1,19 @@
 """Tests for store supervisor client."""
 
+from json import loads
+
 from aioresponses import aioresponses
 import pytest
 from yarl import URL
 
 from aiohasupervisor import SupervisorClient
+from aiohasupervisor.exceptions import (
+    AddonNotSupportedArchitectureError,
+    AddonNotSupportedHomeAssistantVersionError,
+    AddonNotSupportedMachineTypeError,
+    SupervisorBadRequestError,
+    SupervisorError,
+)
 from aiohasupervisor.models import StoreAddonUpdate, StoreAddRepository
 from aiohasupervisor.models.addons import StoreAddonInstall
 
@@ -169,6 +178,92 @@ async def test_store_addon_update(
     )
 
 
+async def test_store_addon_availability(
+    responses: aioresponses, supervisor_client: SupervisorClient
+) -> None:
+    """Test store addon availability API."""
+    responses.get(
+        f"{SUPERVISOR_URL}/store/addons/core_mosquitto/availability", status=200
+    )
+
+    assert (await supervisor_client.store.addon_availability("core_mosquitto")) is None
+
+
+@pytest.mark.parametrize(
+    ("error_fixture", "error_key", "exc_type"),
+    [
+        (
+            "store_addon_availability_error_architecture.json",
+            "addon_not_supported_architecture_error",
+            AddonNotSupportedArchitectureError,
+        ),
+        (
+            "store_addon_availability_error_machine.json",
+            "addon_not_supported_machine_type_error",
+            AddonNotSupportedMachineTypeError,
+        ),
+        (
+            "store_addon_availability_error_home_assistant.json",
+            "addon_not_supported_home_assistant_version_error",
+            AddonNotSupportedHomeAssistantVersionError,
+        ),
+        (
+            "store_addon_availability_error_other.json",
+            None,
+            SupervisorBadRequestError,
+        ),
+    ],
+)
+async def test_store_addon_availability_error(
+    responses: aioresponses,
+    supervisor_client: SupervisorClient,
+    error_fixture: str,
+    error_key: str | None,
+    exc_type: type[SupervisorError],
+) -> None:
+    """Test store addon availability errors."""
+    error_body = load_fixture(error_fixture)
+    error_data = loads(error_body)
+
+    def check_availability_error(err: SupervisorError) -> bool:
+        assert err.error_key == error_key
+        assert err.extra_fields == error_data["extra_fields"]
+        return True
+
+    # Availability API
+    responses.get(
+        f"{SUPERVISOR_URL}/store/addons/core_mosquitto/availability",
+        status=400,
+        body=error_body,
+    )
+    with pytest.raises(
+        exc_type, match=error_data["message"], check=check_availability_error
+    ):
+        await supervisor_client.store.addon_availability("core_mosquitto")
+
+    # Install API
+    responses.post(
+        f"{SUPERVISOR_URL}/store/addons/core_mosquitto/install",
+        status=400,
+        body=error_body,
+    )
+    with pytest.raises(
+        exc_type, match=error_data["message"], check=check_availability_error
+    ):
+        await supervisor_client.store.install_addon("core_mosquitto")
+
+    # Update API
+    responses.post(
+        f"{SUPERVISOR_URL}/store/addons/core_mosquitto/update",
+        status=400,
+        body=error_body,
+    )
+    with pytest.raises(
+        exc_type, match=error_data["message"], check=check_availability_error
+    ):
+        await supervisor_client.store.update_addon("core_mosquitto")
+
+
 async def test_store_reload(
     responses: aioresponses, supervisor_client: SupervisorClient
 ) -> None: