diff --git a/src/wled/exceptions.py b/src/wled/exceptions.py index a37c824a..78ad6e24 100644 --- a/src/wled/exceptions.py +++ b/src/wled/exceptions.py @@ -9,6 +9,10 @@ class WLEDEmptyResponseError(Exception): """WLED empty API response exception.""" +class WLEDInvalidResponseError(WLEDError): + """WLED invalid API response exception.""" + + class WLEDConnectionError(WLEDError): """WLED connection exception.""" diff --git a/src/wled/wled.py b/src/wled/wled.py index 12b9dbf0..afbd3e22 100644 --- a/src/wled/wled.py +++ b/src/wled/wled.py @@ -21,6 +21,7 @@ WLEDConnectionTimeoutError, WLEDEmptyResponseError, WLEDError, + WLEDInvalidResponseError, WLEDUpgradeError, ) from .models import Device, Playlist, Preset, Releases @@ -233,10 +234,20 @@ async def request( {"message": contents.decode("utf8")}, ) - response_data = await response.text() + 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 if "application/json" in content_type: - response_data = orjson.loads(response_data) - + try: + response_data = orjson.loads(response_data) + except orjson.JSONDecodeError as exception: + msg = ( + "Received an invalid JSON response " + f"from request: {method} {uri}" + ) + raise WLEDInvalidResponseError(msg) from exception except TimeoutError as exception: msg = f"Timeout occurred while connecting to WLED device at {self.host}" raise WLEDConnectionTimeoutError(msg) from exception diff --git a/tests/conftest.py b/tests/conftest.py index e58e7dc9..2d2f8b76 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -7,6 +7,7 @@ import aiohttp import pytest +import pytest_asyncio from aioresponses import aioresponses from wled import WLED @@ -28,18 +29,26 @@ def full_device_data() -> dict[str, Any]: return data -def mock_json_and_presets(mocked: aioresponses) -> None: +def mock_json_and_presets( + mocked: aioresponses, + wled_data: dict[str, Any] | None = None, + presets_data: dict[str, Any] | None = None, +) -> None: """Register the two GET endpoints that WLED.update() calls.""" + if wled_data is None: + wled_data = load_fixture_json("wled") mocked.get( "http://example.com/json", status=200, - body=json.dumps(load_fixture_json("wled")), + body=json.dumps(wled_data), content_type="application/json", ) + if presets_data is None: + presets_data = load_fixture_json("presets") mocked.get( "http://example.com/presets.json", status=200, - body=json.dumps(load_fixture_json("presets")), + body=json.dumps(presets_data), content_type="application/json", ) @@ -57,8 +66,16 @@ def responses() -> Generator[aioresponses, None, None]: yield mocker -@pytest.fixture +@pytest_asyncio.fixture +async def session() -> AsyncGenerator[aiohttp.ClientSession, None]: + """Yield a new aiohttp client session.""" + async with aiohttp.ClientSession() as aio_session: + yield aio_session + + +@pytest_asyncio.fixture async def wled() -> AsyncGenerator[WLED, None]: """Yield a WLED client wired to example.com with default settings.""" - async with aiohttp.ClientSession() as session: - yield WLED("example.com", session=session) + wled_instance = WLED("example.com") + yield wled_instance + await wled_instance.close() diff --git a/tests/test_wled.py b/tests/test_wled.py index 2226d04d..627ceff0 100644 --- a/tests/test_wled.py +++ b/tests/test_wled.py @@ -6,8 +6,10 @@ from unittest.mock import AsyncMock, MagicMock, patch import aiohttp +import orjson import pytest from aioresponses import aioresponses +from yarl import URL from wled import WLED, Device, Releases from wled.const import LiveDataOverride @@ -17,149 +19,138 @@ WLEDConnectionTimeoutError, WLEDEmptyResponseError, WLEDError, + WLEDInvalidResponseError, WLEDUpgradeError, ) from wled.wled import WLEDReleases from .conftest import full_device_data, load_fixture_json, mock_json_and_presets + +def assert_post_payload(mocked: aioresponses, path: str, expected: dict) -> None: + """Assert a POST request payload sent to WLED.""" + if not mocked.requests or not ( + requests := mocked.requests.get(("POST", URL(path))) + ): + msg = f"No POST request made to {path}" + raise AssertionError(msg) + request_call = requests[-1] + assert orjson.loads(request_call.kwargs["data"]) == expected + + # ========================================================================= # Section 1: Existing tests (preserved) # ========================================================================= -async def test_json_request() -> None: +async def test_json_request(responses: aioresponses, wled: WLED) -> None: """Test JSON response is handled correctly.""" - with aioresponses() as mocked: - mocked.get( - "http://example.com/", - status=200, - body='{"status": "ok"}', - content_type="application/json", - ) - async with aiohttp.ClientSession() as session: - wled = WLED("example.com", session=session) - response = await wled.request("/") - assert response["status"] == "ok" + responses.get( + "http://example.com/", + status=200, + body='{"status": "ok"}', + content_type="application/json", + ) + + response = await wled.request("/") + + assert response["status"] == "ok" -async def test_text_request() -> None: +async def test_text_request(responses: aioresponses, wled: WLED) -> None: """Test non-JSON response is handled correctly.""" - with aioresponses() as mocked: - mocked.get( - "http://example.com/", - status=200, - body="OK", - content_type="text/plain", - ) - async with aiohttp.ClientSession() as session: - wled = WLED("example.com", session=session) - response = await wled.request("/") - assert response == "OK" + responses.get( + "http://example.com/", + status=200, + body="OK", + content_type="text/plain", + ) + + response = await wled.request("/") + + assert response == "OK" -async def test_internal_session() -> None: +async def test_internal_session(responses: aioresponses) -> None: """Test internal session is created and works correctly.""" - with aioresponses() as mocked: - mocked.get( - "http://example.com/", - status=200, - body='{"status": "ok"}', - content_type="application/json", - ) - async with WLED("example.com") as wled: - response = await wled.request("/") - assert response["status"] == "ok" + responses.get( + "http://example.com/", + status=200, + body='{"status": "ok"}', + content_type="application/json", + ) + async with WLED("example.com") as wled: + response = await wled.request("/") + assert response["status"] == "ok" -async def test_post_request() -> None: +async def test_post_request(responses: aioresponses, wled: WLED) -> None: """Test POST requests are handled correctly.""" - with aioresponses() as mocked: - mocked.post( - "http://example.com/", - status=200, - body="OK", - content_type="text/plain", - ) - async with aiohttp.ClientSession() as session: - wled = WLED("example.com", session=session) - response = await wled.request("/", method="POST") - assert response == "OK" + responses.post( + "http://example.com/", + status=200, + body="OK", + content_type="text/plain", + ) + response = await wled.request("/", method="POST") -async def test_backoff() -> None: + assert response == "OK" + + +async def test_backoff(responses: aioresponses, wled: WLED) -> None: """Test requests are handled with retries.""" - with aioresponses() as mocked: - mocked.get( - "http://example.com/", - exception=TimeoutError(), - ) - mocked.get( - "http://example.com/", - exception=TimeoutError(), - ) - mocked.get( - "http://example.com/", - status=200, - body="OK", - content_type="text/plain", - ) - async with aiohttp.ClientSession() as session: - wled = WLED("example.com", session=session, request_timeout=0.1) - response = await wled.request("/") - assert response == "OK" + responses.get("http://example.com/", exception=TimeoutError()) + responses.get("http://example.com/", exception=TimeoutError()) + responses.get( + "http://example.com/", + status=200, + body="OK", + content_type="text/plain", + ) + wled.request_timeout = 0.1 + + response = await wled.request("/") + assert response == "OK" -async def test_timeout() -> None: + +async def test_timeout(responses: aioresponses, wled: WLED) -> None: """Test request timeout from WLED.""" - with aioresponses() as mocked: - # Backoff will try 3 times - mocked.get( - "http://example.com/", - exception=TimeoutError(), - ) - mocked.get( - "http://example.com/", - exception=TimeoutError(), - ) - mocked.get( - "http://example.com/", - exception=TimeoutError(), - ) - async with aiohttp.ClientSession() as session: - wled = WLED("example.com", session=session, request_timeout=0.1) - with pytest.raises(WLEDConnectionError): - assert await wled.request("/") + # Backoff will try 3 times + responses.get("http://example.com/", exception=TimeoutError()) + responses.get("http://example.com/", exception=TimeoutError()) + responses.get("http://example.com/", exception=TimeoutError()) + wled.request_timeout = 0.1 + + with pytest.raises(WLEDConnectionError): + assert await wled.request("/") -async def test_http_error404() -> None: +async def test_http_error404(responses: aioresponses, wled: WLED) -> None: """Test HTTP 404 response handling.""" - with aioresponses() as mocked: - mocked.get( - "http://example.com/", - status=404, - body="OMG PUPPIES!", - content_type="text/plain", - ) - async with aiohttp.ClientSession() as session: - wled = WLED("example.com", session=session) - with pytest.raises(WLEDError): - assert await wled.request("/") + responses.get( + "http://example.com/", + status=404, + body="OMG PUPPIES!", + content_type="text/plain", + ) + with pytest.raises(WLEDError): + assert await wled.request("/") -async def test_http_error500() -> None: + +async def test_http_error500(responses: aioresponses, wled: WLED) -> None: """Test HTTP 500 response handling.""" - with aioresponses() as mocked: - mocked.get( - "http://example.com/", - status=500, - body='{"status":"nok"}', - content_type="application/json", - ) - async with aiohttp.ClientSession() as session: - wled = WLED("example.com", session=session) - with pytest.raises(WLEDError): - assert await wled.request("/") + responses.get( + "http://example.com/", + status=500, + body='{"status":"nok"}', + content_type="application/json", + ) + + with pytest.raises(WLEDError): + assert await wled.request("/") # ========================================================================= @@ -167,174 +158,184 @@ async def test_http_error500() -> None: # ========================================================================= -async def test_update_creates_device() -> None: +async def test_update_creates_device(responses: aioresponses, wled: WLED) -> None: """Test that update() creates a Device from API responses.""" - with aioresponses() as mocked: - mock_json_and_presets(mocked) - async with aiohttp.ClientSession() as session: - wled = WLED("example.com", session=session) - device = await wled.update() - assert isinstance(device, Device) - assert device.info.name == "WLED" - assert device.state.on is True + mock_json_and_presets(responses) + + device = await wled.update() + + assert isinstance(device, Device) + assert device.info.name == "WLED" + assert device.state.on is True -async def test_update_uses_existing_device() -> None: +async def test_update_uses_existing_device(responses: aioresponses, wled: WLED) -> None: """Test that subsequent update() calls use update_from_dict.""" - with aioresponses() as mocked: - mock_json_and_presets(mocked) - mock_json_and_presets(mocked) - async with aiohttp.ClientSession() as session: - wled = WLED("example.com", session=session) - device1 = await wled.update() - device2 = await wled.update() - assert device1 is device2 + mock_json_and_presets(responses) + mock_json_and_presets(responses) + device1 = await wled.update() + device2 = await wled.update() -async def test_update_empty_json_response() -> None: - """Test update() raises on empty /json response.""" - with aioresponses() as mocked: - # Backoff on update() retries 3 times for WLEDEmptyResponseError - for _ in range(3): - mocked.get( - "http://example.com/json", - status=200, - body="", - content_type="text/plain", - ) - async with aiohttp.ClientSession() as session: - wled = WLED("example.com", session=session) - with pytest.raises(WLEDEmptyResponseError): - await wled.update() - - -async def test_update_empty_presets_response() -> None: - """Test update() raises on empty /presets.json response.""" - with aioresponses() as mocked: - # Backoff on update() retries 3 times for WLEDEmptyResponseError - for _ in range(3): - mocked.get( - "http://example.com/json", - status=200, - body=json.dumps(load_fixture_json("wled")), - content_type="application/json", - ) - mocked.get( - "http://example.com/presets.json", - status=200, - body="", - content_type="text/plain", - ) - async with aiohttp.ClientSession() as session: - wled = WLED("example.com", session=session) - with pytest.raises(WLEDEmptyResponseError): - await wled.update() - - -async def test_update_skips_presets_when_unchanged() -> None: - """Test update() skips fetching presets.json when presets haven't changed.""" - wled_data = load_fixture_json("wled") + assert device1 is device2 - with aioresponses() as mocked: - # First update: fetches both /json and /presets.json - mocked.get( - "http://example.com/json", - status=200, - body=json.dumps(wled_data), - content_type="application/json", - ) - mocked.get( - "http://example.com/presets.json", - status=200, - body=json.dumps(load_fixture_json("presets")), - content_type="application/json", - ) - # Second update: same pmt and uptime, only /json fetched - mocked.get( + +async def test_update_empty_json_response(responses: aioresponses, wled: WLED) -> None: + """Test update() raises on empty /json response.""" + # Backoff on update() retries 3 times for WLEDEmptyResponseError + for _ in range(3): + responses.get( "http://example.com/json", status=200, - body=json.dumps(wled_data), - content_type="application/json", + body="", + content_type="text/plain", ) - # Third update: pmt changed, fetches /presets.json again - changed_data = json.loads(json.dumps(wled_data)) - changed_data["info"]["fs"]["pmt"] = 9999999999.0 - mocked.get( + + with pytest.raises(WLEDEmptyResponseError): + await wled.update() + + +@pytest.mark.parametrize("body", ["AAAA", b"\xff\xfe"]) +async def test_update_corrupt_json_response( + responses: aioresponses, wled: WLED, body: str | bytes +) -> None: + """Test update() raises on corrupt (invalid JSON or non-UTF-8) /json response.""" + responses.get( + "http://example.com/json", + status=200, + body=body, + content_type="application/json", + ) + with pytest.raises(WLEDInvalidResponseError): + await wled.update() + + +@pytest.mark.parametrize("body", ["AAAA", b"\xff\xfe"]) +async def test_update_corrupt_presets_response( + responses: aioresponses, wled: WLED, body: str | bytes +) -> None: + """Test update() raises on corrupt /presets.json response.""" + wled_data = load_fixture_json("wled") + responses.get( + "http://example.com/json", + status=200, + body=json.dumps(wled_data), + content_type="application/json", + ) + responses.get( + "http://example.com/presets.json", + status=200, + body=body, + content_type="application/json", + ) + with pytest.raises(WLEDInvalidResponseError): + await wled.update() + + +async def test_update_empty_presets_response( + responses: aioresponses, wled: WLED +) -> None: + """Test update() raises on empty /presets.json response.""" + # Backoff on update() retries 3 times for WLEDEmptyResponseError + for _ in range(3): + responses.get( "http://example.com/json", status=200, - body=json.dumps(changed_data), + body=json.dumps(load_fixture_json("wled")), content_type="application/json", ) - mocked.get( + responses.get( "http://example.com/presets.json", status=200, - body=json.dumps({"0": {}, "1": {"n": "Updated Preset"}}), - content_type="application/json", + body="", + content_type="text/plain", ) - async with aiohttp.ClientSession() as session: - wled = WLED("example.com", session=session) + with pytest.raises(WLEDEmptyResponseError): + await wled.update() - # First call: presets fetched - device = await wled.update() - assert device.presets[1].name == "My Preset" - # Second call: presets unchanged, not refetched - device = await wled.update() - assert device.presets[1].name == "My Preset" +async def test_update_skips_presets_when_unchanged( + responses: aioresponses, wled: WLED +) -> None: + """Test update() skips fetching presets.json when presets haven't changed.""" + wled_data = load_fixture_json("wled") - # Third call: pmt changed, presets refetched - device = await wled.update() - assert device.presets[1].name == "Updated Preset" + # First update: fetches both /json and /presets.json + responses.get( + "http://example.com/json", + status=200, + body=json.dumps(wled_data), + content_type="application/json", + ) + responses.get( + "http://example.com/presets.json", + status=200, + body=json.dumps(load_fixture_json("presets")), + content_type="application/json", + ) + # Second update: same pmt and uptime, only /json fetched + responses.get( + "http://example.com/json", + status=200, + body=json.dumps(wled_data), + content_type="application/json", + ) + # Third update: pmt changed, fetches /presets.json again + changed_data = json.loads(json.dumps(wled_data)) + changed_data["info"]["fs"]["pmt"] = 9999999999.0 + responses.get( + "http://example.com/json", + status=200, + body=json.dumps(changed_data), + content_type="application/json", + ) + responses.get( + "http://example.com/presets.json", + status=200, + body=json.dumps({"0": {}, "1": {"n": "Updated Preset"}}), + content_type="application/json", + ) + # First call: presets fetched + device = await wled.update() + assert device.presets[1].name == "My Preset" -async def test_update_refetches_presets_when_info_incomplete() -> None: + # Second call: presets unchanged, not refetched + device = await wled.update() + assert device.presets[1].name == "My Preset" + + # Third call: pmt changed, presets refetched + device = await wled.update() + assert device.presets[1].name == "Updated Preset" + + +async def test_update_refetches_presets_when_info_incomplete( + responses: aioresponses, wled: WLED +) -> None: """Test update() refetches presets when pmt is zero/missing.""" wled_data = load_fixture_json("wled") # Set pmt to 0 so version can't be determined wled_data["info"]["fs"]["pmt"] = 0 - with aioresponses() as mocked: - mocked.get( - "http://example.com/json", - status=200, - body=json.dumps(wled_data), - content_type="application/json", - ) - mocked.get( - "http://example.com/presets.json", - status=200, - body=json.dumps(load_fixture_json("presets")), - content_type="application/json", - ) - # Second call: still no fs, presets refetched again - mocked.get( - "http://example.com/json", - status=200, - body=json.dumps(wled_data), - content_type="application/json", - ) - mocked.get( - "http://example.com/presets.json", - status=200, - body=json.dumps(load_fixture_json("presets")), - content_type="application/json", - ) + mock_json_and_presets(responses, wled_data) + # Second call: still no fs, presets refetched again + mock_json_and_presets(responses, wled_data) - async with aiohttp.ClientSession() as session: - wled = WLED("example.com", session=session) - await wled.update() - # Without fs/pmt, every update refetches presets - await wled.update() + await wled.update() + # Without fs/pmt, every update refetches presets + await wled.update() -async def test_listen_preset_change_via_websocket() -> None: +async def test_listen_preset_change_via_websocket( + responses: aioresponses, wled: WLED +) -> None: """Test listen() detects preset changes and refetches presets.json.""" wled_data = load_fixture_json("wled") - wled = WLED("example.com") mock_client = MagicMock() mock_client.closed = False + mock_client.close = AsyncMock() wled._client = mock_client # pylint: disable=protected-access wled._device = Device.from_dict(full_device_data()) # pylint: disable=protected-access @@ -348,28 +349,29 @@ async def test_listen_preset_change_via_websocket() -> None: mock_client.receive = AsyncMock(side_effect=[text_msg, close_msg]) - with aioresponses() as mocked: - mocked.get( - "http://example.com/presets.json", - status=200, - body=json.dumps(load_fixture_json("presets")), - content_type="application/json", - ) + responses.get( + "http://example.com/presets.json", + status=200, + body=json.dumps(load_fixture_json("presets")), + content_type="application/json", + ) - callback = MagicMock() - with pytest.raises(WLEDConnectionClosedError): - await wled.listen(callback) + callback = MagicMock() + with pytest.raises(WLEDConnectionClosedError): + await wled.listen(callback) - callback.assert_called_once() + callback.assert_called_once() -async def test_listen_preset_change_empty_response() -> None: +async def test_listen_preset_change_empty_response( + responses: aioresponses, wled: WLED +) -> None: """Test listen() raises when preset refetch returns empty.""" wled_data = load_fixture_json("wled") - wled = WLED("example.com") mock_client = MagicMock() mock_client.closed = False + mock_client.close = AsyncMock() wled._client = mock_client # pylint: disable=protected-access wled._device = Device.from_dict(full_device_data()) # pylint: disable=protected-access @@ -379,16 +381,15 @@ async def test_listen_preset_change_empty_response() -> None: mock_client.receive = AsyncMock(return_value=text_msg) - with aioresponses() as mocked: - mocked.get( - "http://example.com/presets.json", - status=200, - body="", - content_type="text/plain", - ) + responses.get( + "http://example.com/presets.json", + status=200, + body="", + content_type="text/plain", + ) - with pytest.raises(WLEDEmptyResponseError): - await wled.listen(MagicMock()) + with pytest.raises(WLEDEmptyResponseError): + await wled.listen(MagicMock()) # ========================================================================= @@ -396,60 +397,74 @@ async def test_listen_preset_change_empty_response() -> None: # ========================================================================= -async def test_master_brightness() -> None: +async def test_master_brightness(responses: aioresponses, wled: WLED) -> None: """Test setting master brightness.""" - with aioresponses() as mocked: - mocked.post( - "http://example.com/json/state", - status=200, - body='{"on": true, "bri": 200}', - content_type="application/json", - ) - async with aiohttp.ClientSession() as session: - wled = WLED("example.com", session=session) - await wled.master(brightness=200) + responses.post( + "http://example.com/json/state", + status=200, + body='{"on": true, "bri": 200}', + content_type="application/json", + ) + await wled.master(brightness=200) -async def test_master_on() -> None: + assert_post_payload( + responses, + "http://example.com/json/state", + {"bri": 200, "v": True}, + ) + + +async def test_master_on(responses: aioresponses, wled: WLED) -> None: """Test setting master on/off.""" - with aioresponses() as mocked: - mocked.post( - "http://example.com/json/state", - status=200, - body='{"on": true}', - content_type="application/json", - ) - async with aiohttp.ClientSession() as session: - wled = WLED("example.com", session=session) - await wled.master(on=True) + responses.post( + "http://example.com/json/state", + status=200, + body='{"on": true}', + content_type="application/json", + ) + + await wled.master(on=True) + + assert_post_payload( + responses, + "http://example.com/json/state", + {"on": True, "v": True}, + ) -async def test_master_transition() -> None: +async def test_master_transition(responses: aioresponses, wled: WLED) -> None: """Test setting master transition.""" - with aioresponses() as mocked: - mocked.post( - "http://example.com/json/state", - status=200, - body='{"on": true}', - content_type="application/json", - ) - async with aiohttp.ClientSession() as session: - wled = WLED("example.com", session=session) - await wled.master(transition=10) + responses.post( + "http://example.com/json/state", + status=200, + body='{"on": true}', + content_type="application/json", + ) + + await wled.master(transition=10) + + assert_post_payload( + responses, "http://example.com/json/state", {"tt": 10, "v": True} + ) -async def test_master_all_params() -> None: +async def test_master_all_params(responses: aioresponses, wled: WLED) -> None: """Test setting all master parameters at once.""" - with aioresponses() as mocked: - mocked.post( - "http://example.com/json/state", - status=200, - body='{"on": true, "bri": 100}', - content_type="application/json", - ) - async with aiohttp.ClientSession() as session: - wled = WLED("example.com", session=session) - await wled.master(brightness=100, on=True, transition=5) + responses.post( + "http://example.com/json/state", + status=200, + body='{"on": true, "bri": 100}', + content_type="application/json", + ) + + await wled.master(brightness=100, on=True, transition=5) + + assert_post_payload( + responses, + "http://example.com/json/state", + {"bri": 100, "on": True, "tt": 5, "v": True}, + ) # ========================================================================= @@ -457,263 +472,386 @@ async def test_master_all_params() -> None: # ========================================================================= -async def _get_wled_with_device( - mocked: aioresponses, session: aiohttp.ClientSession +async def prepare_wled_with_device( + responses: aioresponses, + wled: WLED, + wled_data: dict | None = None, ) -> WLED: - """Create a WLED instance with a loaded device.""" - mock_json_and_presets(mocked) - wled = WLED("example.com", session=session) + """Prepare a WLED instance with a loaded device.""" + if wled_data is None: + wled_data = load_fixture_json("wled") + mock_json_and_presets(responses, wled_data) + await wled.update() return wled -async def test_segment_basic() -> None: +async def test_segment_basic(responses: aioresponses, wled: WLED) -> None: """Test basic segment control.""" - with aioresponses() as mocked: - async with aiohttp.ClientSession() as session: - wled = await _get_wled_with_device(mocked, session) - mocked.post( - "http://example.com/json/state", - status=200, - body="{}", - content_type="application/json", - ) - await wled.segment(0, brightness=200, on=True) - - -async def test_segment_effect_by_name() -> None: + await prepare_wled_with_device(responses, wled) + responses.post( + "http://example.com/json/state", + status=200, + body="{}", + content_type="application/json", + ) + + await wled.segment(0, brightness=200, on=True) + + assert_post_payload( + responses, + "http://example.com/json/state", + { + "seg": [ + {"bri": 200, "on": True, "id": 0}, + ], + "v": True, + }, + ) + + +async def test_segment_effect_by_name(responses: aioresponses, wled: WLED) -> None: """Test setting segment effect by name.""" - with aioresponses() as mocked: - async with aiohttp.ClientSession() as session: - wled = await _get_wled_with_device(mocked, session) - mocked.post( - "http://example.com/json/state", - status=200, - body="{}", - content_type="application/json", - ) - await wled.segment(0, effect="Blink") - - -async def test_segment_palette_by_name() -> None: + await prepare_wled_with_device(responses, wled) + responses.post( + "http://example.com/json/state", + status=200, + body="{}", + content_type="application/json", + ) + + await wled.segment(0, effect="Blink") + + assert_post_payload( + responses, + "http://example.com/json/state", + { + "seg": [ + {"fx": 1, "id": 0}, + ], + "v": True, + }, + ) + + +async def test_segment_palette_by_name(responses: aioresponses, wled: WLED) -> None: """Test setting segment palette by name.""" - with aioresponses() as mocked: - async with aiohttp.ClientSession() as session: - wled = await _get_wled_with_device(mocked, session) - mocked.post( - "http://example.com/json/state", - status=200, - body="{}", - content_type="application/json", - ) - await wled.segment(0, palette="Random Cycle") - - -async def test_segment_color_primary() -> None: + await prepare_wled_with_device(responses, wled) + responses.post( + "http://example.com/json/state", + status=200, + body="{}", + content_type="application/json", + ) + + await wled.segment(0, palette="Random Cycle") + + assert_post_payload( + responses, + "http://example.com/json/state", + { + "seg": [ + {"pal": 1, "id": 0}, + ], + "v": True, + }, + ) + + +async def test_segment_color_primary(responses: aioresponses, wled: WLED) -> None: """Test setting primary color.""" - with aioresponses() as mocked: - async with aiohttp.ClientSession() as session: - wled = await _get_wled_with_device(mocked, session) - mocked.post( - "http://example.com/json/state", - status=200, - body="{}", - content_type="application/json", - ) - await wled.segment(0, color_primary=(255, 0, 0)) - - -async def test_segment_color_secondary_only() -> None: + await prepare_wled_with_device(responses, wled) + responses.post( + "http://example.com/json/state", + status=200, + body="{}", + content_type="application/json", + ) + + await wled.segment(0, color_primary=(255, 0, 0)) + + assert_post_payload( + responses, + "http://example.com/json/state", + { + "seg": [ + {"col": [[255, 0, 0]], "id": 0}, + ], + "v": True, + }, + ) + + +async def test_segment_color_secondary_only( + responses: aioresponses, wled: WLED +) -> None: """Test setting secondary color fills primary from current state.""" - with aioresponses() as mocked: - async with aiohttp.ClientSession() as session: - wled = await _get_wled_with_device(mocked, session) - mocked.post( - "http://example.com/json/state", - status=200, - body="{}", - content_type="application/json", - ) - await wled.segment(0, color_secondary=(0, 255, 0)) - - -async def test_segment_color_tertiary_only() -> None: + await prepare_wled_with_device(responses, wled) + responses.post( + "http://example.com/json/state", + status=200, + body="{}", + content_type="application/json", + ) + + await wled.segment(0, color_secondary=(0, 255, 0)) + + assert_post_payload( + responses, + "http://example.com/json/state", + { + "seg": [ + { + "col": [ + [255, 159, 0], + [0, 255, 0], + ], + "id": 0, + } + ], + "v": True, + }, + ) + + +async def test_segment_color_tertiary_only(responses: aioresponses, wled: WLED) -> None: """Test setting tertiary color fills primary and secondary.""" - with aioresponses() as mocked: - async with aiohttp.ClientSession() as session: - wled = await _get_wled_with_device(mocked, session) - mocked.post( - "http://example.com/json/state", - status=200, - body="{}", - content_type="application/json", - ) - await wled.segment(0, color_tertiary=(0, 0, 255)) - - -async def test_segment_all_colors() -> None: + await prepare_wled_with_device(responses, wled) + responses.post( + "http://example.com/json/state", + status=200, + body="{}", + content_type="application/json", + ) + + await wled.segment(0, color_tertiary=(0, 0, 255)) + + assert_post_payload( + responses, + "http://example.com/json/state", + { + "seg": [ + { + "col": [ + [255, 159, 0], + [0, 0, 0], + [0, 0, 255], + ], + "id": 0, + } + ], + "v": True, + }, + ) + + +async def test_segment_all_colors(responses: aioresponses, wled: WLED) -> None: """Test setting all three colors at once.""" - with aioresponses() as mocked: - async with aiohttp.ClientSession() as session: - wled = await _get_wled_with_device(mocked, session) - mocked.post( - "http://example.com/json/state", - status=200, - body="{}", - content_type="application/json", - ) - await wled.segment( - 0, - color_primary=(255, 0, 0), - color_secondary=(0, 255, 0), - color_tertiary=(0, 0, 255), - ) - - -async def test_segment_with_transition() -> None: + await prepare_wled_with_device(responses, wled) + responses.post( + "http://example.com/json/state", + status=200, + body="{}", + content_type="application/json", + ) + + await wled.segment( + 0, + color_primary=(255, 0, 0), + color_secondary=(0, 255, 0), + color_tertiary=(0, 0, 255), + ) + + assert_post_payload( + responses, + "http://example.com/json/state", + { + "seg": [ + { + "col": [ + [255, 0, 0], + [0, 255, 0], + [0, 0, 255], + ], + "id": 0, + } + ], + "v": True, + }, + ) + + +async def test_segment_with_transition(responses: aioresponses, wled: WLED) -> None: """Test setting segment with transition.""" - with aioresponses() as mocked: - async with aiohttp.ClientSession() as session: - wled = await _get_wled_with_device(mocked, session) - mocked.post( - "http://example.com/json/state", - status=200, - body="{}", - content_type="application/json", - ) - await wled.segment(0, brightness=100, transition=5) - - -async def test_segment_calls_update_when_no_device() -> None: + await prepare_wled_with_device(responses, wled) + responses.post( + "http://example.com/json/state", + status=200, + body="{}", + content_type="application/json", + ) + + await wled.segment(0, brightness=100, transition=5) + + assert_post_payload( + responses, + "http://example.com/json/state", + { + "seg": [ + {"bri": 100, "id": 0}, + ], + "tt": 5, + "v": True, + }, + ) + + +async def test_segment_calls_update_when_no_device( + responses: aioresponses, wled: WLED +) -> None: """Test segment() calls update() if no device loaded.""" - with aioresponses() as mocked: - mock_json_and_presets(mocked) - mocked.post( - "http://example.com/json/state", - status=200, - body="{}", - content_type="application/json", - ) - async with aiohttp.ClientSession() as session: - wled = WLED("example.com", session=session) - await wled.segment(0, on=True) + mock_json_and_presets(responses) + responses.post( + "http://example.com/json/state", + status=200, + body="{}", + content_type="application/json", + ) + await wled.segment(0, on=True) -async def test_segment_no_device_raises() -> None: + assert_post_payload( + responses, + "http://example.com/json/state", + { + "seg": [ + {"on": True, "id": 0}, + ], + "v": True, + }, + ) + + +async def test_segment_no_device_raises(wled: WLED) -> None: """Test segment() raises if update cannot load device.""" - async with aiohttp.ClientSession() as session: - wled = WLED("example.com", session=session) - # Patch update to do nothing (leave _device as None) - with ( - patch.object(wled, "update", new_callable=AsyncMock), - pytest.raises(WLEDError, match="Unable to communicate"), - ): - await wled.segment(0, on=True) + # Patch update to do nothing (leave _device as None) + with ( + patch.object(wled, "update", new_callable=AsyncMock), + pytest.raises(WLEDError, match="Unable to communicate"), + ): + await wled.segment(0, on=True) -async def test_segment_individual() -> None: +async def test_segment_individual(responses: aioresponses, wled: WLED) -> None: """Test setting individual LED colors.""" - with aioresponses() as mocked: - async with aiohttp.ClientSession() as session: - wled = await _get_wled_with_device(mocked, session) - mocked.post( - "http://example.com/json/state", - status=200, - body="{}", - content_type="application/json", - ) - await wled.segment(0, individual=[(255, 0, 0), (0, 255, 0)]) - - -async def test_segment_color_tertiary_no_secondary_in_state() -> None: + await prepare_wled_with_device(responses, wled) + responses.post( + "http://example.com/json/state", + status=200, + body="{}", + content_type="application/json", + ) + + await wled.segment(0, individual=[(255, 0, 0), (0, 255, 0)]) + + assert_post_payload( + responses, + "http://example.com/json/state", + { + "seg": [ + {"i": [[255, 0, 0], [0, 255, 0]], "id": 0}, + ], + "v": True, + }, + ) + + +async def test_segment_color_tertiary_no_secondary_in_state( + responses: aioresponses, wled: WLED +) -> None: """Test tertiary color when segment has no secondary color in state.""" - with aioresponses() as mocked: - # Build device data where the segment color has no secondary - wled_data = load_fixture_json("wled") - wled_data["state"]["seg"][0]["col"] = [[255, 0, 0]] - mocked.get( - "http://example.com/json", - status=200, - body=json.dumps(wled_data), - content_type="application/json", - ) - mocked.get( - "http://example.com/presets.json", - status=200, - body=json.dumps(load_fixture_json("presets")), - content_type="application/json", - ) - async with aiohttp.ClientSession() as session: - wled = WLED("example.com", session=session) - await wled.update() - mocked.post( - "http://example.com/json/state", - status=200, - body="{}", - content_type="application/json", - ) - await wled.segment(0, color_tertiary=(0, 0, 255)) - - -async def test_segment_secondary_no_color_in_state() -> None: + # Build device data where the segment color has no secondary + wled_data = load_fixture_json("wled") + wled_data["state"]["seg"][0]["col"] = [[255, 0, 0]] + await prepare_wled_with_device(responses, wled, wled_data=wled_data) + responses.post( + "http://example.com/json/state", + status=200, + body="{}", + content_type="application/json", + ) + + await wled.segment(0, color_tertiary=(0, 0, 255)) + + assert_post_payload( + responses, + "http://example.com/json/state", + { + "seg": [ + {"col": [[255, 0, 0], [0, 0, 0], [0, 0, 255]], "id": 0}, + ], + "v": True, + }, + ) + + +async def test_segment_secondary_no_color_in_state( + responses: aioresponses, wled: WLED +) -> None: """Test secondary color when segment has no color at all in state.""" - with aioresponses() as mocked: - # Build device data where the segment has no col - wled_data = load_fixture_json("wled") - del wled_data["state"]["seg"][0]["col"] - mocked.get( - "http://example.com/json", - status=200, - body=json.dumps(wled_data), - content_type="application/json", - ) - mocked.get( - "http://example.com/presets.json", - status=200, - body=json.dumps(load_fixture_json("presets")), - content_type="application/json", - ) - async with aiohttp.ClientSession() as session: - wled = WLED("example.com", session=session) - await wled.update() - mocked.post( - "http://example.com/json/state", - status=200, - body="{}", - content_type="application/json", - ) - # color is None, so it should use (0,0,0) fallback - await wled.segment(0, color_secondary=(0, 255, 0)) - - -async def test_segment_tertiary_no_color_in_state() -> None: + # Build device data where the segment has no col + wled_data = load_fixture_json("wled") + del wled_data["state"]["seg"][0]["col"] + await prepare_wled_with_device(responses, wled, wled_data=wled_data) + responses.post( + "http://example.com/json/state", + status=200, + body="{}", + content_type="application/json", + ) + + # color is None, so it should use (0,0,0) fallback + await wled.segment(0, color_secondary=(0, 255, 0)) + + assert_post_payload( + responses, + "http://example.com/json/state", + { + "seg": [ + {"col": [[0, 0, 0], [0, 255, 0]], "id": 0}, + ], + "v": True, + }, + ) + + +async def test_segment_tertiary_no_color_in_state( + responses: aioresponses, wled: WLED +) -> None: """Test tertiary color when segment has no color at all in state.""" - with aioresponses() as mocked: - # Build device data where the segment has no col - wled_data = load_fixture_json("wled") - del wled_data["state"]["seg"][0]["col"] - mocked.get( - "http://example.com/json", - status=200, - body=json.dumps(wled_data), - content_type="application/json", - ) - mocked.get( - "http://example.com/presets.json", - status=200, - body=json.dumps(load_fixture_json("presets")), - content_type="application/json", - ) - async with aiohttp.ClientSession() as session: - wled = WLED("example.com", session=session) - await wled.update() - mocked.post( - "http://example.com/json/state", - status=200, - body="{}", - content_type="application/json", - ) - await wled.segment(0, color_tertiary=(0, 0, 255)) + # Build device data where the segment has no col + wled_data = load_fixture_json("wled") + del wled_data["state"]["seg"][0]["col"] + await prepare_wled_with_device(responses, wled, wled_data=wled_data) + responses.post( + "http://example.com/json/state", + status=200, + body="{}", + content_type="application/json", + ) + + await wled.segment(0, color_tertiary=(0, 0, 255)) + + assert_post_payload( + responses, + "http://example.com/json/state", + { + "seg": [ + {"col": [[0, 0, 0], [0, 0, 0], [0, 0, 255]], "id": 0}, + ], + "v": True, + }, + ) # ========================================================================= @@ -721,204 +859,311 @@ async def test_segment_tertiary_no_color_in_state() -> None: # ========================================================================= -async def test_preset_by_id() -> None: - """Test setting a preset by integer ID.""" - with aioresponses() as mocked: - async with aiohttp.ClientSession() as session: - wled = await _get_wled_with_device(mocked, session) - mocked.post( - "http://example.com/json/state", - status=200, - body="{}", - content_type="application/json", - ) - await wled.preset(1) - - -async def test_preset_by_name() -> None: - """Test setting a preset by name.""" - with aioresponses() as mocked: - async with aiohttp.ClientSession() as session: - wled = await _get_wled_with_device(mocked, session) - mocked.post( - "http://example.com/json/state", - status=200, - body="{}", - content_type="application/json", - ) - await wled.preset("My Preset") - - -async def test_preset_by_object() -> None: - """Test setting a preset using a Preset object.""" - with aioresponses() as mocked: - async with aiohttp.ClientSession() as session: - wled = await _get_wled_with_device(mocked, session) - mocked.post( - "http://example.com/json/state", - status=200, - body="{}", - content_type="application/json", - ) - assert wled._device is not None # pylint: disable=protected-access - preset_obj = wled._device.presets[1] # pylint: disable=protected-access - await wled.preset(preset_obj) - - -async def test_preset_name_not_found() -> None: - """Test setting a preset by name that does not exist passes string.""" - with aioresponses() as mocked: - async with aiohttp.ClientSession() as session: - wled = await _get_wled_with_device(mocked, session) - mocked.post( - "http://example.com/json/state", - status=200, - body="{}", - content_type="application/json", - ) - await wled.preset("NonExistent") - - -async def test_playlist_by_id() -> None: - """Test setting a playlist by integer ID.""" - with aioresponses() as mocked: - async with aiohttp.ClientSession() as session: - wled = await _get_wled_with_device(mocked, session) - mocked.post( - "http://example.com/json/state", - status=200, - body="{}", - content_type="application/json", - ) - await wled.playlist(2) - - -async def test_playlist_by_name() -> None: - """Test setting a playlist by name.""" - with aioresponses() as mocked: - async with aiohttp.ClientSession() as session: - wled = await _get_wled_with_device(mocked, session) - mocked.post( - "http://example.com/json/state", - status=200, - body="{}", - content_type="application/json", - ) - await wled.playlist("My Playlist") - - -async def test_playlist_by_object() -> None: - """Test setting a playlist using a Playlist object.""" - with aioresponses() as mocked: - async with aiohttp.ClientSession() as session: - wled = await _get_wled_with_device(mocked, session) - mocked.post( - "http://example.com/json/state", - status=200, - body="{}", - content_type="application/json", - ) - assert wled._device is not None # pylint: disable=protected-access - playlist_obj = wled._device.playlists[2] # pylint: disable=protected-access - await wled.playlist(playlist_obj) - - -async def test_playlist_name_not_found() -> None: - """Test setting a playlist by name that does not exist passes string.""" - with aioresponses() as mocked: - async with aiohttp.ClientSession() as session: - wled = await _get_wled_with_device(mocked, session) - mocked.post( - "http://example.com/json/state", - status=200, - body="{}", - content_type="application/json", - ) - await wled.playlist("NonExistent") - - -async def test_transition() -> None: - """Test setting default transition.""" - with aioresponses() as mocked: - mocked.post( - "http://example.com/json/state", - status=200, - body="{}", - content_type="application/json", - ) - async with aiohttp.ClientSession() as session: - wled = WLED("example.com", session=session) - await wled.transition(10) +async def test_preset_by_id(responses: aioresponses, wled: WLED) -> None: + """Test setting a preset by integer ID.""" + await prepare_wled_with_device(responses, wled) + responses.post( + "http://example.com/json/state", + status=200, + body="{}", + content_type="application/json", + ) + + await wled.preset(1) + + assert_post_payload( + responses, + "http://example.com/json/state", + { + "ps": 1, + "v": True, + }, + ) + + +async def test_preset_by_name(responses: aioresponses, wled: WLED) -> None: + """Test setting a preset by name.""" + await prepare_wled_with_device(responses, wled) + responses.post( + "http://example.com/json/state", + status=200, + body="{}", + content_type="application/json", + ) + + await wled.preset("My Preset") + + assert_post_payload( + responses, + "http://example.com/json/state", + { + "ps": 1, + "v": True, + }, + ) + + +async def test_preset_by_object(responses: aioresponses, wled: WLED) -> None: + """Test setting a preset using a Preset object.""" + await prepare_wled_with_device(responses, wled) + responses.post( + "http://example.com/json/state", + status=200, + body="{}", + content_type="application/json", + ) + assert wled._device is not None # pylint: disable=protected-access + preset_obj = wled._device.presets[1] # pylint: disable=protected-access + await wled.preset(preset_obj) + + assert_post_payload( + responses, + "http://example.com/json/state", + { + "ps": 1, + "v": True, + }, + ) + + +async def test_preset_name_not_found(responses: aioresponses, wled: WLED) -> None: + """Test setting a preset by name that does not exist passes string.""" + await prepare_wled_with_device(responses, wled) + responses.post( + "http://example.com/json/state", + status=200, + body="{}", + content_type="application/json", + ) + + await wled.preset("NonExistent") + + assert_post_payload( + responses, + "http://example.com/json/state", + { + "ps": "NonExistent", + "v": True, + }, + ) + + +async def test_playlist_by_id(responses: aioresponses, wled: WLED) -> None: + """Test setting a playlist by integer ID.""" + await prepare_wled_with_device(responses, wled) + responses.post( + "http://example.com/json/state", + status=200, + body="{}", + content_type="application/json", + ) + + await wled.playlist(2) + + assert_post_payload( + responses, + "http://example.com/json/state", + { + "ps": 2, + "v": True, + }, + ) + + +async def test_playlist_by_name(responses: aioresponses, wled: WLED) -> None: + """Test setting a playlist by name.""" + await prepare_wled_with_device(responses, wled) + responses.post( + "http://example.com/json/state", + status=200, + body="{}", + content_type="application/json", + ) + + await wled.playlist("My Playlist") + + assert_post_payload( + responses, + "http://example.com/json/state", + { + "ps": 2, + "v": True, + }, + ) + + +async def test_playlist_by_object(responses: aioresponses, wled: WLED) -> None: + """Test setting a playlist using a Playlist object.""" + await prepare_wled_with_device(responses, wled) + responses.post( + "http://example.com/json/state", + status=200, + body="{}", + content_type="application/json", + ) + assert wled._device is not None # pylint: disable=protected-access + playlist_obj = wled._device.playlists[2] # pylint: disable=protected-access + + await wled.playlist(playlist_obj) + + assert_post_payload( + responses, + "http://example.com/json/state", + { + "ps": 2, + "v": True, + }, + ) + + +async def test_playlist_name_not_found(responses: aioresponses, wled: WLED) -> None: + """Test setting a playlist by name that does not exist passes string.""" + await prepare_wled_with_device(responses, wled) + responses.post( + "http://example.com/json/state", + status=200, + body="{}", + content_type="application/json", + ) + + await wled.playlist("NonExistent") + + assert_post_payload( + responses, + "http://example.com/json/state", + { + "ps": "NonExistent", + "v": True, + }, + ) + + +async def test_transition(responses: aioresponses, wled: WLED) -> None: + """Test setting default transition.""" + responses.post( + "http://example.com/json/state", + status=200, + body="{}", + content_type="application/json", + ) + + await wled.transition(10) + + assert_post_payload( + responses, + "http://example.com/json/state", + {"transition": 10, "v": True}, + ) -async def test_live() -> None: +async def test_live(responses: aioresponses, wled: WLED) -> None: """Test setting live data override.""" - with aioresponses() as mocked: - mocked.post( - "http://example.com/json/state", - status=200, - body="{}", - content_type="application/json", - ) - async with aiohttp.ClientSession() as session: - wled = WLED("example.com", session=session) - await wled.live(LiveDataOverride.ON) + responses.post( + "http://example.com/json/state", + status=200, + body="{}", + content_type="application/json", + ) + + await wled.live(LiveDataOverride.ON) + + assert_post_payload( + responses, + "http://example.com/json/state", + { + "lor": LiveDataOverride.ON.value, + "v": True, + }, + ) -async def test_sync_send() -> None: +async def test_sync_send(responses: aioresponses, wled: WLED) -> None: """Test setting sync send.""" - with aioresponses() as mocked: - mocked.post( - "http://example.com/json/state", - status=200, - body="{}", - content_type="application/json", - ) - async with aiohttp.ClientSession() as session: - wled = WLED("example.com", session=session) - await wled.sync(send=True) + responses.post( + "http://example.com/json/state", + status=200, + body="{}", + content_type="application/json", + ) + + await wled.sync(send=True) + + assert_post_payload( + responses, + "http://example.com/json/state", + { + "udpn": {"send": True}, + "v": True, + }, + ) -async def test_sync_receive() -> None: +async def test_sync_receive(responses: aioresponses, wled: WLED) -> None: """Test setting sync receive.""" - with aioresponses() as mocked: - mocked.post( - "http://example.com/json/state", - status=200, - body="{}", - content_type="application/json", - ) - async with aiohttp.ClientSession() as session: - wled = WLED("example.com", session=session) - await wled.sync(receive=True) + responses.post( + "http://example.com/json/state", + status=200, + body="{}", + content_type="application/json", + ) + + await wled.sync(receive=True) + + assert_post_payload( + responses, + "http://example.com/json/state", + { + "udpn": {"recv": True}, + "v": True, + }, + ) -async def test_nightlight_on() -> None: +async def test_nightlight_on(responses: aioresponses, wled: WLED) -> None: """Test turning on nightlight.""" - with aioresponses() as mocked: - mocked.post( - "http://example.com/json/state", - status=200, - body="{}", - content_type="application/json", - ) - async with aiohttp.ClientSession() as session: - wled = WLED("example.com", session=session) - await wled.nightlight(on=True) + responses.post( + "http://example.com/json/state", + status=200, + body="{}", + content_type="application/json", + ) + + await wled.nightlight(on=True) + + assert_post_payload( + responses, + "http://example.com/json/state", + { + "nl": {"on": True}, + "v": True, + }, + ) -async def test_nightlight_all_params() -> None: +async def test_nightlight_all_params(responses: aioresponses, wled: WLED) -> None: """Test nightlight with all parameters.""" - with aioresponses() as mocked: - mocked.post( - "http://example.com/json/state", - status=200, - body="{}", - content_type="application/json", - ) - async with aiohttp.ClientSession() as session: - wled = WLED("example.com", session=session) - await wled.nightlight(duration=30, fade=True, on=True, target_brightness=50) + responses.post( + "http://example.com/json/state", + status=200, + body="{}", + content_type="application/json", + ) + + await wled.nightlight(duration=30, fade=True, on=True, target_brightness=50) + + assert_post_payload( + responses, + "http://example.com/json/state", + { + "nl": { + "dur": 30, + "fade": True, + "on": True, + "tbri": 50, + }, + "v": True, + }, + ) # ========================================================================= @@ -926,57 +1171,54 @@ async def test_nightlight_all_params() -> None: # ========================================================================= -async def test_reset() -> None: +async def test_reset(responses: aioresponses, wled: WLED) -> None: """Test reset method calls /reset.""" - with aioresponses() as mocked: - mocked.get( - "http://example.com/reset", - status=200, - body="OK", - content_type="text/plain", - ) - async with aiohttp.ClientSession() as session: - wled = WLED("example.com", session=session) - await wled.reset() + responses.get( + "http://example.com/reset", + status=200, + body="OK", + content_type="text/plain", + ) + await wled.reset() -async def test_close_with_internal_session() -> None: + +async def test_close_with_internal_session(responses: aioresponses) -> None: """Test close() closes internally created session.""" - with aioresponses() as mocked: - mocked.get( - "http://example.com/", - status=200, - body='{"status": "ok"}', - content_type="application/json", - ) - wled = WLED("example.com") - await wled.request("/") - assert wled.session is not None - assert wled._close_session is True # pylint: disable=protected-access - await wled.close() + responses.get( + "http://example.com/", + status=200, + body='{"status": "ok"}', + content_type="application/json", + ) + wled = WLED("example.com") + await wled.request("/") + assert wled.session is not None + assert wled._close_session is True # pylint: disable=protected-access + await wled.close() -async def test_close_with_external_session() -> None: +async def test_close_with_external_session( + session: aiohttp.ClientSession, +) -> None: """Test close() does not close externally provided session.""" - async with aiohttp.ClientSession() as session: - wled = WLED("example.com", session=session) - assert wled._close_session is False # pylint: disable=protected-access - await wled.close() - assert not session.closed + wled = WLED("example.com", session=session) + assert wled._close_session is False # pylint: disable=protected-access + await wled.close() + assert not session.closed -async def test_context_manager() -> None: +async def test_context_manager(responses: aioresponses) -> None: """Test async context manager.""" - with aioresponses() as mocked: - mocked.get( - "http://example.com/", - status=200, - body='{"status": "ok"}', - content_type="application/json", - ) - async with WLED("example.com") as wled: - response = await wled.request("/") - assert response["status"] == "ok" + responses.get( + "http://example.com/", + status=200, + body='{"status": "ok"}', + content_type="application/json", + ) + async with WLED("example.com") as wled: + response = await wled.request("/") + assert response["status"] == "ok" async def test_connected_no_client() -> None: @@ -1018,60 +1260,53 @@ async def test_connect_already_connected() -> None: # Should return without doing anything -async def test_connect_no_websocket_support() -> None: +async def test_connect_no_websocket_support( + responses: aioresponses, wled: WLED +) -> None: """Test connect() raises when device has no WebSocket support.""" - with aioresponses() as mocked: - # Build data with ws=-1 (no websocket support) - wled_data = load_fixture_json("wled") - wled_data["info"]["ws"] = -1 - mocked.get( - "http://example.com/json", - status=200, - body=json.dumps(wled_data), - content_type="application/json", - ) - mocked.get( - "http://example.com/presets.json", - status=200, - body=json.dumps(load_fixture_json("presets")), - content_type="application/json", - ) - async with aiohttp.ClientSession() as session: - wled = WLED("example.com", session=session) - await wled.update() - with pytest.raises(WLEDError, match="does not support WebSockets"): - await wled.connect() + # Build data with ws=-1 (no websocket support) + wled_data = load_fixture_json("wled") + wled_data["info"]["ws"] = -1 + mock_json_and_presets(responses, wled_data) + await wled.update() + + with pytest.raises(WLEDError, match="does not support WebSockets"): + await wled.connect() -async def test_connect_connection_error() -> None: +async def test_connect_connection_error(responses: aioresponses, wled: WLED) -> None: """Test connect() raises WLEDConnectionError on connection failure.""" - with aioresponses() as mocked: - mock_json_and_presets(mocked) - async with aiohttp.ClientSession() as session: - wled = WLED("example.com", session=session) - await wled.update() - with ( - patch.object( - session, - "ws_connect", - side_effect=aiohttp.ClientConnectionError("fail"), - ), - pytest.raises(WLEDConnectionError), - ): - await wled.connect() - - -async def test_connect_calls_update_when_no_device() -> None: + mock_json_and_presets(responses) + + await wled.update() + assert wled.session is not None + with ( + patch.object( + wled.session, + "ws_connect", + side_effect=aiohttp.ClientConnectionError("fail"), + ), + pytest.raises(WLEDConnectionError), + ): + await wled.connect() + + +async def test_connect_calls_update_when_no_device( + responses: aioresponses, wled: WLED +) -> None: """Test connect() calls update() if no device is loaded.""" - with aioresponses() as mocked: - mock_json_and_presets(mocked) - async with aiohttp.ClientSession() as session: - wled = WLED("example.com", session=session) - # Device has ws=0, so connect should try to ws_connect - with patch.object(session, "ws_connect", new_callable=AsyncMock) as mock_ws: - mock_ws.return_value = MagicMock(closed=False) - await wled.connect() - mock_ws.assert_called_once() + mock_json_and_presets(responses) + + mock_client = MagicMock(closed=False) + mock_client.close = AsyncMock() + + with patch.object( + aiohttp.ClientSession, "ws_connect", new_callable=AsyncMock + ) as mock_ws: + mock_ws.return_value = mock_client + assert wled._device is None # pylint: disable=protected-access + await wled.connect() + assert wled._device is not None # pylint: disable=protected-access async def test_listen_not_connected() -> None: @@ -1160,42 +1395,37 @@ async def test_disconnect_not_connected() -> None: # ========================================================================= -async def test_post_state_adds_v_true() -> None: +async def test_post_state_adds_v_true(responses: aioresponses, wled: WLED) -> None: """Test POST to /json/state adds v=True to data.""" + mock_json_and_presets(responses) + state_response = json.dumps(load_fixture_json("wled")["state"]) - with aioresponses() as mocked: - mock_json_and_presets(mocked) - mocked.post( - "http://example.com/json/state", - status=200, - body=state_response, - content_type="application/json", - ) - async with aiohttp.ClientSession() as session: - wled = WLED("example.com", session=session) - await wled.update() # Need device for state update path - await wled.request("/json/state", method="POST", data={"on": True}) + responses.post( + "http://example.com/json/state", + status=200, + body=state_response, + content_type="application/json", + ) + await wled.update() # Need device for state update path + await wled.request("/json/state", method="POST", data={"on": True}) + + assert_post_payload( + responses, + "http://example.com/json/state", + {"on": True, "v": True}, + ) -async def test_client_error_raises_connection_error() -> None: +async def test_client_error_raises_connection_error( + responses: aioresponses, wled: WLED +) -> None: """Test aiohttp.ClientError raises WLEDConnectionError.""" - with aioresponses() as mocked: - mocked.get( - "http://example.com/test", - exception=aiohttp.ClientError("fail"), - ) - mocked.get( - "http://example.com/test", - exception=aiohttp.ClientError("fail"), - ) - mocked.get( - "http://example.com/test", - exception=aiohttp.ClientError("fail"), - ) - async with aiohttp.ClientSession() as session: - wled = WLED("example.com", session=session) - with pytest.raises(WLEDConnectionError): - await wled.request("/test") + responses.get("http://example.com/test", exception=aiohttp.ClientError("fail")) + responses.get("http://example.com/test", exception=aiohttp.ClientError("fail")) + responses.get("http://example.com/test", exception=aiohttp.ClientError("fail")) + + with pytest.raises(WLEDConnectionError): + await wled.request("/test") # ========================================================================= @@ -1203,9 +1433,9 @@ async def test_client_error_raises_connection_error() -> None: # ========================================================================= -async def _get_wled_with_device_for_upgrade( # pylint: disable=too-many-arguments, too-many-positional-arguments - mocked: aioresponses, - session: aiohttp.ClientSession, +async def prepare_wled_for_upgrade( # pylint: disable=too-many-arguments, too-many-positional-arguments + responses: aioresponses, + wled: WLED, arch: str = "esp32", version: str = "0.14.0", wifi_bssid: str = "AA:BB:CC:DD:EE:FF", @@ -1216,90 +1446,60 @@ async def _get_wled_with_device_for_upgrade( # pylint: disable=too-many-argumen wled_data["info"]["ver"] = version if wifi_bssid is not None: wled_data["info"]["wifi"]["bssid"] = wifi_bssid - mocked.get( - "http://example.com/json", - status=200, - body=json.dumps(wled_data), - content_type="application/json", - ) - mocked.get( - "http://example.com/presets.json", - status=200, - body=json.dumps(load_fixture_json("presets")), - content_type="application/json", - ) - wled = WLED("example.com", session=session) + + mock_json_and_presets(responses, wled_data) await wled.update() return wled -async def test_upgrade_unsupported_architecture() -> None: +async def test_upgrade_unsupported_architecture( + responses: aioresponses, wled: WLED +) -> None: """Test upgrade raises for unsupported architecture.""" - with aioresponses() as mocked: - async with aiohttp.ClientSession() as session: - wled = await _get_wled_with_device_for_upgrade( - mocked, session, arch="unknown_arch" - ) - with pytest.raises(WLEDUpgradeError, match="only supported"): - await wled.upgrade(version="0.15.0") + await prepare_wled_for_upgrade(responses, wled, arch="unknown_arch") + with pytest.raises(WLEDUpgradeError, match="only supported"): + await wled.upgrade(version="0.15.0") -async def test_upgrade_same_version() -> None: +async def test_upgrade_same_version(responses: aioresponses, wled: WLED) -> None: """Test upgrade raises when already on requested version.""" - with aioresponses() as mocked: - async with aiohttp.ClientSession() as session: - wled = await _get_wled_with_device_for_upgrade(mocked, session) - with pytest.raises(WLEDUpgradeError, match="already running"): - await wled.upgrade(version="0.14.0") + await prepare_wled_for_upgrade(responses, wled) + with pytest.raises(WLEDUpgradeError, match="already running"): + await wled.upgrade(version="0.14.0") -async def test_upgrade_no_version() -> None: +async def test_upgrade_no_version(responses: aioresponses, wled: WLED) -> None: """Test upgrade raises when current version is unknown.""" - with aioresponses() as mocked: - async with aiohttp.ClientSession() as session: - # Build device with invalid version - wled_data = load_fixture_json("wled") - wled_data["info"]["ver"] = "0.14.0" - mocked.get( - "http://example.com/json", - status=200, - body=json.dumps(wled_data), - content_type="application/json", - ) - mocked.get( - "http://example.com/presets.json", - status=200, - body=json.dumps(load_fixture_json("presets")), - content_type="application/json", - ) - wled = WLED("example.com", session=session) - await wled.update() - assert wled._device is not None # pylint: disable=protected-access - # Manually set version to None - wled._device.info.version = None # pylint: disable=protected-access - with pytest.raises(WLEDUpgradeError, match="version is unknown"): - await wled.upgrade(version="0.15.0") - - -async def test_upgrade_calls_update_when_no_device() -> None: + # Build device with invalid version + wled_data = load_fixture_json("wled") + wled_data["info"]["ver"] = "0.14.0" + mock_json_and_presets(responses, wled_data) + await wled.update() + assert wled._device is not None # pylint: disable=protected-access + # Manually set version to None + wled._device.info.version = None # pylint: disable=protected-access + with pytest.raises(WLEDUpgradeError, match="version is unknown"): + await wled.upgrade(version="0.15.0") + + +async def test_upgrade_calls_update_when_no_device( + responses: aioresponses, wled: WLED +) -> None: """Test upgrade() calls update() if no device loaded.""" - with aioresponses() as mocked: - mock_json_and_presets(mocked) - # Mock the download and upload - mocked.get( - "https://github.com/wled/WLED/releases/download/v0.15.0/WLED_0.15.0_ESP32.bin", - status=200, - body=b"fake firmware", - ) - mocked.post( - "http://example.com/update", - status=200, - body="OK", - content_type="text/plain", - ) - async with aiohttp.ClientSession() as session: - wled = WLED("example.com", session=session) - await wled.upgrade(version="0.15.0") + mock_json_and_presets(responses) + # Mock the download and upload + responses.get( + "https://github.com/wled/WLED/releases/download/v0.15.0/WLED_0.15.0_ESP32.bin", + status=200, + body=b"fake firmware", + ) + responses.post( + "http://example.com/update", + status=200, + body="OK", + content_type="text/plain", + ) + await wled.upgrade(version="0.15.0") async def test_upgrade_no_session_raises() -> None: @@ -1313,134 +1513,106 @@ async def test_upgrade_no_session_raises() -> None: await wled.upgrade(version="0.15.0") -async def test_upgrade_success() -> None: +async def test_upgrade_success(responses: aioresponses, wled: WLED) -> None: """Test successful upgrade.""" - with aioresponses() as mocked: - async with aiohttp.ClientSession() as session: - wled = await _get_wled_with_device_for_upgrade(mocked, session) - mocked.get( - "https://github.com/wled/WLED/releases/download/v0.15.0/WLED_0.15.0_ESP32.bin", - status=200, - body=b"fake firmware", - ) - mocked.post( - "http://example.com/update", - status=200, - body="OK", - content_type="text/plain", - ) - await wled.upgrade(version="0.15.0") - - -async def test_upgrade_ethernet_board() -> None: + await prepare_wled_for_upgrade(responses, wled) + responses.get( + "https://github.com/wled/WLED/releases/download/v0.15.0/WLED_0.15.0_ESP32.bin", + status=200, + body=b"fake firmware", + ) + responses.post( + "http://example.com/update", + status=200, + body="OK", + content_type="text/plain", + ) + await wled.upgrade(version="0.15.0") + + +async def test_upgrade_ethernet_board(responses: aioresponses, wled: WLED) -> None: """Test upgrade with Ethernet board (empty bssid).""" - with aioresponses() as mocked: - async with aiohttp.ClientSession() as session: - wled = await _get_wled_with_device_for_upgrade( - mocked, session, wifi_bssid="" - ) - mocked.get( - "https://github.com/wled/WLED/releases/download/v0.15.0/WLED_0.15.0_ESP32_Ethernet.bin", - status=200, - body=b"fake firmware", - ) - mocked.post( - "http://example.com/update", - status=200, - body="OK", - content_type="text/plain", - ) - await wled.upgrade(version="0.15.0") - - -async def test_upgrade_esp02_gzip() -> None: + await prepare_wled_for_upgrade(responses, wled, wifi_bssid="") + responses.get( + "https://github.com/wled/WLED/releases/download/v0.15.0/WLED_0.15.0_ESP32_Ethernet.bin", + status=200, + body=b"fake firmware", + ) + responses.post( + "http://example.com/update", + status=200, + body="OK", + content_type="text/plain", + ) + await wled.upgrade(version="0.15.0") + + +async def test_upgrade_esp02_gzip(responses: aioresponses, wled: WLED) -> None: """Test upgrade for esp02 (2M ESP8266) includes .gz suffix.""" - with aioresponses() as mocked: - async with aiohttp.ClientSession() as session: - wled_data = load_fixture_json("wled") - wled_data["info"]["arch"] = "esp8266" - wled_data["info"]["ver"] = "0.14.0" - # Small filesystem for esp02 detection - wled_data["info"]["fs"]["t"] = 512 - mocked.get( - "http://example.com/json", - status=200, - body=json.dumps(wled_data), - content_type="application/json", - ) - mocked.get( - "http://example.com/presets.json", - status=200, - body=json.dumps(load_fixture_json("presets")), - content_type="application/json", - ) - wled = WLED("example.com", session=session) - await wled.update() - mocked.get( - "https://github.com/wled/WLED/releases/download/v0.15.0/WLED_0.15.0_ESP02.bin.gz", - status=200, - body=b"fake firmware", - ) - mocked.post( - "http://example.com/update", - status=200, - body="OK", - content_type="text/plain", - ) - await wled.upgrade(version="0.15.0") - - -async def test_upgrade_404() -> None: + wled_data = load_fixture_json("wled") + wled_data["info"]["arch"] = "esp8266" + wled_data["info"]["ver"] = "0.14.0" + # Small filesystem for esp02 detection + wled_data["info"]["fs"]["t"] = 512 + mock_json_and_presets(responses, wled_data) + await wled.update() + responses.get( + "https://github.com/wled/WLED/releases/download/v0.15.0/WLED_0.15.0_ESP02.bin.gz", + status=200, + body=b"fake firmware", + ) + responses.post( + "http://example.com/update", + status=200, + body="OK", + content_type="text/plain", + ) + await wled.upgrade(version="0.15.0") + + +async def test_upgrade_404(responses: aioresponses, wled: WLED) -> None: """Test upgrade with 404 download raises WLEDUpgradeError.""" - with aioresponses() as mocked: - async with aiohttp.ClientSession() as session: - wled = await _get_wled_with_device_for_upgrade(mocked, session) - mocked.get( - "https://github.com/wled/WLED/releases/download/v0.99.0/WLED_0.99.0_ESP32.bin", - status=404, - ) - with pytest.raises(WLEDUpgradeError, match="does not exist"): - await wled.upgrade(version="0.99.0") - - -async def test_upgrade_other_http_error() -> None: + await prepare_wled_for_upgrade(responses, wled) + responses.get( + "https://github.com/wled/WLED/releases/download/v0.99.0/WLED_0.99.0_ESP32.bin", + status=404, + ) + with pytest.raises(WLEDUpgradeError, match="does not exist"): + await wled.upgrade(version="0.99.0") + + +async def test_upgrade_other_http_error(responses: aioresponses, wled: WLED) -> None: """Test upgrade with non-404 HTTP error raises WLEDUpgradeError.""" - with aioresponses() as mocked: - async with aiohttp.ClientSession() as session: - wled = await _get_wled_with_device_for_upgrade(mocked, session) - mocked.get( - "https://github.com/wled/WLED/releases/download/v0.15.0/WLED_0.15.0_ESP32.bin", - status=500, - ) - with pytest.raises(WLEDUpgradeError, match="Could not download"): - await wled.upgrade(version="0.15.0") - - -async def test_upgrade_connection_error() -> None: + await prepare_wled_for_upgrade(responses, wled) + responses.get( + "https://github.com/wled/WLED/releases/download/v0.15.0/WLED_0.15.0_ESP32.bin", + status=500, + ) + with pytest.raises(WLEDUpgradeError, match="Could not download"): + await wled.upgrade(version="0.15.0") + + +async def test_upgrade_connection_error(responses: aioresponses, wled: WLED) -> None: """Test upgrade with connection error raises WLEDConnectionError.""" - with aioresponses() as mocked: - async with aiohttp.ClientSession() as session: - wled = await _get_wled_with_device_for_upgrade(mocked, session) - mocked.get( - "https://github.com/wled/WLED/releases/download/v0.15.0/WLED_0.15.0_ESP32.bin", - exception=aiohttp.ClientError("fail"), - ) - with pytest.raises(WLEDConnectionError): - await wled.upgrade(version="0.15.0") - - -async def test_upgrade_timeout() -> None: + await prepare_wled_for_upgrade(responses, wled) + responses.get( + "https://github.com/wled/WLED/releases/download/v0.15.0/WLED_0.15.0_ESP32.bin", + exception=aiohttp.ClientError("fail"), + ) + with pytest.raises(WLEDConnectionError): + await wled.upgrade(version="0.15.0") + + +async def test_upgrade_timeout(responses: aioresponses, wled: WLED) -> None: """Test upgrade with timeout raises WLEDConnectionTimeoutError.""" - with aioresponses() as mocked: - async with aiohttp.ClientSession() as session: - wled = await _get_wled_with_device_for_upgrade(mocked, session) - wled.request_timeout = 0.001 - mocked.get( - "https://github.com/wled/WLED/releases/download/v0.15.0/WLED_0.15.0_ESP32.bin", - exception=TimeoutError(), - ) - with pytest.raises(WLEDConnectionTimeoutError): - await wled.upgrade(version="0.15.0") + await prepare_wled_for_upgrade(responses, wled) + wled.request_timeout = 0.001 + responses.get( + "https://github.com/wled/WLED/releases/download/v0.15.0/WLED_0.15.0_ESP32.bin", + exception=TimeoutError(), + ) + with pytest.raises(WLEDConnectionTimeoutError): + await wled.upgrade(version="0.15.0") # ========================================================================= @@ -1448,7 +1620,9 @@ async def test_upgrade_timeout() -> None: # ========================================================================= -async def test_releases_success() -> None: +async def test_releases_success( + responses: aioresponses, session: aiohttp.ClientSession +) -> None: """Test successful release fetching.""" releases_data = [ { @@ -1468,27 +1642,28 @@ async def test_releases_success() -> None: "prerelease": True, }, ] - with aioresponses() as mocked: - mocked.get( - "https://api.github.com/repos/wled/WLED/releases", - status=200, - body=json.dumps(releases_data), - content_type="application/json", - ) - async with aiohttp.ClientSession() as session: - wled_releases = WLEDReleases(session=session) - releases = await wled_releases.releases() - assert isinstance(releases, Releases) - assert releases.stable is not None - assert str(releases.stable) == "0.15.0" - assert releases.beta is not None - assert str(releases.beta) == "0.15.0b1" - assert releases.nightly is not None - assert str(releases.nightly) == "17.0.0-dev20260416" - assert releases.repo == "wled/WLED" - - -async def test_releases_custom_repo() -> None: + responses.get( + "https://api.github.com/repos/wled/WLED/releases", + status=200, + body=json.dumps(releases_data), + content_type="application/json", + ) + wled_releases = WLEDReleases(session=session) + releases = await wled_releases.releases() + + assert isinstance(releases, Releases) + assert releases.stable is not None + assert str(releases.stable) == "0.15.0" + assert releases.beta is not None + assert str(releases.beta) == "0.15.0b1" + assert releases.nightly is not None + assert str(releases.nightly) == "17.0.0-dev20260416" + assert releases.repo == "wled/WLED" + + +async def test_releases_custom_repo( + responses: aioresponses, session: aiohttp.ClientSession +) -> None: """Test fetching releases from a custom repository.""" releases_data = [ { @@ -1496,21 +1671,21 @@ async def test_releases_custom_repo() -> None: "prerelease": False, }, ] - with aioresponses() as mocked: - mocked.get( - "https://api.github.com/repos/MoonModules/WLED/releases", - status=200, - body=json.dumps(releases_data), - content_type="application/json", - ) - async with aiohttp.ClientSession() as session: - wled_releases = WLEDReleases(repo="MoonModules/WLED", session=session) - releases = await wled_releases.releases() - assert releases.repo == "MoonModules/WLED" - assert str(releases.stable) == "0.14.0" + responses.get( + "https://api.github.com/repos/MoonModules/WLED/releases", + status=200, + body=json.dumps(releases_data), + content_type="application/json", + ) + wled_releases = WLEDReleases(repo="MoonModules/WLED", session=session) + releases = await wled_releases.releases() + assert releases.repo == "MoonModules/WLED" + assert str(releases.stable) == "0.14.0" -async def test_releases_with_b_in_tag_name() -> None: +async def test_releases_with_b_in_tag_name( + responses: aioresponses, session: aiohttp.ClientSession +) -> None: """Test releases with 'b' in tag name are treated as beta.""" releases_data = [ { @@ -1522,23 +1697,23 @@ async def test_releases_with_b_in_tag_name() -> None: "prerelease": False, }, ] - with aioresponses() as mocked: - mocked.get( - "https://api.github.com/repos/wled/WLED/releases", - status=200, - body=json.dumps(releases_data), - content_type="application/json", - ) - async with aiohttp.ClientSession() as session: - wled_releases = WLEDReleases(session=session) - releases = await wled_releases.releases() - assert releases.beta is not None - assert str(releases.beta) == "0.14.1b2" - assert releases.stable is not None - assert str(releases.stable) == "0.14.0" + responses.get( + "https://api.github.com/repos/wled/WLED/releases", + status=200, + body=json.dumps(releases_data), + content_type="application/json", + ) + wled_releases = WLEDReleases(session=session) + releases = await wled_releases.releases() + assert releases.beta is not None + assert str(releases.beta) == "0.14.1b2" + assert releases.stable is not None + assert str(releases.stable) == "0.14.0" -async def test_releases_no_beta() -> None: +async def test_releases_no_beta( + responses: aioresponses, session: aiohttp.ClientSession +) -> None: """Test releases when no beta is available.""" releases_data = [ { @@ -1546,148 +1721,142 @@ async def test_releases_no_beta() -> None: "prerelease": False, }, ] - with aioresponses() as mocked: - mocked.get( - "https://api.github.com/repos/wled/WLED/releases", - status=200, - body=json.dumps(releases_data), - content_type="application/json", - ) - async with aiohttp.ClientSession() as session: - wled_releases = WLEDReleases(session=session) - releases = await wled_releases.releases() - assert releases.stable is not None - assert releases.beta is None + responses.get( + "https://api.github.com/repos/wled/WLED/releases", + status=200, + body=json.dumps(releases_data), + content_type="application/json", + ) + wled_releases = WLEDReleases(session=session) + releases = await wled_releases.releases() + assert releases.stable is not None + assert releases.beta is None -async def test_releases_context_manager() -> None: +async def test_releases_context_manager(responses: aioresponses) -> None: """Test WLEDReleases as context manager.""" releases_data = [ {"tag_name": "v0.14.0", "prerelease": False}, ] - with aioresponses() as mocked: - mocked.get( - "https://api.github.com/repos/wled/WLED/releases", - status=200, - body=json.dumps(releases_data), - content_type="application/json", - ) - async with WLEDReleases() as wled_releases: - releases = await wled_releases.releases() - assert releases.stable is not None + responses.get( + "https://api.github.com/repos/wled/WLED/releases", + status=200, + body=json.dumps(releases_data), + content_type="application/json", + ) + async with WLEDReleases() as wled_releases: + releases = await wled_releases.releases() + assert releases.stable is not None -async def test_releases_internal_session() -> None: +async def test_releases_internal_session(responses: aioresponses) -> None: """Test WLEDReleases creates internal session.""" releases_data = [ {"tag_name": "v0.14.0", "prerelease": False}, ] - with aioresponses() as mocked: - mocked.get( - "https://api.github.com/repos/wled/WLED/releases", - status=200, - body=json.dumps(releases_data), - content_type="application/json", - ) - wled_releases = WLEDReleases() - assert wled_releases.session is None - await wled_releases.releases() - assert wled_releases.session is not None - assert wled_releases._close_session is True # pylint: disable=protected-access - await wled_releases.close() + responses.get( + "https://api.github.com/repos/wled/WLED/releases", + status=200, + body=json.dumps(releases_data), + content_type="application/json", + ) + wled_releases = WLEDReleases() + assert wled_releases.session is None + await wled_releases.releases() + assert wled_releases.session is not None + assert wled_releases._close_session is True # pylint: disable=protected-access + await wled_releases.close() -async def test_releases_close_external_session() -> None: +async def test_releases_close_external_session( + session: aiohttp.ClientSession, +) -> None: """Test close() does not close externally provided session.""" - async with aiohttp.ClientSession() as session: - wled_releases = WLEDReleases(session=session) - await wled_releases.close() - assert not session.closed + wled_releases = WLEDReleases(session=session) + await wled_releases.close() + assert not session.closed -async def test_releases_http_error() -> None: +async def test_releases_http_error( + responses: aioresponses, session: aiohttp.ClientSession +) -> None: """Test releases raises WLEDError on HTTP error.""" - with aioresponses() as mocked: - mocked.get( - "https://api.github.com/repos/wled/WLED/releases", - status=500, - body='{"message": "error"}', - content_type="application/json", - ) - async with aiohttp.ClientSession() as session: - wled_releases = WLEDReleases(session=session) - with pytest.raises(WLEDError): - await wled_releases.releases() + responses.get( + "https://api.github.com/repos/wled/WLED/releases", + status=500, + body='{"message": "error"}', + content_type="application/json", + ) + wled_releases = WLEDReleases(session=session) + with pytest.raises(WLEDError): + await wled_releases.releases() -async def test_releases_http_error_text() -> None: +async def test_releases_http_error_text( + responses: aioresponses, session: aiohttp.ClientSession +) -> None: """Test releases raises WLEDError on HTTP error with text.""" - with aioresponses() as mocked: - mocked.get( - "https://api.github.com/repos/wled/WLED/releases", - status=403, - body="Forbidden", - content_type="text/plain", - ) - async with aiohttp.ClientSession() as session: - wled_releases = WLEDReleases(session=session) - with pytest.raises(WLEDError): - await wled_releases.releases() + responses.get( + "https://api.github.com/repos/wled/WLED/releases", + status=403, + body="Forbidden", + content_type="text/plain", + ) + wled_releases = WLEDReleases(session=session) + with pytest.raises(WLEDError): + await wled_releases.releases() -async def test_releases_non_json_response() -> None: +async def test_releases_non_json_response( + responses: aioresponses, session: aiohttp.ClientSession +) -> None: """Test releases raises WLEDError on non-JSON response.""" - with aioresponses() as mocked: - mocked.get( - "https://api.github.com/repos/wled/WLED/releases", - status=200, - body="Not JSON", - content_type="text/plain", - ) - async with aiohttp.ClientSession() as session: - wled_releases = WLEDReleases(session=session) - with pytest.raises(WLEDError, match="No JSON"): - await wled_releases.releases() + responses.get( + "https://api.github.com/repos/wled/WLED/releases", + status=200, + body="Not JSON", + content_type="text/plain", + ) + wled_releases = WLEDReleases(session=session) + with pytest.raises(WLEDError, match="No JSON"): + await wled_releases.releases() -async def test_releases_timeout() -> None: +async def test_releases_timeout( + responses: aioresponses, session: aiohttp.ClientSession +) -> None: """Test releases raises on timeout.""" - with aioresponses() as mocked: - mocked.get( - "https://api.github.com/repos/wled/WLED/releases", - exception=TimeoutError(), - ) - mocked.get( - "https://api.github.com/repos/wled/WLED/releases", - exception=TimeoutError(), - ) - mocked.get( - "https://api.github.com/repos/wled/WLED/releases", - exception=TimeoutError(), - ) - async with aiohttp.ClientSession() as session: - wled_releases = WLEDReleases(session=session, request_timeout=0.1) + responses.get( + "https://api.github.com/repos/wled/WLED/releases", exception=TimeoutError() + ) + responses.get( + "https://api.github.com/repos/wled/WLED/releases", exception=TimeoutError() + ) + responses.get( + "https://api.github.com/repos/wled/WLED/releases", exception=TimeoutError() + ) + wled_releases = WLEDReleases(session=session, request_timeout=0.1) - with pytest.raises(WLEDConnectionError): - await wled_releases.releases() + with pytest.raises(WLEDConnectionError): + await wled_releases.releases() -async def test_releases_connection_error() -> None: +async def test_releases_connection_error( + responses: aioresponses, session: aiohttp.ClientSession +) -> None: """Test releases raises on connection error.""" - with aioresponses() as mocked: - mocked.get( - "https://api.github.com/repos/wled/WLED/releases", - exception=aiohttp.ClientError("fail"), - ) - mocked.get( - "https://api.github.com/repos/wled/WLED/releases", - exception=aiohttp.ClientError("fail"), - ) - mocked.get( - "https://api.github.com/repos/wled/WLED/releases", - exception=aiohttp.ClientError("fail"), - ) - async with aiohttp.ClientSession() as session: - wled_releases = WLEDReleases(session=session) - with pytest.raises(WLEDConnectionError): - await wled_releases.releases() + responses.get( + "https://api.github.com/repos/wled/WLED/releases", + exception=aiohttp.ClientError("fail"), + ) + responses.get( + "https://api.github.com/repos/wled/WLED/releases", + exception=aiohttp.ClientError("fail"), + ) + responses.get( + "https://api.github.com/repos/wled/WLED/releases", + exception=aiohttp.ClientError("fail"), + ) + wled_releases = WLEDReleases(session=session) + with pytest.raises(WLEDConnectionError): + await wled_releases.releases()