Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
async def test_internal_session(aresponses):
"""Test JSON response is handled correctly."""
aresponses.add(
"example.com",
"/",
"GET",
aresponses.Response(
status=200,
headers={"Content-Type": "application/json"},
text='{"status": "ok"}',
),
)
async with WLED("example.com") as wled:
response = await wled._request("/")
assert response["status"] == "ok"
async def test_authenticated_request(aresponses):
"""Test JSON response is handled correctly."""
aresponses.add(
"example.com",
"/",
"GET",
aresponses.Response(
status=200,
headers={"Content-Type": "application/json"},
text='{"status": "ok"}',
),
)
async with aiohttp.ClientSession() as session:
wled = WLED(
"example.com", username="frenck", password="zerocool", session=session,
)
response = await wled._request("/")
assert response["status"] == "ok"
"example.com",
"/json/",
"GET",
aresponses.Response(
status=200,
headers={"Content-Type": "application/json"},
text=(
'{"state": {"on": true},'
'"effects": [], "palettes": [],'
'"info": {"ver": "0.9.1"}}'
),
),
)
async with aiohttp.ClientSession() as session:
wled = WLED("example.com", session=session)
await wled.update()
assert not wled._supports_si_request
headers={"Content-Type": "application/json"},
text='{"ver": "1.0"}',
),
)
aresponses.add(
"example.com",
"/json/state",
"GET",
aresponses.Response(
status=200,
headers={"Content-Type": "application/json"},
text='{"on": false}',
),
)
async with aiohttp.ClientSession() as session:
wled = WLED("example.com", session=session)
device = await wled.update()
assert device.state.on
device = await wled.update()
assert not device.state.on
async def test_request_custom_user_agent(aresponses):
"""Test WLED client sending correct user agent headers."""
# Handle to run asserts on request in
async def response_handler(request):
assert request.headers["User-Agent"] == "LoremIpsum/1.0"
return aresponses.Response(text="TEDDYBEAR", status=200)
aresponses.add("example.com", "/", "GET", response_handler)
async with aiohttp.ClientSession() as session:
wled = WLED(
"example.com", base_path="/", session=session, user_agent="LoremIpsum/1.0",
)
await wled._request("/")
async def test_request_base_path(aresponses):
"""Test WLED running on different base path."""
aresponses.add(
"example.com",
"/admin/status",
"GET",
aresponses.Response(text="OMG PUPPIES!", status=200),
)
async with aiohttp.ClientSession() as session:
wled = WLED("example.com", base_path="/admin", session=session)
response = await wled._request("status")
assert response == "OMG PUPPIES!"
async def test_backoff(aresponses):
"""Test requests are handled with retries."""
async def response_handler(_):
await asyncio.sleep(0.2)
return aresponses.Response(body="Goodmorning!")
aresponses.add(
"example.com", "/", "GET", response_handler, repeat=2,
)
aresponses.add(
"example.com", "/", "GET", aresponses.Response(status=200, text="OK")
)
async with aiohttp.ClientSession() as session:
wled = WLED("example.com", session=session, request_timeout=0.1)
response = await wled._request("/")
assert response == "OK"
async def test_request_port(aresponses):
"""Test WLED running on non-standard port."""
aresponses.add(
"example.com:3333",
"/",
"GET",
aresponses.Response(text="OMG PUPPIES!", status=200),
)
async with aiohttp.ClientSession() as session:
wled = WLED("example.com", port=3333, session=session)
response = await wled._request("/")
assert response == "OMG PUPPIES!"
async def main():
"""Show example on controlling your WLED device."""
async with WLED("wled-frenck.local") as led:
device = await led.update()
print(device.info.version)
# Turn strip on, full brightness
await led.light(on=True, brightness=255)
# pylint: disable=no-member # https://github.com/PyCQA/pylint/issues/3167
source = self.context.get("source")
# Request user input, unless we are preparing discovery flow
if user_input is None and not prepare:
if source == SOURCE_ZEROCONF:
return self._show_confirm_dialog()
return self._show_setup_form()
if source == SOURCE_ZEROCONF:
# pylint: disable=no-member # https://github.com/PyCQA/pylint/issues/3167
user_input[CONF_HOST] = self.context.get(CONF_HOST)
errors = {}
session = async_get_clientsession(self.hass)
wled = WLED(user_input[CONF_HOST], loop=self.hass.loop, session=session)
try:
device = await wled.update()
except WLEDConnectionError:
if source == SOURCE_ZEROCONF:
return self.async_abort(reason="connection_error")
errors["base"] = "connection_error"
return self._show_setup_form(errors)
# Check if already configured
mac_address = device.info.mac_address
for entry in self._async_current_entries():
if entry.data[CONF_MAC] == mac_address:
# This mac address is already configured
return self.async_abort(reason="already_configured")