ledgrab/server/tests/test_hue_segment.py

"""Tests for Hue gradient-lightstrip segment (channel) mapping.

The DTLS handshake and the actual gradient rendering need a real bridge +
gradient strip to validate; here we lock down the parts that DON'T: channel
discovery/ordering from an entertainment_configuration payload, the v2 frame
builder's channel-id keying (vs the legacy per-light index), the resample on
send, and the ``hue_gradient_mode`` config round-trip through the store.
"""

from __future__ import annotations

import struct

import pytest

from ledgrab.core.devices.hue_client import (
    HEADER_SIZE,
    PROTOCOL_NAME,
    HueClient,
    _build_entertainment_frame,
    parse_entertainment_channels,
)
from ledgrab.storage.device_store import Device


class _FakeSock:
    def __init__(self) -> None:
        self.sent: list[bytes] = []

    def sendto(self, data: bytes, addr) -> None:
        self.sent.append(bytes(data))


class TestParseChannels:
    def test_orders_by_position_x_then_y(self):
        cfg = {
            "data": [
                {
                    "channels": [
                        {"channel_id": 2, "position": {"x": 0.5, "y": 0.0}},
                        {"channel_id": 0, "position": {"x": -0.5, "y": 0.0}},
                        {"channel_id": 1, "position": {"x": 0.0, "y": 0.0}},
                    ]
                }
            ]
        }
        assert parse_entertainment_channels(cfg) == [0, 1, 2]

    def test_gradient_strip_contributes_multiple_channels(self):
        # A 5-segment gradient strip → five channels, all from one light.
        channels = [{"channel_id": i, "position": {"x": i * 0.1, "y": 0.0}} for i in range(5)]
        cfg = {"data": [{"channels": channels}]}
        assert parse_entertainment_channels(cfg) == [0, 1, 2, 3, 4]

    def test_empty_payload_is_empty(self):
        assert parse_entertainment_channels({}) == []
        assert parse_entertainment_channels({"data": []}) == []


class TestFrameBuilder:
    def test_header_is_huestream_v2(self):
        frame = _build_entertainment_frame([(255, 0, 0)], sequence=7)
        assert frame[0:9] == PROTOCOL_NAME
        assert frame[9] == 2 and frame[10] == 0  # version 2.0
        assert frame[11] == 7  # sequence

    def test_record_is_seven_bytes_keyed_by_channel(self):
        colors = [(255, 0, 0), (0, 255, 0)]
        frame = _build_entertainment_frame(colors, channel_ids=[3, 1])
        assert len(frame) == HEADER_SIZE + 7 * 2
        cid0, r0, g0, b0 = struct.unpack_from(">BHHH", frame, HEADER_SIZE)
        assert (cid0, r0, g0, b0) == (3, 255 * 257, 0, 0)
        cid1, r1, g1, b1 = struct.unpack_from(">BHHH", frame, HEADER_SIZE + 7)
        assert (cid1, r1, g1, b1) == (1, 0, 255 * 257, 0)

    def test_falls_back_to_index_without_channel_map(self):
        frame = _build_entertainment_frame([(1, 2, 3), (4, 5, 6)])
        cid0 = struct.unpack_from(">B", frame, HEADER_SIZE)[0]
        cid1 = struct.unpack_from(">B", frame, HEADER_SIZE + 7)[0]
        assert (cid0, cid1) == (0, 1)


class TestSendAndCount:
    def _client(self, *, channel_ids, led_count, gradient_mode=True) -> HueClient:
        c = HueClient("hue://1.2.3.4", led_count=led_count, gradient_mode=gradient_mode)
        c._connected = True
        c._sock = _FakeSock()  # type: ignore[assignment]
        c._dtls_sock = None
        c._channel_ids = channel_ids
        return c

    def test_resamples_strip_across_channels(self):
        c = self._client(channel_ids=[0, 1, 2, 3, 4], led_count=1)
        c.send_pixels_fast([(10, 0, 0)] * 20)
        frame = c._sock.sent[0]  # type: ignore[attr-defined]
        # 5 channels → 5 records.
        assert len(frame) == HEADER_SIZE + 7 * 5

    def test_legacy_path_uses_led_count(self):
        c = self._client(channel_ids=[], led_count=3)
        c.send_pixels_fast([(5, 5, 5)] * 8)
        frame = c._sock.sent[0]  # type: ignore[attr-defined]
        assert len(frame) == HEADER_SIZE + 7 * 3

    def test_device_led_count_reflects_channels(self):
        assert self._client(channel_ids=[0, 1, 2, 3, 4], led_count=1).device_led_count == 5
        assert self._client(channel_ids=[], led_count=3).device_led_count == 3
        # Gradient mode off → channel count ignored even if present.
        off = self._client(channel_ids=[0, 1], led_count=2, gradient_mode=False)
        assert off.device_led_count == 2


class TestConfigRoundTrip:
    def _device(self, gradient_mode: bool) -> Device:
        return Device(
            device_id="d1",
            name="Gradient strip",
            url="hue://1.2.3.4",
            led_count=5,
            device_type="hue",
            hue_gradient_mode=gradient_mode,
        )

    def test_default_on_is_omitted(self):
        d = self._device(True)
        assert "hue_gradient_mode" not in d.to_dict()
        back = Device.from_dict(d.to_dict())
        assert back.hue_gradient_mode is True
        assert back.to_config().hue_gradient_mode is True

    def test_opt_out_round_trips(self):
        d = self._device(False)
        assert d.to_dict().get("hue_gradient_mode") is False
        back = Device.from_dict(d.to_dict())
        assert back.hue_gradient_mode is False
        assert back.to_config().hue_gradient_mode is False


if __name__ == "__main__":
    pytest.main([__file__, "-v"])