Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions src/wled/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,8 @@
WLEDEmptyResponseError,
WLEDError,
WLEDInvalidResponseError,
WLEDResponseError,
WLEDStatusError,
WLEDUnsupportedVersionError,
WLEDUpgradeError,
)
Expand Down Expand Up @@ -66,6 +68,8 @@
"WLEDError",
"WLEDInvalidResponseError",
"WLEDReleases",
"WLEDResponseError",
"WLEDStatusError",
"WLEDUnsupportedVersionError",
"WLEDUpgradeError",
"Wifi",
Expand Down
50 changes: 42 additions & 8 deletions src/wled/exceptions.py
Original file line number Diff line number Diff line change
Expand Up @@ -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."""

Expand All @@ -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."""

Expand Down
54 changes: 42 additions & 12 deletions src/wled/wled.py
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@
WLEDEmptyResponseError,
WLEDError,
WLEDInvalidResponseError,
WLEDStatusError,
WLEDUpgradeError,
)
from .models import Device, Playlist, Preset, Releases
Expand Down Expand Up @@ -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:
Expand All @@ -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)
Expand Down Expand Up @@ -232,26 +241,43 @@ 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:
msg = (
"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)
Expand All @@ -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
Expand Down Expand Up @@ -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:
Expand Down
48 changes: 45 additions & 3 deletions tests/test_wled.py
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@
WLEDEmptyResponseError,
WLEDError,
WLEDInvalidResponseError,
WLEDStatusError,
WLEDUpgradeError,
)
from wled.wled import WLEDReleases
Expand Down Expand Up @@ -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"),
[
Expand Down Expand Up @@ -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(
Expand All @@ -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(
Expand Down Expand Up @@ -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"


# =========================================================================
Expand Down
Loading