diff --git a/tests/test_e2e_protocol.py b/tests/test_e2e_protocol.py new file mode 100644 index 0000000..9cd5445 --- /dev/null +++ b/tests/test_e2e_protocol.py @@ -0,0 +1,197 @@ +"""End-to-end protocol validation tests. + +Tests the full stack: animation render -> encode -> UDP send -> simulator receive -> validate. +Uses asyncio.run() pattern (no pytest-asyncio dependency). +""" +import asyncio +import struct + +from lightsync.protocol.drgb import ( + encode_drgb, + encode_drgbw, + encode_dnrgb_packets, + PROTOCOL_DRGB, + PROTOCOL_DRGBW, + PROTOCOL_DNRGB, + WLED_PORT, +) +from lightsync.protocol.animation_cmd import encode_animation_cmd, ANIM_CMD_MARKER +from lightsync.protocol.udp_sender import UDPSender +from lightsync.protocol.simulator import SimulatorProtocol, run_simulator +from lightsync.animations import create_animation + + +# --------------------------------------------------------------------------- +# Shared async helper +# --------------------------------------------------------------------------- + +async def _setup(port: int = 0) -> tuple[asyncio.DatagramTransport, SimulatorProtocol, UDPSender, int]: + """Start simulator + sender. Returns (transport, protocol, sender, actual_port).""" + loop = asyncio.get_running_loop() + transport, protocol = await loop.create_datagram_endpoint( + SimulatorProtocol, + local_addr=("127.0.0.1", port), + ) + actual_port = transport.get_extra_info("sockname")[1] + sender = UDPSender() + await sender.start() + return transport, protocol, sender, actual_port + + +async def _teardown(transport: asyncio.DatagramTransport, sender: UDPSender) -> None: + await sender.stop() + transport.close() + # Give loop a moment to flush + await asyncio.sleep(0.01) + + +# --------------------------------------------------------------------------- +# Tests +# --------------------------------------------------------------------------- + +def test_e2e_drgb_10_pixels(): + """Encode DRGB 10 pixels, send, simulator receives DRGB with led_count=10.""" + asyncio.run(_test_e2e_drgb_10_pixels()) + + +async def _test_e2e_drgb_10_pixels(): + transport, protocol, sender, port = await _setup() + try: + pixels = [(255, 0, 0)] * 10 + packet = encode_drgb(pixels) + sender.send(packet, "127.0.0.1", port) + await asyncio.sleep(0.05) + + assert len(protocol.packets) == 1 + log = protocol.packets[0] + assert log.protocol == "DRGB" + assert log.led_count == 10 + finally: + await _teardown(transport, sender) + + +def test_e2e_drgbw_10_pixels(): + """Encode DRGBW 10 pixels, send, simulator receives DRGBW with led_count=10.""" + asyncio.run(_test_e2e_drgbw_10_pixels()) + + +async def _test_e2e_drgbw_10_pixels(): + transport, protocol, sender, port = await _setup() + try: + pixels = [(255, 0, 0, 128)] * 10 + packet = encode_drgbw(pixels) + sender.send(packet, "127.0.0.1", port) + await asyncio.sleep(0.05) + + assert len(protocol.packets) == 1 + log = protocol.packets[0] + assert log.protocol == "DRGBW" + assert log.led_count == 10 + finally: + await _teardown(transport, sender) + + +def test_e2e_dnrgb_600_pixels(): + """600-pixel frame splits into 2 DNRGB packets with correct start indices (0, 489).""" + asyncio.run(_test_e2e_dnrgb_600_pixels()) + + +async def _test_e2e_dnrgb_600_pixels(): + transport, protocol, sender, port = await _setup() + try: + pixels = [(100, 200, 50)] * 600 + packets = encode_dnrgb_packets(pixels) + assert len(packets) == 2, f"Expected 2 packets, got {len(packets)}" + + for pkt in packets: + sender.send(pkt, "127.0.0.1", port) + await asyncio.sleep(0.05) + + assert len(protocol.packets) == 2 + # Both should be DNRGB + assert all(p.protocol == "DNRGB" for p in protocol.packets) + # Sort by start_index to ensure order + logs = sorted(protocol.packets, key=lambda p: p.start_index) + assert logs[0].start_index == 0 + assert logs[1].start_index == 489 + finally: + await _teardown(transport, sender) + + +def test_e2e_animation_cmd_chase(): + """Encode animation_cmd 'chase', send, simulator decodes animation=='chase'.""" + asyncio.run(_test_e2e_animation_cmd_chase()) + + +async def _test_e2e_animation_cmd_chase(): + transport, protocol, sender, port = await _setup() + try: + params = {"speed": 0.7, "color": [0, 255, 100], "reverse": False} + packet = encode_animation_cmd("chase", params) + sender.send(packet, "127.0.0.1", port) + await asyncio.sleep(0.05) + + assert len(protocol.packets) == 1 + log = protocol.packets[0] + assert log.protocol == "ANIM_CMD" + assert log.decoded is not None + assert log.decoded["animation"] == "chase" + finally: + await _teardown(transport, sender) + + +def test_e2e_full_pipeline(): + """Full pipeline: solid_color render -> encode_drgb -> send -> simulator receives 20-LED DRGB frame.""" + asyncio.run(_test_e2e_full_pipeline()) + + +async def _test_e2e_full_pipeline(): + transport, protocol, sender, port = await _setup() + try: + anim = create_animation("solid_color", {"color": [255, 0, 0]}) + pixels = anim.render(0.0, 20) + assert len(pixels) == 20 + + # solid_color returns RGB tuples + packet = encode_drgb([(r, g, b) for r, g, b in pixels]) + sender.send(packet, "127.0.0.1", port) + await asyncio.sleep(0.05) + + assert len(protocol.packets) == 1 + log = protocol.packets[0] + assert log.protocol == "DRGB" + assert log.led_count == 20 + finally: + await _teardown(transport, sender) + + +def test_e2e_sk6812_rgbw_pipeline(): + """SK6812 pipeline: solid_color render with white -> encode_drgbw -> simulator receives DRGBW.""" + asyncio.run(_test_e2e_sk6812_rgbw_pipeline()) + + +async def _test_e2e_sk6812_rgbw_pipeline(): + transport, protocol, sender, port = await _setup() + try: + anim = create_animation("solid_color", {"color": [255, 0, 0], "white": 128}) + pixels = anim.render(0.0, 15) + assert len(pixels) == 15 + + # Build RGBW tuples with white channel + rgbw_pixels = [(r, g, b, 128) for r, g, b in pixels] + packet = encode_drgbw(rgbw_pixels) + sender.send(packet, "127.0.0.1", port) + await asyncio.sleep(0.05) + + assert len(protocol.packets) == 1 + log = protocol.packets[0] + assert log.protocol == "DRGBW" + assert log.led_count == 15 + finally: + await _teardown(transport, sender) + + +def test_simulator_standalone_import(): + """run_simulator is importable without error.""" + from lightsync.protocol.simulator import run_simulator # noqa: F401 + assert callable(run_simulator)