test(02-02): add failing tests for ESP32Transport UDP client

- Tests for send_command() v:1 injection, field preservation - Tests for stop/brightness commands, pre-connection silent drop - Tests for 512-byte payload limit enforcement - Tests for STATUS ok response → connected=True + last_status - Tests for malformed JSON silent handling, on_status callback - Tests for send_status_ping, connection_lost, create_esp32_transport factory - Tests for led_sync.transport package re-exports
2026-04-03 14:18:00 +02:00
parent 4fe76b65ef
commit f10fcf016e
1 changed files with 216 additions and 0 deletions
--- a/tests/test_transport.py
+++ 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