Skip to content
Open
Show file tree
Hide file tree
Changes from 1 commit
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
26 changes: 25 additions & 1 deletion src/wled/exceptions.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,13 +5,37 @@ class WLEDError(Exception):
"""Generic WLED exception."""


class WLEDEmptyResponseError(Exception):
class WLEDEmptyResponseError(WLEDError):
"""WLED empty API response exception."""

def __init__(
self,
message: str,
*,
method: str | None = None,
path: str | None = None,
) -> None:
Comment thread
mik-laj marked this conversation as resolved.
"""Initialize WLEDEmptyResponseError."""
super().__init__(message)
self.method = method
self.path = path
Comment thread
mik-laj marked this conversation as resolved.


class WLEDInvalidResponseError(WLEDError):
"""WLED invalid API response exception."""

def __init__(
self,
message: str,
*,
method: str | None = None,
path: str | None = None,
) -> None:
Comment thread
mik-laj marked this conversation as resolved.
"""Initialize WLEDInvalidResponseError."""
super().__init__(message)
self.method = method
self.path = path
Comment thread
mik-laj marked this conversation as resolved.


class WLEDConnectionError(WLEDError):
"""WLED connection exception."""
Expand Down
35 changes: 25 additions & 10 deletions src/wled/wled.py
Original file line number Diff line number Diff line change
Expand Up @@ -118,6 +118,10 @@ 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.
Comment thread
mik-laj marked this conversation as resolved.

"""
if not self._client or not self.connected or not self._device:
Expand All @@ -138,9 +142,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,7 +238,9 @@ async def request(
"Received an invalid JSON error response "
f"from request: {method} {uri}"
)
raise WLEDInvalidResponseError(msg) from exception
raise WLEDInvalidResponseError(
msg, method=method, path=uri
) from exception
raise WLEDError(response.status, error_body)
try:
message = contents.decode("utf-8")
Expand All @@ -241,7 +249,9 @@ async def request(
"Received a non-UTF-8 error response "
f"from request: {method} {uri}"
)
raise WLEDInvalidResponseError(msg) from exception
raise WLEDInvalidResponseError(
msg, method=method, path=uri
) from exception
raise WLEDError(
response.status,
{"message": message},
Expand All @@ -251,7 +261,9 @@ async def request(
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 +272,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 +311,24 @@ async def update(self) -> Device:
Raises
------
WLEDEmptyResponseError: The WLED device returned an empty response.
WLEDInvalidResponseError: The WLED device returned an invalid response.
Comment thread
mik-laj marked this conversation as resolved.

"""
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
14 changes: 11 additions & 3 deletions tests/test_wled.py
Original file line number Diff line number Diff line change
Expand Up @@ -246,8 +246,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 +273,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 +412,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