diff --git a/tests/test_transport.py b/tests/test_transport.py new file mode 100644 index 0000000..0040953 --- /dev/null +++ b/tests/test_transport.py @@ -0,0 +1,216 @@ +""" +Tests for ESP32 UDP transport client. +Protocol: docs/protocol.md — v1, port 4210, fire-and-forget JSON datagrams. +""" +import asyncio +import json +import pytest + + +class MockTransport: + """Minimal mock for asyncio.DatagramTransport.""" + + def __init__(self): + self.sent = [] + self.closed = False + + def sendto(self, data: bytes, addr=None): + self.sent.append(json.loads(data.decode("utf-8"))) + + def close(self): + self.closed = True + + +# --- send_command() tests --- + +def test_send_command_injects_v1(): + """send_command() must always inject 'v':1 per protocol spec.""" + from led_sync.transport.udp_client import ESP32Transport + mock = MockTransport() + proto = ESP32Transport() + proto.connection_made(mock) + proto.send_command({"zone": "wand", "animation": "chase", "params": {"speed": 0.5}}) + assert mock.sent[-1]["v"] == 1 + + +def test_send_command_preserves_fields(): + """send_command() must pass through all caller-supplied fields.""" + from led_sync.transport.udp_client import ESP32Transport + mock = MockTransport() + proto = ESP32Transport() + proto.connection_made(mock) + proto.send_command({"zone": "schrank", "animation": "breathe", "params": {"intensity": 0.8}}) + payload = mock.sent[-1] + assert payload["zone"] == "schrank" + assert payload["animation"] == "breathe" + assert payload["params"]["intensity"] == 0.8 + + +def test_send_command_stop(): + """Stop command must include 'v':1, cmd, zone.""" + from led_sync.transport.udp_client import ESP32Transport + mock = MockTransport() + proto = ESP32Transport() + proto.connection_made(mock) + proto.send_command({"cmd": "stop", "zone": "all"}) + payload = mock.sent[-1] + assert payload["v"] == 1 + assert payload["cmd"] == "stop" + assert payload["zone"] == "all" + + +def test_send_command_brightness(): + """Brightness command must include 'v':1, cmd, zone, value.""" + from led_sync.transport.udp_client import ESP32Transport + mock = MockTransport() + proto = ESP32Transport() + proto.connection_made(mock) + proto.send_command({"cmd": "brightness", "zone": "all", "value": 40}) + payload = mock.sent[-1] + assert payload["v"] == 1 + assert payload["cmd"] == "brightness" + assert payload["value"] == 40 + + +def test_send_command_silent_drop_before_connection(): + """send_command() must silently drop packets when transport is None (pre-connection).""" + from led_sync.transport.udp_client import ESP32Transport + proto = ESP32Transport() + # Must not raise — no transport set + proto.send_command({"zone": "all", "cmd": "stop"}) + + +def test_send_command_drops_oversized_payload(): + """Payloads exceeding 512 bytes must be silently dropped.""" + from led_sync.transport.udp_client import ESP32Transport + mock = MockTransport() + proto = ESP32Transport() + proto.connection_made(mock) + # Build a command that produces >512 bytes + big_colors = [[255, 0, 0, 0]] * 100 + proto.send_command({"zone": "wand", "animation": "color_wash", "params": {"colors": big_colors}}) + # No packet sent (or zero packets sent after this drop) + assert len(mock.sent) == 0 + + +# --- STATUS response handling --- + +def test_datagram_received_status_ok_sets_connected(): + """STATUS ok response must set connected=True and store last_status.""" + from led_sync.transport.udp_client import ESP32Transport + proto = ESP32Transport() + proto.connection_made(MockTransport()) + assert not proto.connected + status_msg = {"v": 1, "status": "ok", "wand": "chase", "schrank": "breathe", "brightness": 40} + proto.datagram_received(json.dumps(status_msg).encode(), ("192.168.1.100", 4210)) + assert proto.connected + assert proto.last_status is not None + assert proto.last_status["wand"] == "chase" + assert proto.last_status["schrank"] == "breathe" + assert proto.last_status["brightness"] == 40 + + +def test_datagram_received_status_ok_invokes_callback(): + """STATUS ok response must invoke on_status callback if provided.""" + from led_sync.transport.udp_client import ESP32Transport + received = [] + proto = ESP32Transport(on_status=received.append) + proto.connection_made(MockTransport()) + status_msg = {"v": 1, "status": "ok", "wand": "rainbow", "schrank": "pulse", "brightness": 20} + proto.datagram_received(json.dumps(status_msg).encode(), ("192.168.1.100", 4210)) + assert len(received) == 1 + assert received[0]["wand"] == "rainbow" + + +def test_datagram_received_non_ok_status_does_not_set_connected(): + """Only 'status':'ok' should set connected=True.""" + from led_sync.transport.udp_client import ESP32Transport + proto = ESP32Transport() + proto.connection_made(MockTransport()) + # Send a message that has a 'status' key but not 'ok' + msg = {"v": 1, "status": "error", "reason": "unknown"} + proto.datagram_received(json.dumps(msg).encode(), ("192.168.1.100", 4210)) + assert not proto.connected + + +def test_datagram_received_malformed_json_silent(): + """Malformed JSON in datagram_received() must not raise.""" + from led_sync.transport.udp_client import ESP32Transport + proto = ESP32Transport() + proto.connection_made(MockTransport()) + proto.datagram_received(b"not json at all {{{", ("192.168.1.100", 4210)) + # Must not raise + + +def test_datagram_received_empty_bytes_silent(): + """Empty datagram must not raise.""" + from led_sync.transport.udp_client import ESP32Transport + proto = ESP32Transport() + proto.connection_made(MockTransport()) + proto.datagram_received(b"", ("192.168.1.100", 4210)) + + +# --- send_status_ping --- + +def test_send_status_ping(): + """send_status_ping() must send {'v':1,'cmd':'status'}.""" + from led_sync.transport.udp_client import ESP32Transport + mock = MockTransport() + proto = ESP32Transport() + proto.connection_made(mock) + proto.send_status_ping() + assert len(mock.sent) == 1 + assert mock.sent[0]["v"] == 1 + assert mock.sent[0]["cmd"] == "status" + + +# --- connection_lost --- + +def test_connection_lost_sets_disconnected(): + """connection_lost() must set connected=False and clear transport.""" + from led_sync.transport.udp_client import ESP32Transport + mock = MockTransport() + proto = ESP32Transport() + proto.connection_made(mock) + # Simulate a connected state + status_msg = {"v": 1, "status": "ok", "wand": "chase", "schrank": "breathe", "brightness": 40} + proto.datagram_received(json.dumps(status_msg).encode(), ("192.168.1.100", 4210)) + assert proto.connected + proto.connection_lost(None) + assert not proto.connected + + +# --- create_esp32_transport factory --- + +@pytest.mark.asyncio +async def test_create_esp32_transport_returns_instance(): + """create_esp32_transport() must return an ESP32Transport instance.""" + from led_sync.transport.udp_client import ESP32Transport, create_esp32_transport + # We can't actually send UDP in unit tests, but we can test the coroutine is awaitable + # by patching the loop endpoint creation + import unittest.mock as mock + + fake_protocol = ESP32Transport() + fake_mock_transport = MockTransport() + fake_protocol.connection_made(fake_mock_transport) + + with mock.patch("asyncio.get_running_loop") as mock_loop: + loop_instance = mock.AsyncMock() + loop_instance.create_datagram_endpoint = mock.AsyncMock( + return_value=(fake_mock_transport, fake_protocol) + ) + mock_loop.return_value = loop_instance + + result = await create_esp32_transport("192.168.1.100", port=4210) + assert isinstance(result, ESP32Transport) + # Factory sends an initial status ping + assert any(p.get("cmd") == "status" for p in fake_mock_transport.sent) + + +# --- __init__.py re-export --- + +def test_package_re_exports(): + """led_sync.transport must re-export ESP32Transport and create_esp32_transport.""" + from led_sync.transport import ESP32Transport, create_esp32_transport + assert ESP32Transport is not None + assert create_esp32_transport is not None