From f7487dd17dfe45d2d1e09ae00509ed62e2d39e7f Mon Sep 17 00:00:00 2001 From: mik-laj <12058428+mik-laj@users.noreply.github.com> Date: Thu, 30 Apr 2026 03:46:27 +0200 Subject: [PATCH 1/5] Strengthen WLED request payload tests --- tests/test_wled.py | 168 +++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 168 insertions(+) diff --git a/tests/test_wled.py b/tests/test_wled.py index 2226d04d..c4471ea7 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 @@ -28,6 +30,20 @@ # ========================================================================= +def assert_post_payload(mocked: aioresponses, path: str, expected: dict) -> None: + """Assert a POST request payload sent to WLED.""" + mocked.assert_called_with( + url=URL(path), + method="POST", + data=orjson.dumps(expected), + headers={ + "Accept": "application/json, text/plain, */*", + "Content-Type": "application/json", + }, + allow_redirects=True, + ) + + async def test_json_request() -> None: """Test JSON response is handled correctly.""" with aioresponses() as mocked: @@ -408,6 +424,9 @@ async def test_master_brightness() -> None: async with aiohttp.ClientSession() as session: wled = WLED("example.com", session=session) await wled.master(brightness=200) + assert_post_payload( + mocked, "http://example.com/json/state", {"bri": 200, "v": True} + ) async def test_master_on() -> None: @@ -422,6 +441,9 @@ async def test_master_on() -> None: async with aiohttp.ClientSession() as session: wled = WLED("example.com", session=session) await wled.master(on=True) + assert_post_payload( + mocked, "http://example.com/json/state", {"on": True, "v": True} + ) async def test_master_transition() -> None: @@ -436,6 +458,9 @@ async def test_master_transition() -> None: async with aiohttp.ClientSession() as session: wled = WLED("example.com", session=session) await wled.master(transition=10) + assert_post_payload( + mocked, "http://example.com/json/state", {"tt": 10, "v": True} + ) async def test_master_all_params() -> None: @@ -450,6 +475,11 @@ async def test_master_all_params() -> None: async with aiohttp.ClientSession() as session: wled = WLED("example.com", session=session) await wled.master(brightness=100, on=True, transition=5) + assert_post_payload( + mocked, + "http://example.com/json/state", + {"bri": 100, "on": True, "tt": 5, "v": True}, + ) # ========================================================================= @@ -479,6 +509,11 @@ async def test_segment_basic() -> None: content_type="application/json", ) await wled.segment(0, brightness=200, on=True) + assert_post_payload( + mocked, + "http://example.com/json/state", + {"seg": [{"bri": 200, "on": True, "id": 0}], "v": True}, + ) async def test_segment_effect_by_name() -> None: @@ -493,6 +528,11 @@ async def test_segment_effect_by_name() -> None: content_type="application/json", ) await wled.segment(0, effect="Blink") + assert_post_payload( + mocked, + "http://example.com/json/state", + {"seg": [{"fx": 1, "id": 0}], "v": True}, + ) async def test_segment_palette_by_name() -> None: @@ -507,6 +547,11 @@ async def test_segment_palette_by_name() -> None: content_type="application/json", ) await wled.segment(0, palette="Random Cycle") + assert_post_payload( + mocked, + "http://example.com/json/state", + {"seg": [{"pal": 1, "id": 0}], "v": True}, + ) async def test_segment_color_primary() -> None: @@ -521,6 +566,11 @@ async def test_segment_color_primary() -> None: content_type="application/json", ) await wled.segment(0, color_primary=(255, 0, 0)) + assert_post_payload( + mocked, + "http://example.com/json/state", + {"seg": [{"col": [[255, 0, 0]], "id": 0}], "v": True}, + ) async def test_segment_color_secondary_only() -> None: @@ -535,6 +585,11 @@ async def test_segment_color_secondary_only() -> None: content_type="application/json", ) await wled.segment(0, color_secondary=(0, 255, 0)) + assert_post_payload( + mocked, + "http://example.com/json/state", + {"seg": [{"col": [[255, 159, 0], [0, 255, 0]], "id": 0}], "v": True}, + ) async def test_segment_color_tertiary_only() -> None: @@ -549,6 +604,14 @@ async def test_segment_color_tertiary_only() -> None: content_type="application/json", ) await wled.segment(0, color_tertiary=(0, 0, 255)) + assert_post_payload( + mocked, + "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() -> None: @@ -568,6 +631,14 @@ async def test_segment_all_colors() -> None: color_secondary=(0, 255, 0), color_tertiary=(0, 0, 255), ) + assert_post_payload( + mocked, + "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() -> None: @@ -582,6 +653,11 @@ async def test_segment_with_transition() -> None: content_type="application/json", ) await wled.segment(0, brightness=100, transition=5) + assert_post_payload( + mocked, + "http://example.com/json/state", + {"seg": [{"bri": 100, "id": 0}], "tt": 5, "v": True}, + ) async def test_segment_calls_update_when_no_device() -> None: @@ -597,6 +673,11 @@ async def test_segment_calls_update_when_no_device() -> None: async with aiohttp.ClientSession() as session: wled = WLED("example.com", session=session) await wled.segment(0, on=True) + assert_post_payload( + mocked, + "http://example.com/json/state", + {"seg": [{"on": True, "id": 0}], "v": True}, + ) async def test_segment_no_device_raises() -> None: @@ -623,6 +704,14 @@ async def test_segment_individual() -> None: content_type="application/json", ) await wled.segment(0, individual=[(255, 0, 0), (0, 255, 0)]) + assert_post_payload( + mocked, + "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() -> None: @@ -653,6 +742,14 @@ async def test_segment_color_tertiary_no_secondary_in_state() -> None: content_type="application/json", ) await wled.segment(0, color_tertiary=(0, 0, 255)) + assert_post_payload( + mocked, + "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() -> None: @@ -684,6 +781,11 @@ async def test_segment_secondary_no_color_in_state() -> None: ) # color is None, so it should use (0,0,0) fallback await wled.segment(0, color_secondary=(0, 255, 0)) + assert_post_payload( + mocked, + "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() -> None: @@ -714,6 +816,14 @@ async def test_segment_tertiary_no_color_in_state() -> None: content_type="application/json", ) await wled.segment(0, color_tertiary=(0, 0, 255)) + assert_post_payload( + mocked, + "http://example.com/json/state", + { + "seg": [{"col": [[0, 0, 0], [0, 0, 0], [0, 0, 255]], "id": 0}], + "v": True, + }, + ) # ========================================================================= @@ -733,6 +843,9 @@ async def test_preset_by_id() -> None: content_type="application/json", ) await wled.preset(1) + assert_post_payload( + mocked, "http://example.com/json/state", {"ps": 1, "v": True} + ) async def test_preset_by_name() -> None: @@ -747,6 +860,9 @@ async def test_preset_by_name() -> None: content_type="application/json", ) await wled.preset("My Preset") + assert_post_payload( + mocked, "http://example.com/json/state", {"ps": 1, "v": True} + ) async def test_preset_by_object() -> None: @@ -763,6 +879,9 @@ async def test_preset_by_object() -> None: 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( + mocked, "http://example.com/json/state", {"ps": 1, "v": True} + ) async def test_preset_name_not_found() -> None: @@ -777,6 +896,11 @@ async def test_preset_name_not_found() -> None: content_type="application/json", ) await wled.preset("NonExistent") + assert_post_payload( + mocked, + "http://example.com/json/state", + {"ps": "NonExistent", "v": True}, + ) async def test_playlist_by_id() -> None: @@ -791,6 +915,9 @@ async def test_playlist_by_id() -> None: content_type="application/json", ) await wled.playlist(2) + assert_post_payload( + mocked, "http://example.com/json/state", {"ps": 2, "v": True} + ) async def test_playlist_by_name() -> None: @@ -805,6 +932,9 @@ async def test_playlist_by_name() -> None: content_type="application/json", ) await wled.playlist("My Playlist") + assert_post_payload( + mocked, "http://example.com/json/state", {"ps": 2, "v": True} + ) async def test_playlist_by_object() -> None: @@ -821,6 +951,9 @@ async def test_playlist_by_object() -> None: 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( + mocked, "http://example.com/json/state", {"ps": 2, "v": True} + ) async def test_playlist_name_not_found() -> None: @@ -835,6 +968,11 @@ async def test_playlist_name_not_found() -> None: content_type="application/json", ) await wled.playlist("NonExistent") + assert_post_payload( + mocked, + "http://example.com/json/state", + {"ps": "NonExistent", "v": True}, + ) async def test_transition() -> None: @@ -849,6 +987,11 @@ async def test_transition() -> None: async with aiohttp.ClientSession() as session: wled = WLED("example.com", session=session) await wled.transition(10) + assert_post_payload( + mocked, + "http://example.com/json/state", + {"transition": 10, "v": True}, + ) async def test_live() -> None: @@ -863,6 +1006,11 @@ async def test_live() -> None: async with aiohttp.ClientSession() as session: wled = WLED("example.com", session=session) await wled.live(LiveDataOverride.ON) + assert_post_payload( + mocked, + "http://example.com/json/state", + {"lor": LiveDataOverride.ON.value, "v": True}, + ) async def test_sync_send() -> None: @@ -877,6 +1025,11 @@ async def test_sync_send() -> None: async with aiohttp.ClientSession() as session: wled = WLED("example.com", session=session) await wled.sync(send=True) + assert_post_payload( + mocked, + "http://example.com/json/state", + {"udpn": {"send": True}, "v": True}, + ) async def test_sync_receive() -> None: @@ -891,6 +1044,11 @@ async def test_sync_receive() -> None: async with aiohttp.ClientSession() as session: wled = WLED("example.com", session=session) await wled.sync(receive=True) + assert_post_payload( + mocked, + "http://example.com/json/state", + {"udpn": {"recv": True}, "v": True}, + ) async def test_nightlight_on() -> None: @@ -905,6 +1063,11 @@ async def test_nightlight_on() -> None: async with aiohttp.ClientSession() as session: wled = WLED("example.com", session=session) await wled.nightlight(on=True) + assert_post_payload( + mocked, + "http://example.com/json/state", + {"nl": {"on": True}, "v": True}, + ) async def test_nightlight_all_params() -> None: @@ -919,6 +1082,11 @@ async def test_nightlight_all_params() -> None: async with aiohttp.ClientSession() as session: wled = WLED("example.com", session=session) await wled.nightlight(duration=30, fade=True, on=True, target_brightness=50) + assert_post_payload( + mocked, + "http://example.com/json/state", + {"nl": {"dur": 30, "fade": True, "on": True, "tbri": 50}, "v": True}, + ) # ========================================================================= From 8c568081ded1e8b3bbbca0a72a86a3af3f8c4c77 Mon Sep 17 00:00:00 2001 From: mik-laj <12058428+mik-laj@users.noreply.github.com> Date: Thu, 30 Apr 2026 04:26:10 +0200 Subject: [PATCH 2/5] fixup! Strengthen WLED request payload tests --- tests/test_wled.py | 17 +++++++---------- 1 file changed, 7 insertions(+), 10 deletions(-) diff --git a/tests/test_wled.py b/tests/test_wled.py index c4471ea7..3b2e22df 100644 --- a/tests/test_wled.py +++ b/tests/test_wled.py @@ -32,16 +32,13 @@ def assert_post_payload(mocked: aioresponses, path: str, expected: dict) -> None: """Assert a POST request payload sent to WLED.""" - mocked.assert_called_with( - url=URL(path), - method="POST", - data=orjson.dumps(expected), - headers={ - "Accept": "application/json, text/plain, */*", - "Content-Type": "application/json", - }, - allow_redirects=True, - ) + 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 async def test_json_request() -> None: From bbed26b18eda7004aad28d9aac786b2271ac5bdb Mon Sep 17 00:00:00 2001 From: mik-laj <12058428+mik-laj@users.noreply.github.com> Date: Thu, 30 Apr 2026 06:06:33 +0200 Subject: [PATCH 3/5] fixup! fixup! Strengthen WLED request payload tests --- tests/conftest.py | 29 +- tests/test_wled.py | 2461 +++++++++++++++++++++----------------------- 2 files changed, 1170 insertions(+), 1320 deletions(-) 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 3b2e22df..6a296920 100644 --- a/tests/test_wled.py +++ b/tests/test_wled.py @@ -41,138 +41,114 @@ def assert_post_payload(mocked: aioresponses, path: str, expected: dict) -> None assert orjson.loads(request_call.kwargs["data"]) == expected -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("/") -async def test_internal_session() -> None: + assert response == "OK" + + +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") + assert response == "OK" -async def test_backoff() -> None: + +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("/") -async def test_timeout() -> None: + assert response == "OK" + + +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("/") # ========================================================================= @@ -180,174 +156,168 @@ 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() + + +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() + + +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") + + # 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" + # 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" + # 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" + # 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() -> None: +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", - ) + 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 call: still no fs, presets refetched again + 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", + ) - 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 @@ -361,28 +331,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 @@ -392,16 +363,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()) # ========================================================================= @@ -409,74 +379,70 @@ 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) - assert_post_payload( - mocked, "http://example.com/json/state", {"bri": 200, "v": True} - ) + 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) - assert_post_payload( - mocked, "http://example.com/json/state", {"on": True, "v": 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) - assert_post_payload( - mocked, "http://example.com/json/state", {"tt": 10, "v": True} - ) + 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) - assert_post_payload( - mocked, - "http://example.com/json/state", - {"bri": 100, "on": True, "tt": 5, "v": True}, - ) + 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}, + ) # ========================================================================= @@ -484,343 +450,312 @@ 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) - assert_post_payload( - mocked, - "http://example.com/json/state", - {"seg": [{"bri": 200, "on": True, "id": 0}], "v": True}, - ) + 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() -> None: +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") - assert_post_payload( - mocked, - "http://example.com/json/state", - {"seg": [{"fx": 1, "id": 0}], "v": True}, - ) + 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() -> None: +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") - assert_post_payload( - mocked, - "http://example.com/json/state", - {"seg": [{"pal": 1, "id": 0}], "v": True}, - ) + 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() -> None: + +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)) - assert_post_payload( - mocked, - "http://example.com/json/state", - {"seg": [{"col": [[255, 0, 0]], "id": 0}], "v": True}, - ) + 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)) -async def test_segment_color_secondary_only() -> None: + 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)) - assert_post_payload( - mocked, - "http://example.com/json/state", - {"seg": [{"col": [[255, 159, 0], [0, 255, 0]], "id": 0}], "v": True}, - ) + 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)) -async def test_segment_color_tertiary_only() -> None: + 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)) - assert_post_payload( - mocked, - "http://example.com/json/state", - { - "seg": [{"col": [[255, 159, 0], [0, 0, 0], [0, 0, 255]], "id": 0}], - "v": True, - }, - ) + 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)) -async def test_segment_all_colors() -> None: + 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), - ) - assert_post_payload( - mocked, - "http://example.com/json/state", - { - "seg": [{"col": [[255, 0, 0], [0, 255, 0], [0, 0, 255]], "id": 0}], - "v": True, - }, - ) + 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), + ) -async def test_segment_with_transition() -> None: + 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) - assert_post_payload( - mocked, - "http://example.com/json/state", - {"seg": [{"bri": 100, "id": 0}], "tt": 5, "v": True}, - ) + 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() -> None: +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) - assert_post_payload( - mocked, - "http://example.com/json/state", - {"seg": [{"on": True, "id": 0}], "v": 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) + + assert_post_payload( + responses, + "http://example.com/json/state", + {"seg": [{"on": True, "id": 0}], "v": True}, + ) -async def test_segment_no_device_raises() -> None: +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)]) - assert_post_payload( - mocked, - "http://example.com/json/state", - { - "seg": [{"i": [[255, 0, 0], [0, 255, 0]], "id": 0}], - "v": True, - }, - ) + 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() -> None: +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)) - assert_post_payload( - mocked, - "http://example.com/json/state", - { - "seg": [{"col": [[255, 0, 0], [0, 0, 0], [0, 0, 255]], "id": 0}], - "v": True, - }, - ) + # 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() -> None: +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)) - assert_post_payload( - mocked, - "http://example.com/json/state", - {"seg": [{"col": [[0, 0, 0], [0, 255, 0]], "id": 0}], "v": True}, - ) + # 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() -> None: +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)) - assert_post_payload( - mocked, - "http://example.com/json/state", - { - "seg": [{"col": [[0, 0, 0], [0, 0, 0], [0, 0, 255]], "id": 0}], - "v": True, - }, - ) + # 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, + }, + ) # ========================================================================= @@ -828,262 +763,243 @@ async def test_segment_tertiary_no_color_in_state() -> None: # ========================================================================= -async def test_preset_by_id() -> None: +async def test_preset_by_id(responses: aioresponses, wled: WLED) -> 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) - assert_post_payload( - mocked, "http://example.com/json/state", {"ps": 1, "v": True} - ) + 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() -> None: +async def test_preset_by_name(responses: aioresponses, wled: WLED) -> 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") - assert_post_payload( - mocked, "http://example.com/json/state", {"ps": 1, "v": True} - ) + 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() -> None: + +async def test_preset_by_object(responses: aioresponses, wled: WLED) -> 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) - assert_post_payload( - mocked, "http://example.com/json/state", {"ps": 1, "v": True} - ) + 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() -> None: +async def test_preset_name_not_found(responses: aioresponses, wled: WLED) -> 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") - assert_post_payload( - mocked, - "http://example.com/json/state", - {"ps": "NonExistent", "v": True}, - ) + 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") -async def test_playlist_by_id() -> None: + 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.""" - 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) - assert_post_payload( - mocked, "http://example.com/json/state", {"ps": 2, "v": True} - ) + 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) -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") - assert_post_payload( - mocked, "http://example.com/json/state", {"ps": 2, "v": True} - ) + assert_post_payload( + responses, "http://example.com/json/state", {"ps": 2, "v": True} + ) -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) - assert_post_payload( - mocked, "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") -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") - assert_post_payload( - mocked, - "http://example.com/json/state", - {"ps": "NonExistent", "v": True}, - ) + assert_post_payload( + responses, "http://example.com/json/state", {"ps": "NonExistent", "v": True} + ) -async def test_transition() -> None: +async def test_transition(responses: aioresponses, wled: WLED) -> 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) - assert_post_payload( - mocked, - "http://example.com/json/state", - {"transition": 10, "v": True}, - ) + 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) - assert_post_payload( - mocked, - "http://example.com/json/state", - {"lor": LiveDataOverride.ON.value, "v": True}, - ) + 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) - assert_post_payload( - mocked, - "http://example.com/json/state", - {"udpn": {"send": True}, "v": 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) - assert_post_payload( - mocked, - "http://example.com/json/state", - {"udpn": {"recv": True}, "v": 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) - assert_post_payload( - mocked, - "http://example.com/json/state", - {"nl": {"on": True}, "v": 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) - assert_post_payload( - mocked, - "http://example.com/json/state", - {"nl": {"dur": 30, "fade": True, "on": True, "tbri": 50}, "v": True}, - ) + 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}, + ) # ========================================================================= @@ -1091,57 +1007,53 @@ 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: @@ -1183,60 +1095,54 @@ 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) + + await wled.update() + + # Device has ws=0, so connect should try to ws_connect + assert wled.session is not None + with patch.object(wled.session, "ws_connect", new_callable=AsyncMock) as mock_ws: + mock_client = MagicMock(closed=False) + mock_client.close = AsyncMock() + mock_ws.return_value = mock_client + await wled.connect() + mock_ws.assert_called_once() async def test_listen_not_connected() -> None: @@ -1325,42 +1231,30 @@ 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.""" 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}) + mock_json_and_presets(responses) + 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}) -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") # ========================================================================= @@ -1368,9 +1262,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", @@ -1381,90 +1275,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: @@ -1478,134 +1342,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") # ========================================================================= @@ -1613,7 +1449,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 = [ { @@ -1633,27 +1471,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 = [ { @@ -1661,21 +1500,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 = [ { @@ -1687,23 +1526,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 = [ { @@ -1711,148 +1550,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() From 5bd612b14ee04c83872bc8b7c8c60bf276252b16 Mon Sep 17 00:00:00 2001 From: mik-laj <12058428+mik-laj@users.noreply.github.com> Date: Thu, 30 Apr 2026 23:47:48 +0200 Subject: [PATCH 4/5] fixup! fixup! fixup! Strengthen WLED request payload tests --- tests/test_wled.py | 251 ++++++++++++++++++++++++++++++++++----------- 1 file changed, 192 insertions(+), 59 deletions(-) diff --git a/tests/test_wled.py b/tests/test_wled.py index 6a296920..ffc3625a 100644 --- a/tests/test_wled.py +++ b/tests/test_wled.py @@ -25,10 +25,6 @@ from .conftest import full_device_data, load_fixture_json, mock_json_and_presets -# ========================================================================= -# Section 1: Existing tests (preserved) -# ========================================================================= - def assert_post_payload(mocked: aioresponses, path: str, expected: dict) -> None: """Assert a POST request payload sent to WLED.""" @@ -41,6 +37,11 @@ def assert_post_payload(mocked: aioresponses, path: str, expected: dict) -> None assert orjson.loads(request_call.kwargs["data"]) == expected +# ========================================================================= +# Section 1: Existing tests (preserved) +# ========================================================================= + + async def test_json_request(responses: aioresponses, wled: WLED) -> None: """Test JSON response is handled correctly.""" responses.get( @@ -188,6 +189,7 @@ async def test_update_empty_json_response(responses: aioresponses, wled: WLED) - body="", content_type="text/plain", ) + with pytest.raises(WLEDEmptyResponseError): await wled.update() @@ -278,31 +280,9 @@ async def test_update_refetches_presets_when_info_incomplete( # Set pmt to 0 so version can't be determined wled_data["info"]["fs"]["pmt"] = 0 - 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", - ) + mock_json_and_presets(responses, wled_data) # Second call: still no fs, presets refetched again - 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", - ) + mock_json_and_presets(responses, wled_data) await wled.update() # Without fs/pmt, every update refetches presets @@ -391,7 +371,9 @@ async def test_master_brightness(responses: aioresponses, wled: WLED) -> None: await wled.master(brightness=200) assert_post_payload( - responses, "http://example.com/json/state", {"bri": 200, "v": True} + responses, + "http://example.com/json/state", + {"bri": 200, "v": True}, ) @@ -407,7 +389,9 @@ async def test_master_on(responses: aioresponses, wled: WLED) -> None: await wled.master(on=True) assert_post_payload( - responses, "http://example.com/json/state", {"on": True, "v": True} + responses, + "http://example.com/json/state", + {"on": True, "v": True}, ) @@ -479,7 +463,12 @@ async def test_segment_basic(responses: aioresponses, wled: WLED) -> None: assert_post_payload( responses, "http://example.com/json/state", - {"seg": [{"bri": 200, "on": True, "id": 0}], "v": True}, + { + "seg": [ + {"bri": 200, "on": True, "id": 0}, + ], + "v": True, + }, ) @@ -498,7 +487,12 @@ async def test_segment_effect_by_name(responses: aioresponses, wled: WLED) -> No assert_post_payload( responses, "http://example.com/json/state", - {"seg": [{"fx": 1, "id": 0}], "v": True}, + { + "seg": [ + {"fx": 1, "id": 0}, + ], + "v": True, + }, ) @@ -517,7 +511,12 @@ async def test_segment_palette_by_name(responses: aioresponses, wled: WLED) -> N assert_post_payload( responses, "http://example.com/json/state", - {"seg": [{"pal": 1, "id": 0}], "v": True}, + { + "seg": [ + {"pal": 1, "id": 0}, + ], + "v": True, + }, ) @@ -536,7 +535,12 @@ async def test_segment_color_primary(responses: aioresponses, wled: WLED) -> Non assert_post_payload( responses, "http://example.com/json/state", - {"seg": [{"col": [[255, 0, 0]], "id": 0}], "v": True}, + { + "seg": [ + {"col": [[255, 0, 0]], "id": 0}, + ], + "v": True, + }, ) @@ -557,7 +561,18 @@ async def test_segment_color_secondary_only( assert_post_payload( responses, "http://example.com/json/state", - {"seg": [{"col": [[255, 159, 0], [0, 255, 0]], "id": 0}], "v": True}, + { + "seg": [ + { + "col": [ + [255, 159, 0], + [0, 255, 0], + ], + "id": 0, + } + ], + "v": True, + }, ) @@ -577,7 +592,16 @@ async def test_segment_color_tertiary_only(responses: aioresponses, wled: WLED) responses, "http://example.com/json/state", { - "seg": [{"col": [[255, 159, 0], [0, 0, 0], [0, 0, 255]], "id": 0}], + "seg": [ + { + "col": [ + [255, 159, 0], + [0, 0, 0], + [0, 0, 255], + ], + "id": 0, + } + ], "v": True, }, ) @@ -604,7 +628,16 @@ async def test_segment_all_colors(responses: aioresponses, wled: WLED) -> None: responses, "http://example.com/json/state", { - "seg": [{"col": [[255, 0, 0], [0, 255, 0], [0, 0, 255]], "id": 0}], + "seg": [ + { + "col": [ + [255, 0, 0], + [0, 255, 0], + [0, 0, 255], + ], + "id": 0, + } + ], "v": True, }, ) @@ -625,7 +658,13 @@ async def test_segment_with_transition(responses: aioresponses, wled: WLED) -> N assert_post_payload( responses, "http://example.com/json/state", - {"seg": [{"bri": 100, "id": 0}], "tt": 5, "v": True}, + { + "seg": [ + {"bri": 100, "id": 0}, + ], + "tt": 5, + "v": True, + }, ) @@ -646,7 +685,12 @@ async def test_segment_calls_update_when_no_device( assert_post_payload( responses, "http://example.com/json/state", - {"seg": [{"on": True, "id": 0}], "v": True}, + { + "seg": [ + {"on": True, "id": 0}, + ], + "v": True, + }, ) @@ -675,7 +719,12 @@ async def test_segment_individual(responses: aioresponses, wled: WLED) -> None: assert_post_payload( responses, "http://example.com/json/state", - {"seg": [{"i": [[255, 0, 0], [0, 255, 0]], "id": 0}], "v": True}, + { + "seg": [ + {"i": [[255, 0, 0], [0, 255, 0]], "id": 0}, + ], + "v": True, + }, ) @@ -700,7 +749,9 @@ async def test_segment_color_tertiary_no_secondary_in_state( responses, "http://example.com/json/state", { - "seg": [{"col": [[255, 0, 0], [0, 0, 0], [0, 0, 255]], "id": 0}], + "seg": [ + {"col": [[255, 0, 0], [0, 0, 0], [0, 0, 255]], "id": 0}, + ], "v": True, }, ) @@ -727,7 +778,12 @@ async def test_segment_secondary_no_color_in_state( assert_post_payload( responses, "http://example.com/json/state", - {"seg": [{"col": [[0, 0, 0], [0, 255, 0]], "id": 0}], "v": True}, + { + "seg": [ + {"col": [[0, 0, 0], [0, 255, 0]], "id": 0}, + ], + "v": True, + }, ) @@ -752,7 +808,9 @@ async def test_segment_tertiary_no_color_in_state( responses, "http://example.com/json/state", { - "seg": [{"col": [[0, 0, 0], [0, 0, 0], [0, 0, 255]], "id": 0}], + "seg": [ + {"col": [[0, 0, 0], [0, 0, 0], [0, 0, 255]], "id": 0}, + ], "v": True, }, ) @@ -776,7 +834,12 @@ async def test_preset_by_id(responses: aioresponses, wled: WLED) -> None: await wled.preset(1) assert_post_payload( - responses, "http://example.com/json/state", {"ps": 1, "v": True} + responses, + "http://example.com/json/state", + { + "ps": 1, + "v": True, + }, ) @@ -793,7 +856,12 @@ async def test_preset_by_name(responses: aioresponses, wled: WLED) -> None: await wled.preset("My Preset") assert_post_payload( - responses, "http://example.com/json/state", {"ps": 1, "v": True} + responses, + "http://example.com/json/state", + { + "ps": 1, + "v": True, + }, ) @@ -811,7 +879,12 @@ async def test_preset_by_object(responses: aioresponses, wled: WLED) -> None: await wled.preset(preset_obj) assert_post_payload( - responses, "http://example.com/json/state", {"ps": 1, "v": True} + responses, + "http://example.com/json/state", + { + "ps": 1, + "v": True, + }, ) @@ -828,7 +901,12 @@ async def test_preset_name_not_found(responses: aioresponses, wled: WLED) -> Non await wled.preset("NonExistent") assert_post_payload( - responses, "http://example.com/json/state", {"ps": "NonExistent", "v": True} + responses, + "http://example.com/json/state", + { + "ps": "NonExistent", + "v": True, + }, ) @@ -845,7 +923,12 @@ async def test_playlist_by_id(responses: aioresponses, wled: WLED) -> None: await wled.playlist(2) assert_post_payload( - responses, "http://example.com/json/state", {"ps": 2, "v": True} + responses, + "http://example.com/json/state", + { + "ps": 2, + "v": True, + }, ) @@ -862,7 +945,12 @@ async def test_playlist_by_name(responses: aioresponses, wled: WLED) -> None: await wled.playlist("My Playlist") assert_post_payload( - responses, "http://example.com/json/state", {"ps": 2, "v": True} + responses, + "http://example.com/json/state", + { + "ps": 2, + "v": True, + }, ) @@ -881,7 +969,12 @@ async def test_playlist_by_object(responses: aioresponses, wled: WLED) -> None: await wled.playlist(playlist_obj) assert_post_payload( - responses, "http://example.com/json/state", {"ps": 2, "v": True} + responses, + "http://example.com/json/state", + { + "ps": 2, + "v": True, + }, ) @@ -898,7 +991,12 @@ async def test_playlist_name_not_found(responses: aioresponses, wled: WLED) -> N await wled.playlist("NonExistent") assert_post_payload( - responses, "http://example.com/json/state", {"ps": "NonExistent", "v": True} + responses, + "http://example.com/json/state", + { + "ps": "NonExistent", + "v": True, + }, ) @@ -914,7 +1012,9 @@ async def test_transition(responses: aioresponses, wled: WLED) -> None: await wled.transition(10) assert_post_payload( - responses, "http://example.com/json/state", {"transition": 10, "v": True} + responses, + "http://example.com/json/state", + {"transition": 10, "v": True}, ) @@ -932,7 +1032,10 @@ async def test_live(responses: aioresponses, wled: WLED) -> None: assert_post_payload( responses, "http://example.com/json/state", - {"lor": LiveDataOverride.ON.value, "v": True}, + { + "lor": LiveDataOverride.ON.value, + "v": True, + }, ) @@ -948,7 +1051,12 @@ async def test_sync_send(responses: aioresponses, wled: WLED) -> None: await wled.sync(send=True) assert_post_payload( - responses, "http://example.com/json/state", {"udpn": {"send": True}, "v": True} + responses, + "http://example.com/json/state", + { + "udpn": {"send": True}, + "v": True, + }, ) @@ -964,7 +1072,12 @@ async def test_sync_receive(responses: aioresponses, wled: WLED) -> None: await wled.sync(receive=True) assert_post_payload( - responses, "http://example.com/json/state", {"udpn": {"recv": True}, "v": True} + responses, + "http://example.com/json/state", + { + "udpn": {"recv": True}, + "v": True, + }, ) @@ -980,7 +1093,12 @@ async def test_nightlight_on(responses: aioresponses, wled: WLED) -> None: await wled.nightlight(on=True) assert_post_payload( - responses, "http://example.com/json/state", {"nl": {"on": True}, "v": True} + responses, + "http://example.com/json/state", + { + "nl": {"on": True}, + "v": True, + }, ) @@ -998,7 +1116,15 @@ async def test_nightlight_all_params(responses: aioresponses, wled: WLED) -> Non assert_post_payload( responses, "http://example.com/json/state", - {"nl": {"dur": 30, "fade": True, "on": True, "tbri": 50}, "v": True}, + { + "nl": { + "dur": 30, + "fade": True, + "on": True, + "tbri": 50, + }, + "v": True, + }, ) @@ -1015,6 +1141,7 @@ async def test_reset(responses: aioresponses, wled: WLED) -> None: body="OK", content_type="text/plain", ) + await wled.reset() @@ -1103,7 +1230,6 @@ async def test_connect_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"): @@ -1233,8 +1359,9 @@ async def test_disconnect_not_connected() -> None: async def test_post_state_adds_v_true(responses: aioresponses, wled: WLED) -> None: """Test POST to /json/state adds v=True to data.""" - state_response = json.dumps(load_fixture_json("wled")["state"]) mock_json_and_presets(responses) + + state_response = json.dumps(load_fixture_json("wled")["state"]) responses.post( "http://example.com/json/state", status=200, @@ -1244,6 +1371,12 @@ async def test_post_state_adds_v_true(responses: aioresponses, wled: WLED) -> No 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( responses: aioresponses, wled: WLED From eafcef7904a7c293ca7561b7d4515abb31532e84 Mon Sep 17 00:00:00 2001 From: mik-laj <12058428+mik-laj@users.noreply.github.com> Date: Fri, 1 May 2026 19:20:51 +0200 Subject: [PATCH 5/5] fixup! fixup! fixup! fixup! Strengthen WLED request payload tests Co-Authored-By: Claude Sonnet 4.6 --- tests/test_wled.py | 14 +++++++------- 1 file changed, 7 insertions(+), 7 deletions(-) diff --git a/tests/test_wled.py b/tests/test_wled.py index ffc3625a..13b65b74 100644 --- a/tests/test_wled.py +++ b/tests/test_wled.py @@ -1259,16 +1259,16 @@ async def test_connect_calls_update_when_no_device( """Test connect() calls update() if no device is loaded.""" mock_json_and_presets(responses) - await wled.update() + mock_client = MagicMock(closed=False) + mock_client.close = AsyncMock() - # Device has ws=0, so connect should try to ws_connect - assert wled.session is not None - with patch.object(wled.session, "ws_connect", new_callable=AsyncMock) as mock_ws: - 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() - mock_ws.assert_called_once() + assert wled._device is not None # pylint: disable=protected-access async def test_listen_not_connected() -> None: