Add support for keyed errors

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,14 @@ 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.message_template,
+                    result.extra_fields,
+                    result.job_id,
+                )
             raise exc_type()
 
     async def _request(

aiohasupervisor/exceptions.py

@@ -1,17 +1,47 @@
 """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,
+        message_template: 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.message_template = message_template
+        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):
@@ -46,45 +76,20 @@ class SupervisorResponseError(SupervisorError):
     """Unusable response received from Supervisor with the wrong type or encoding."""
 
 
-class AddonNotSupportedError(SupervisorError):
+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."""
 
-    def __init__(
-        self, addon: str, architectures: str, job_id: str | None = None
-    ) -> None:
-        """Initialize exception."""
-        super().__init__(
-            f"Add-on {addon} not supported on this platform, "
-            f"supported architectures: {architectures}",
-            job_id,
-        )
-
 
+@error_key("addon_not_supported_machine_type_error")
 class AddonNotSupportedMachineTypeError(AddonNotSupportedError):
     """Addon is not supported on this system due to its machine type."""
 
-    def __init__(
-        self, addon: str, machine_types: str, job_id: str | None = None
-    ) -> None:
-        """Initialize exception."""
-        super().__init__(
-            f"Add-on {addon} not supported on this machine, "
-            f"supported machine types: {machine_types}",
-            job_id,
-        )
-
 
+@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."""
-
-    def __init__(self, addon: str, version: str, job_id: str | None = None) -> None:
-        """Initialize exception."""
-        super().__init__(
-            f"Add-on {addon} not supported on this system, "
-            f"requires Home Assistant version {version} or greater",
-            job_id,
-        )

aiohasupervisor/models/base.py

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

aiohasupervisor/store.py

@@ -4,13 +4,6 @@
 
 from .client import _SupervisorComponentClient
 from .const import ResponseType
-from .exceptions import (
-    AddonNotSupportedArchitectureError,
-    AddonNotSupportedError,
-    AddonNotSupportedHomeAssistantVersionError,
-    AddonNotSupportedMachineTypeError,
-    SupervisorBadRequestError,
-)
 from .models.addons import (
     Repository,
     StoreAddon,
@@ -77,35 +70,30 @@ async def addon_availability(self, addon: str) -> None:
 
         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 AddonNotSupportedError for that reason.
+        will raise the generic SupervisorBadRequestError for that reason.
         """
-        try:
-            await self._client.get(
-                f"store/addons/{addon}/availability", response_type=ResponseType.NONE
-            )
-        except SupervisorBadRequestError as err:
-            if match := RE_ADDON_UNAVAILABLE_ARCHITECTURE.match(str(err)):
-                raise AddonNotSupportedArchitectureError(
-                    match.group("addon"), match.group("architectures"), err.job_id
-                ) from None
-            if match := RE_ADDON_UNAVAILABLE_HOME_ASSISTANT.match(str(err)):
-                raise AddonNotSupportedHomeAssistantVersionError(
-                    match.group("addon"), match.group("version"), err.job_id
-                ) from None
-            if match := RE_ADDON_UNAVAILABLE_MACHINE_TYPE.match(str(err)):
-                raise AddonNotSupportedMachineTypeError(
-                    match.group("addon"), match.group("machine_types"), err.job_id
-                ) from None
-            raise AddonNotSupportedError(str(err), err.job_id) from None
+        await self._client.get(
+            f"store/addons/{addon}/availability", response_type=ResponseType.NONE
+        )
 
     async def install_addon(self, addon: str) -> 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.
+        """
         await self._client.post(f"store/addons/{addon}/install", timeout=None)
 
     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.
+        """
         await self._client.post(
             f"store/addons/{addon}/update",
             json=options.to_dict() if options else None,

tests/fixtures/store_addon_availability_error_architecture.json

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

tests/fixtures/store_addon_availability_error_home_assistant.json

@@ -0,0 +1,10 @@
+{
+  "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",
+  "message_template": "Add-on {slug} not supported on this system, requires Home Assistant version {version} or greater",
+  "extra_fields": {
+    "slug": "core_mosquitto",
+    "version": "2023.1.1"
+  }
+}

tests/fixtures/store_addon_availability_error_machine.json

@@ -0,0 +1,10 @@
+{
+  "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",
+  "message_template": "Add-on {slug} not supported on this machine, supported machine types: {machine_types}",
+  "extra_fields": {
+    "slug": "core_mosquitto",
+    "machine_types": "odroid-n2"
+  }
+}

tests/fixtures/store_addon_availability_error_other.json

@@ -0,0 +1,10 @@
+{
+  "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",
+  "message_template": "Add-on {slug} not supported on this system, requires <something new> to be {requirement}",
+  "extra_fields": {
+    "slug": "core_mosquitto",
+    "requirement": "<something else>"
+  }
+}

tests/test_store.py

@@ -1,15 +1,18 @@
 """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,
-    AddonNotSupportedError,
     AddonNotSupportedHomeAssistantVersionError,
     AddonNotSupportedMachineTypeError,
+    SupervisorBadRequestError,
+    SupervisorError,
 )
 from aiohasupervisor.models import StoreAddonUpdate, StoreAddRepository
 
@@ -151,46 +154,80 @@ async def test_store_addon_availability(
 
 
 @pytest.mark.parametrize(
-    ("error_msg", "exc_type"),
+    ("error_fixture", "error_key", "exc_type"),
     [
         (
-            "Add-on core_mosquitto not supported on this platform, "
-            "supported architectures: i386",
+            "store_addon_availability_error_architecture.json",
+            "addon_not_supported_architecture_error",
             AddonNotSupportedArchitectureError,
         ),
         (
-            "Add-on core_mosquitto not supported on this machine, "
-            "supported machine types: odroid-n2",
+            "store_addon_availability_error_machine.json",
+            "addon_not_supported_machine_type_error",
             AddonNotSupportedMachineTypeError,
         ),
         (
-            "Add-on core_mosquitto not supported on this system, "
-            "requires Home Assistant version 2023.1.1 or greater",
+            "store_addon_availability_error_home_assistant.json",
+            "addon_not_supported_home_assistant_version_error",
             AddonNotSupportedHomeAssistantVersionError,
         ),
         (
-            "Add-on core_mosquitto not supported on this system, "
-            "requires <something new> to be <something else>",
-            AddonNotSupportedError,
+            "store_addon_availability_error_other.json",
+            None,
+            SupervisorBadRequestError,
         ),
     ],
 )
 async def test_store_addon_availability_error(
     responses: aioresponses,
     supervisor_client: SupervisorClient,
-    error_msg: str,
-    exc_type: type[AddonNotSupportedError],
+    error_fixture: str,
+    error_key: str | None,
+    exc_type: type[SupervisorError],
 ) -> None:
-    """Test store addon availability API error."""
+    """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.message_template == error_data["message_template"]
+        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=f'{{"result": "error", "message": "{error_msg}"}}',
+        body=error_body,
     )
-
-    with pytest.raises(exc_type):
+    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