Handle errors returned by Somfy (#147)

pymfy/api/error.py

@@ -0,0 +1,68 @@
+from typing import Any, Dict
+
+
+class ServerException(Exception):
+    def __init__(self, response: Dict[str, Any]) -> None:
+        self.error_code = response["fault"]["detail"]["errorcode"]
+        self.fault_string = response["fault"]["faultstring"]
+        super().__init__()
+
+    def __str__(self) -> str:
+        return f"error_code: {self.error_code}, fault_string: {self.fault_string}"
+
+
+class InvalidAccessTokenException(ServerException):
+    pass
+
+
+class QuotaViolationException(ServerException):
+    pass
+
+
+class AccessTokenException(ServerException):
+    pass
+
+
+class ClientException(Exception):
+    def __init__(self, response: Dict[str, Any]) -> None:
+        self.data = response["data"]
+        self.message = response["message"]
+        super().__init__()
+
+    def __str__(self) -> str:
+        return f"message: {self.message}, data: {self.data}"
+
+
+class ValidateException(ClientException):
+    pass
+
+
+class DeviceNotFoundException(ClientException):
+    pass
+
+
+class DefinitionNotFoundException(ClientException):
+    pass
+
+
+class SiteNotFoundException(ClientException):
+    pass
+
+
+class SetupNotFoundException(ClientException):
+    pass
+
+
+SERVER_ERROR = {
+    "oauth.v2.InvalidAccessToken": InvalidAccessTokenException,
+    "keymanagement.service.access_token_expired": AccessTokenException,
+    "policies.ratelimit.QuotaViolation": QuotaViolationException,
+}
+
+CLIENT_ERROR = {
+    "ValidateError": ValidateException,
+    "device_not_found": DeviceNotFoundException,
+    "definition_not_found": DefinitionNotFoundException,
+    "setup_not_found": SetupNotFoundException,
+    "site_not_found": SiteNotFoundException,
+}

pymfy/api/somfy_api.py

@@ -6,6 +6,7 @@
 from requests_oauthlib import OAuth2Session
 
 from pymfy.api.devices.category import Category
+from pymfy.api.error import CLIENT_ERROR, SERVER_ERROR, ClientException, ServerException
 from pymfy.api.model import Command, Device, Site
 
 BASE_URL = "https://api.somfy.com/api/v1"
@@ -124,8 +125,24 @@ def _request(self, method: str, path: str, **kwargs: Any) -> Response:
         """
         url = f"{BASE_URL}{path}"
         try:
-            return getattr(self._oauth, method)(url, **kwargs)
+            response = getattr(self._oauth, method)(url, **kwargs)
         except TokenExpiredError:
             self._oauth.token = self.refresh_tokens()
 
-            return getattr(self._oauth, method)(url, **kwargs)
+            response = getattr(self._oauth, method)(url, **kwargs)
+
+        self._check_response(response)
+        return response
+
+    def _check_response(self, response: Response) -> None:
+        """Check response does not contain any error."""
+        if response.status_code == 200:
+            return
+        raw_content = response.text
+        if "fault" in raw_content:
+            error_code = response.json()["fault"]["detail"]["errorcode"]
+            raise SERVER_ERROR.get(error_code, ServerException)(response.json())
+        if "message" in raw_content:
+            message = response.json()["message"]
+            raise CLIENT_ERROR.get(message, ClientException)(response.json())
+        response.raise_for_status()

tests/data/access_token_expired.json

@@ -0,0 +1,8 @@
+{
+  "fault": {
+    "faultstring": "Access Token expired",
+    "detail": {
+      "errorcode": "keymanagement.service.access_token_expired"
+    }
+  }
+}

tests/data/definition_not_found.json

@@ -0,0 +1,5 @@
+{
+  "uid": "46779aebccfd",
+  "message": "definition_not_found",
+  "data": null
+}

tests/data/device_not_found.json

@@ -0,0 +1,6 @@
+{
+  "uid": "8b09d5a25940",
+  "message": "device_not_found",
+  "transactionId": "a2753340-c918-11eb-b171-a9ae9fdd7072",
+  "data": null
+}

tests/data/invalid_access_token.json

@@ -0,0 +1,8 @@
+{
+  "fault": {
+    "faultstring": "Invalid access token",
+    "detail": {
+      "errorcode": "oauth.v2.InvalidAccessToken"
+    }
+  }
+}

tests/data/quota_violation.json

@@ -0,0 +1,8 @@
+{
+  "fault": {
+    "detail": {
+      "errorcode": "policies.ratelimit.QuotaViolation"
+    },
+    "faultstring": "Rate limit quota violation. Quota limit exceeded."
+  }
+}

tests/data/setup_not_found.json

@@ -0,0 +1,6 @@
+{
+  "uid": "867eaca52921",
+  "message": "setup_not_found",
+  "transactionId": "4e296110-c90b-11eb-bbf3-619c6ebbae8b",
+  "data": null
+}

tests/data/site_not_found.json

@@ -0,0 +1,6 @@
+{
+  "uid": "fbd3dd6bab14",
+  "message": "site_not_found",
+  "transactionId": "30703f00-c919-11eb-b171-a9ae9fdd7072",
+  "data": null
+}

tests/data/validate_error.json

@@ -0,0 +1,12 @@
+{
+  "uid": "dd7bf078a314",
+  "message": "ValidateError",
+  "transactionId": "2fb262b0-c918-11eb-bbf3-619c6ebbae8b",
+  "data": {
+    "fields": {
+      "command.name": {
+        "message": "'name' is required"
+      }
+    }
+  }
+}

tests/test_somfy_api.py

@@ -3,8 +3,19 @@
 import httpretty
 import pytest
 from pytest import fixture
+from requests.models import HTTPError
 
 from pymfy.api.devices.category import Category
+from pymfy.api.error import (
+    AccessTokenException,
+    DefinitionNotFoundException,
+    DeviceNotFoundException,
+    InvalidAccessTokenException,
+    QuotaViolationException,
+    SetupNotFoundException,
+    SiteNotFoundException,
+    ValidateException,
+)
 from pymfy.api.model import Command, Parameter
 from pymfy.api.somfy_api import BASE_URL, SomfyApi
 
@@ -130,3 +141,33 @@ def test_user_agent(self, api, user_agent):
         httpretty.register_uri(httpretty.POST, url, body='{"job_id": "9"}')
         api.send_command("my-id", "close")
         assert httpretty.last_request().headers["user-agent"] == user_agent
+
+    @httpretty.activate
+    @pytest.mark.parametrize(
+        "api, error_file, exception",
+        [
+            (None, "access_token_expired", AccessTokenException),
+            (None, "definition_not_found", DefinitionNotFoundException),
+            (None, "device_not_found", DeviceNotFoundException),
+            (None, "invalid_access_token", InvalidAccessTokenException),
+            (None, "quota_violation", QuotaViolationException),
+            (None, "setup_not_found", SetupNotFoundException),
+            (None, "site_not_found", SiteNotFoundException),
+            (None, "validate_error", ValidateException),
+        ],
+        indirect=["api"],
+    )
+    def test_error(self, api, error_file, exception):
+        path = os.path.join(CURRENT_DIR, f"data/{error_file}.json")
+        with open(path) as error:
+            httpretty.register_uri(
+                httpretty.GET, f"{BASE_URL}/site", body=error.read(), status=400
+            )
+        with pytest.raises(exception):
+            api.get_sites()
+
+    @httpretty.activate
+    def test_unknown_error(self, api):
+        httpretty.register_uri(httpretty.GET, f"{BASE_URL}/site", body="", status=404)
+        with pytest.raises(HTTPError):
+            api.get_sites()