How to use wled - 10 common examples

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()

response = await self._session.request(
                    method,
                    url,
                    auth=auth,
                    data=data,
                    json=json_data,
                    params=params,
                    headers=headers,
                    ssl=self.verify_ssl,
                )
        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"

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"