Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
except asyncio.TimeoutError as exception:
raise WLEDConnectionTimeoutError(
"Timeout occurred while connecting to WLED device."
) from exception
except (aiohttp.ClientError, socket.gaierror) as exception:
raise WLEDConnectionError(
"Error occurred while communicating with WLED device."
) from exception
content_type = response.headers.get("Content-Type", "")
if (response.status // 100) in [4, 5]:
contents = await response.read()
response.close()
if content_type == "application/json":
raise WLEDError(response.status, json.loads(contents.decode("utf8")))
raise WLEDError(response.status, {"message": contents.decode("utf8")})
if "application/json" in content_type:
data = await response.json()
if (
method == "POST"
and uri == "state"
and self._device is not None
and json_data is not None
):
self._device.update_from_dict(data={"state": data})
return data
return await response.text()
raise WLEDConnectionTimeoutError(
"Timeout occurred while connecting to WLED device."
) from exception
except (aiohttp.ClientError, socket.gaierror) as exception:
raise WLEDConnectionError(
"Error occurred while communicating with WLED device."
) from exception
content_type = response.headers.get("Content-Type", "")
if (response.status // 100) in [4, 5]:
contents = await response.read()
response.close()
if content_type == "application/json":
raise WLEDError(response.status, json.loads(contents.decode("utf8")))
raise WLEDError(response.status, {"message": contents.decode("utf8")})
if "application/json" in content_type:
data = await response.json()
if (
method == "POST"
and uri == "state"
and self._device is not None
and json_data is not None
):
self._device.update_from_dict(data={"state": data})
return data
return await response.text()
async def test_http_error500(aresponses):
"""Test HTTP 500 response handling."""
aresponses.add(
"example.com",
"/",
"GET",
aresponses.Response(
body=b'{"status":"nok"}',
status=500,
headers={"Content-Type": "application/json"},
),
)
async with aiohttp.ClientSession() as session:
wled = WLED("example.com", session=session)
with pytest.raises(WLEDError):
assert await wled._request("/")
async def test_http_error400(aresponses):
"""Test HTTP 404 response handling."""
aresponses.add(
"example.com", "/", "GET", aresponses.Response(text="OMG PUPPIES!", status=404)
)
async with aiohttp.ClientSession() as session:
wled = WLED("example.com", session=session)
with pytest.raises(WLEDError):
assert await wled._request("/")
"""Exceptions for WLED."""
class WLEDError(Exception):
"""Generic WLED exception."""
class WLEDEmptyResponseError(Exception):
"""WLED empty API response exception."""
class WLEDConnectionError(WLEDError):
"""WLED connection exception."""
class WLEDConnectionTimeoutError(WLEDConnectionError):
"""WLED connection Timeout exception."""
def __init__(self, data: dict):
"""Initialize an empty WLED device class."""
# Check if all elements are in the passed dict, else raise an Error
if any(
k not in data and data[k] is not None
for k in ["effects", "palettes", "info", "state"]
):
raise WLEDError("WLED data is incomplete, cannot construct device object")
self.update_from_dict(data)
)
self._device = Device(data)
# Try to figure out if this version supports
# a single info and state call
try:
version.Version(self._device.info.version)
self._supports_si_request = version.parse(
self._device.info.version
) >= version.parse("0.10.0")
except version.InvalidVersion:
# Could be a manual build one? Lets poll for it
try:
await self._request("si")
self._supports_si_request = True
except WLEDError:
self._supports_si_request = False
return self._device
# Handle legacy state and update in separate requests
if not self._supports_si_request:
info = await self._request("info")
if not info:
raise WLEDEmptyResponseError(
"WLED device returned an empty API response on info update"
)
state = await self._request("state")
if not state:
raise WLEDEmptyResponseError(
"WLED device returned an empty API response on state update"
length: Optional[int] = None,
on: Optional[bool] = None,
palette: Optional[Union[int, str]] = None,
reverse: Optional[bool] = None,
selected: Optional[bool] = None,
speed: Optional[int] = None,
start: Optional[int] = None,
stop: Optional[int] = None,
transition: Optional[int] = None,
) -> None:
"""Change state of a WLED Light segment."""
if self._device is None:
await self.update()
if self._device is None:
raise WLEDError("Unable to communicate with WLED to get the current state")
state = {}
segment = {
"bri": brightness,
"cln": clones,
"fx": effect,
"ix": intensity,
"len": length,
"on": on,
"pal": palette,
"rev": reverse,
"sel": selected,
"start": start,
"stop": stop,
"sx": speed,
}