LumaOps-Public/lumaops/backend/tests/test_protocol.py

from __future__ import annotations

import struct

import pytest

from lumaops_backend.connectors.base import RGBColor
from lumaops_backend.connectors.openrgb.protocol import (
    HEADER,
    MAGIC,
    Mode,
    ModeFlag,
    PacketId,
    ProtocolError,
    pack_color,
    pack_header,
    pack_mode,
    pack_string,
    pack_update_leds,
    parse_controller,
    parse_header,
)


def controller_packet() -> bytes:
    color = RGBColor(red=16, green=32, blue=48)
    mode = (
        pack_string("Static")
        + struct.pack("<i", 7)
        + struct.pack("<I", int(ModeFlag.HAS_BRIGHTNESS | ModeFlag.HAS_PER_LED_COLOR))
        + struct.pack("<IIIIIIIIII", 0, 0, 0, 255, 1, 1, 0, 128, 0, 1)
        + struct.pack("<H", 1)
        + pack_color(color)
    )
    zone = (
        pack_string("Main")
        + struct.pack("<iIIIH", 1, 2, 2, 2, 0)
        + struct.pack("<H", 1)
        + pack_string("Segment A")
        + struct.pack("<iII", 1, 0, 2)
        + struct.pack("<I", 0)
    )
    leds = pack_string("LED 1") + struct.pack("<I", 1) + pack_string("LED 2") + struct.pack("<I", 2)
    body = (
        struct.pack("<i", 0)
        + pack_string("Test Controller")
        + pack_string("LumaOps")
        + pack_string("Fixture")
        + pack_string("1.2.3")
        + pack_string("SERIAL-1")
        + pack_string("usb:1-2")
        + struct.pack("<H", 1)
        + struct.pack("<i", 0)
        + mode
        + struct.pack("<H", 1)
        + zone
        + struct.pack("<H", 2)
        + leds
        + struct.pack("<H", 2)
        + pack_color(color)
        + pack_color(color)
        + struct.pack("<H", 2)
        + pack_string("A")
        + pack_string("B")
        + struct.pack("<I", 1)
    )
    return struct.pack("<I", len(body) + 4) + body


def test_header_roundtrip_and_limit() -> None:
    raw = pack_header(3, PacketId.UPDATE_LEDS, 42)
    assert len(raw) == HEADER.size
    parsed = parse_header(raw)
    assert parsed.device_index == 3
    assert parsed.packet_id == PacketId.UPDATE_LEDS
    assert parsed.payload_size == 42

    oversized = HEADER.pack(MAGIC, 0, 0, 4097)
    with pytest.raises(ProtocolError):
        parse_header(oversized, max_packet_size=4096)


def test_parse_protocol_v5_controller() -> None:
    controller = parse_controller(controller_packet(), 4)
    assert controller.name == "Test Controller"
    assert controller.vendor == "LumaOps"
    assert controller.firmware_version == "1.2.3"
    assert len(controller.leds) == 2
    assert controller.led_alt_names == ["A", "B"]
    assert controller.zones[0].segments[0].name == "Segment A"
    assert controller.capabilities.brightness is True
    assert controller.capabilities.per_led is True


def test_parser_rejects_truncation_and_bad_declared_size() -> None:
    packet = controller_packet()
    with pytest.raises(ProtocolError):
        parse_controller(packet[:-1], 0)
    corrupted = struct.pack("<I", len(packet) + 100) + packet[4:]
    with pytest.raises(ProtocolError):
        parse_controller(corrupted, 0)


def test_led_updates_require_exact_count() -> None:
    color = RGBColor(red=1, green=2, blue=3)
    payload = pack_update_leds([color, color], 2)
    assert struct.unpack("<I", payload[:4])[0] == len(payload)
    with pytest.raises(ProtocolError):
        pack_update_leds([color], 2)


def test_mode_serializer_validates_ranges() -> None:
    mode = Mode(
        index=0,
        name="Pulse",
        value=1,
        flags=int(ModeFlag.HAS_SPEED | ModeFlag.HAS_BRIGHTNESS),
        speed_min=1,
        speed_max=10,
        brightness_min=0,
        brightness_max=255,
        colors_min=0,
        colors_max=0,
        speed=5,
        brightness=128,
        direction=0,
        color_mode=0,
    )
    payload = pack_mode(mode, 0, brightness_percent=50, speed=7)
    assert struct.unpack("<I", payload[:4])[0] == len(payload)
    with pytest.raises(ProtocolError):
        pack_mode(mode, 0, speed=11)


def test_mode_serializer_carries_effect_colors_and_direction() -> None:
    mode = Mode(
        index=4,
        name="Color Wave",
        value=4,
        flags=int(
            ModeFlag.HAS_SPEED
            | ModeFlag.HAS_DIRECTION_LR
            | ModeFlag.HAS_BRIGHTNESS
            | ModeFlag.HAS_MODE_SPECIFIC_COLOR
        ),
        speed_min=0,
        speed_max=2,
        brightness_min=0,
        brightness_max=255,
        colors_min=2,
        colors_max=2,
        speed=1,
        brightness=128,
        direction=0,
        color_mode=1,
    )
    colors = [RGBColor(red=12, green=34, blue=56), RGBColor(red=78, green=90, blue=123)]
    payload = pack_mode(
        mode,
        mode.index,
        brightness_percent=75,
        speed=2,
        direction=1,
        colors=colors,
    )
    assert struct.unpack("<I", payload[:4])[0] == len(payload)
    assert payload.endswith(pack_color(colors[0]) + pack_color(colors[1]))
    with pytest.raises(ProtocolError):
        pack_mode(mode, mode.index, colors=colors[:1])