Skip to content
Open
Show file tree
Hide file tree
Changes from 3 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:
Comment thread
mik-laj marked this conversation as resolved.
"""Initialize WLEDStatusError."""
super().__init__(*args)
self.method = method
self.path = path
Comment thread
mik-laj marked this conversation as resolved.
self.status = status
self.body = body
Comment thread
mik-laj marked this conversation as resolved.


class WLEDResponseError(WLEDError):
"""WLED response error exception."""

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


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
48 changes: 34 additions & 14 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,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 +143,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 +239,36 @@ 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(
method=method, path=uri, status=response.status, body=error_body
)
Comment thread
mik-laj marked this conversation as resolved.
Comment thread
mik-laj marked this conversation as resolved.
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(
response.status,
{"message": message},
raise WLEDInvalidResponseError(
msg, method=method, path=uri
) from exception
raise WLEDStatusError(
method=method,
path=uri,
status=response.status,
body={"message": message},
)
Comment thread
mik-laj marked this conversation as resolved.

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 +277,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 +316,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