diff --git a/src/wled/__init__.py b/src/wled/__init__.py index d9ff9387..1b9d88e1 100644 --- a/src/wled/__init__.py +++ b/src/wled/__init__.py @@ -14,6 +14,8 @@ WLEDEmptyResponseError, WLEDError, WLEDInvalidResponseError, + WLEDResponseError, + WLEDStatusError, WLEDUnsupportedVersionError, WLEDUpgradeError, ) @@ -66,6 +68,8 @@ "WLEDError", "WLEDInvalidResponseError", "WLEDReleases", + "WLEDResponseError", + "WLEDStatusError", "WLEDUnsupportedVersionError", "WLEDUpgradeError", "Wifi", diff --git a/src/wled/exceptions.py b/src/wled/exceptions.py index 78ad6e24..9fc9ed11 100644 --- a/src/wled/exceptions.py +++ b/src/wled/exceptions.py @@ -5,14 +5,6 @@ class WLEDError(Exception): """Generic WLED exception.""" -class WLEDEmptyResponseError(Exception): - """WLED empty API response exception.""" - - -class WLEDInvalidResponseError(WLEDError): - """WLED invalid API response exception.""" - - class WLEDConnectionError(WLEDError): """WLED connection exception.""" @@ -25,6 +17,48 @@ class WLEDConnectionClosedError(WLEDConnectionError): """WLED WebSocket connection has been closed.""" +class WLEDStatusError(WLEDError): + """WLED HTTP status error exception (4xx/5xx status codes).""" + + def __init__( + self, + *args: object, + method: str | None = None, + path: str | None = None, + status: int | None = None, + body: object = None, + ) -> None: + """Initialize WLEDStatusError.""" + super().__init__(*args) + self.method = method + self.path = path + self.status = status + self.body = body + + +class WLEDResponseError(WLEDError): + """WLED response error exception.""" + + def __init__( + self, + *args: object, + method: str | None = None, + path: str | None = None, + ) -> None: + """Initialize WLEDResponseError.""" + super().__init__(*args) + self.method = method + self.path = path + + +class WLEDEmptyResponseError(WLEDResponseError): + """WLED empty API response exception.""" + + +class WLEDInvalidResponseError(WLEDResponseError): + """WLED invalid API response exception.""" + + class WLEDUnsupportedVersionError(WLEDError): """WLED version is unsupported.""" diff --git a/src/wled/wled.py b/src/wled/wled.py index 65b88ef6..51f74bfc 100644 --- a/src/wled/wled.py +++ b/src/wled/wled.py @@ -22,6 +22,7 @@ WLEDEmptyResponseError, WLEDError, WLEDInvalidResponseError, + WLEDStatusError, WLEDUpgradeError, ) from .models import Device, Playlist, Preset, Releases @@ -118,6 +119,12 @@ async def listen(self, callback: Callable[[Device], None]) -> None: to the WLED device. WLEDConnectionClosedError: The WebSocket connection to the remote WLED has been closed. + WLEDEmptyResponseError: The WLED device returned an empty response + when fetching presets. + WLEDInvalidResponseError: The WLED device returned an invalid response + when fetching presets. + WLEDStatusError: The WLED device returned a 4xx/5xx HTTP status + when fetching presets. """ if not self._client or not self.connected or not self._device: @@ -138,9 +145,11 @@ async def listen(self, callback: Callable[[Device], None]) -> None: if not (presets := await self.request("/presets.json")): msg = ( f"WLED device at {self.host} returned an empty API" - " response on presets update", + " response on presets update" + ) + raise WLEDEmptyResponseError( + msg, method="GET", path="/presets.json" ) - raise WLEDEmptyResponseError(msg) message_data["presets"] = presets device = self._device.update_from_dict(data=message_data) @@ -232,8 +241,17 @@ async def request( "Received an invalid JSON error response " f"from request: {method} {uri}" ) - raise WLEDInvalidResponseError(msg) from exception - raise WLEDError(response.status, error_body) + raise WLEDInvalidResponseError( + msg, method=method, path=uri + ) from exception + raise WLEDStatusError( + response.status, + error_body, + method=method, + path=uri, + status=response.status, + body=error_body, + ) try: message = contents.decode("utf-8") except UnicodeDecodeError as exception: @@ -241,17 +259,25 @@ async def request( "Received a non-UTF-8 error response " f"from request: {method} {uri}" ) - raise WLEDInvalidResponseError(msg) from exception - raise WLEDError( + raise WLEDInvalidResponseError( + msg, method=method, path=uri + ) from exception + raise WLEDStatusError( response.status, {"message": message}, + method=method, + path=uri, + status=response.status, + body={"message": message}, ) try: response_data = await response.text() except UnicodeDecodeError as exception: msg = f"Received a non-UTF-8 response from request: {method} {uri}" - raise WLEDInvalidResponseError(msg) from exception + raise WLEDInvalidResponseError( + msg, method=method, path=uri + ) from exception if "application/json" in content_type: try: response_data = orjson.loads(response_data) @@ -260,7 +286,9 @@ async def request( "Received an invalid JSON response " f"from request: {method} {uri}" ) - raise WLEDInvalidResponseError(msg) from exception + raise WLEDInvalidResponseError( + msg, method=method, path=uri + ) from exception except TimeoutError as exception: msg = f"Timeout occurred while connecting to WLED device at {self.host}" raise WLEDConnectionTimeoutError(msg) from exception @@ -297,23 +325,25 @@ async def update(self) -> Device: Raises ------ WLEDEmptyResponseError: The WLED device returned an empty response. + WLEDInvalidResponseError: The WLED device returned an invalid response. + WLEDStatusError: The WLED device returned a 4xx/5xx HTTP status. """ if not (data := await self.request("/json")): msg = ( f"WLED device at {self.host} returned an empty API" - " response on full update", + " response on full update" ) - raise WLEDEmptyResponseError(msg) + raise WLEDEmptyResponseError(msg, method="GET", path="/json") changed, new_version = self._check_presets_changed(data) if changed: if not (presets := await self.request("/presets.json")): msg = ( f"WLED device at {self.host} returned an empty API" - " response on presets update", + " response on presets update" ) - raise WLEDEmptyResponseError(msg) + raise WLEDEmptyResponseError(msg, method="GET", path="/presets.json") data["presets"] = presets if not self._device: diff --git a/tests/test_wled.py b/tests/test_wled.py index 1a4649fc..622ebf14 100644 --- a/tests/test_wled.py +++ b/tests/test_wled.py @@ -20,6 +20,7 @@ WLEDEmptyResponseError, WLEDError, WLEDInvalidResponseError, + WLEDStatusError, WLEDUpgradeError, ) from wled.wled import WLEDReleases @@ -150,6 +151,39 @@ async def test_http_error( assert await wled.request("/") +@pytest.mark.parametrize( + ("status", "body", "content_type", "expected_body"), + [ + (404, "Not Found", "text/plain", {"message": "Not Found"}), + (500, '{"error":"oops"}', "application/json", {"error": "oops"}), + ], + ids=["404-text", "500-json"], +) +async def test_http_error_raises_status_error( # noqa: PLR0913 # pylint: disable=too-many-arguments,too-many-positional-arguments + responses: aioresponses, + wled: WLED, + status: int, + body: str, + content_type: str, + expected_body: dict, +) -> None: + """Test HTTP error raises WLEDStatusError with structured attributes.""" + responses.get( + "http://example.com/json", + status=status, + body=body, + content_type=content_type, + ) + with pytest.raises(WLEDStatusError) as exc_info: + await wled.request("/json") + err = exc_info.value + assert err.method == "GET" + assert err.path == "/json" + assert err.status == status + assert err.body == expected_body + assert err.args == (status, expected_body) + + @pytest.mark.parametrize( ("body", "content_type"), [ @@ -246,8 +280,12 @@ async def test_update_corrupt_presets_response( body=body, content_type="application/json", ) - with pytest.raises(WLEDInvalidResponseError, match=r"GET /presets\.json"): + with pytest.raises( + WLEDInvalidResponseError, match=r"GET /presets\.json" + ) as exc_info: await wled.update() + assert exc_info.value.method == "GET" + assert exc_info.value.path == "/presets.json" async def test_update_empty_presets_response( @@ -269,8 +307,10 @@ async def test_update_empty_presets_response( content_type="text/plain", ) - with pytest.raises(WLEDEmptyResponseError): + with pytest.raises(WLEDEmptyResponseError) as exc_info: await wled.update() + assert exc_info.value.method == "GET" + assert exc_info.value.path == "/presets.json" async def test_update_skips_presets_when_unchanged( @@ -406,8 +446,10 @@ async def test_listen_preset_change_empty_response( content_type="text/plain", ) - with pytest.raises(WLEDEmptyResponseError): + with pytest.raises(WLEDEmptyResponseError) as exc_info: await wled.listen(MagicMock()) + assert exc_info.value.method == "GET" + assert exc_info.value.path == "/presets.json" # =========================================================================