ledgrab/server/tests/test_govee.py

"""Tests for the Govee LAN LED client + provider."""

from __future__ import annotations

import json
from unittest.mock import MagicMock

import numpy as np
import pytest

from ledgrab.core.devices.device_config import GoveeConfig
from ledgrab.core.devices.govee_client import (
    GOVEE_CONTROL_PORT,
    GoveeClient,
    _average_color,
    _parse_scan_reply,
    parse_govee_url,
)
from ledgrab.core.devices.govee_provider import GoveeDeviceProvider
from ledgrab.core.devices.led_client import ProviderDeps


# ============================================================================
# URL parsing
# ============================================================================


@pytest.mark.parametrize(
    "url,expected",
    [
        ("govee://192.168.1.50", "192.168.1.50"),
        ("192.168.1.50", "192.168.1.50"),
        ("bulb.local", "bulb.local"),
        ("govee://office-light.lan", "office-light.lan"),
    ],
)
def test_parse_govee_url(url, expected):
    assert parse_govee_url(url) == expected


@pytest.mark.parametrize("url", ["", "   ", "govee://", "://192.168.1.1"])
def test_parse_govee_url_rejects_empty(url):
    with pytest.raises(ValueError):
        parse_govee_url(url)


# ============================================================================
# Scan reply parsing
# ============================================================================


def test_parse_scan_reply_extracts_data_dict():
    raw = json.dumps(
        {
            "msg": {
                "cmd": "scan",
                "data": {
                    "ip": "192.168.1.50",
                    "device": "AA:BB:CC:DD:EE:FF",
                    "sku": "H6076",
                    "wifiVersionSoft": "2.3.4",
                },
            }
        }
    ).encode("utf-8")
    parsed = _parse_scan_reply(raw)
    assert parsed is not None
    assert parsed["ip"] == "192.168.1.50"
    assert parsed["sku"] == "H6076"
    assert parsed["device"] == "AA:BB:CC:DD:EE:FF"


def test_parse_scan_reply_rejects_non_scan_cmd():
    raw = json.dumps({"msg": {"cmd": "devStatus", "data": {"onOff": 1}}}).encode("utf-8")
    assert _parse_scan_reply(raw) is None


def test_parse_scan_reply_rejects_malformed_json():
    assert _parse_scan_reply(b"not json") is None
    assert _parse_scan_reply(b"") is None
    assert _parse_scan_reply(b'{"msg":') is None


def test_parse_scan_reply_rejects_non_dict_payload():
    raw = json.dumps(["not", "a", "dict"]).encode("utf-8")
    assert _parse_scan_reply(raw) is None


# ============================================================================
# Helpers
# ============================================================================


def test_average_color_numpy():
    pixels = np.array([[10, 20, 30], [40, 50, 60], [70, 80, 90]], dtype=np.uint8)
    assert _average_color(pixels) == (40, 50, 60)


def test_average_color_list_and_empty():
    assert _average_color([(10, 0, 0), (20, 0, 0), (30, 0, 0)]) == (20, 0, 0)
    assert _average_color([]) == (0, 0, 0)


# ============================================================================
# Payload builders
# ============================================================================


def test_build_color_payload_sets_colorTemInKelvin_to_zero():
    """colorTemInKelvin=0 selects RGB mode; otherwise the bulb uses CCT."""
    payload = GoveeClient._build_color_payload(255, 128, 0)
    assert payload == {
        "msg": {
            "cmd": "colorwc",
            "data": {
                "color": {"r": 255, "g": 128, "b": 0},
                "colorTemInKelvin": 0,
            },
        }
    }


def test_build_color_payload_clamps_channel_overflow():
    payload = GoveeClient._build_color_payload(300, -5, 256)
    data = payload["msg"]["data"]["color"]
    assert data == {"r": 300 & 0xFF, "g": -5 & 0xFF, "b": 256 & 0xFF}


def test_build_brightness_payload_clamps_to_1_100():
    assert GoveeClient._build_brightness_payload(0)["msg"]["data"]["value"] == 1
    assert GoveeClient._build_brightness_payload(50)["msg"]["data"]["value"] == 50
    assert GoveeClient._build_brightness_payload(200)["msg"]["data"]["value"] == 100


def test_build_power_payload_uses_integer_1_or_0():
    """Govee expects value 1/0, NOT JSON true/false."""
    assert GoveeClient._build_power_payload(True)["msg"]["data"]["value"] == 1
    assert GoveeClient._build_power_payload(False)["msg"]["data"]["value"] == 0


# ============================================================================
# GoveeClient (mocked transport)
# ============================================================================


def _make_connected_client(min_interval_s: float = 0.0) -> GoveeClient:
    client = GoveeClient("govee://127.0.0.1", led_count=10, min_interval_s=min_interval_s)
    transport = MagicMock()
    transport.sendto = MagicMock()
    transport.close = MagicMock()
    client._transport = transport
    client._protocol = MagicMock()
    client._connected = True
    return client


def _sent_payloads(client: GoveeClient) -> list[dict]:
    return [
        json.loads(call.args[0].decode("utf-8")) for call in client._transport.sendto.call_args_list
    ]


def test_client_targets_control_port_4003():
    client = GoveeClient("govee://192.168.1.50", led_count=1)
    assert client.port == GOVEE_CONTROL_PORT


@pytest.mark.asyncio
async def test_send_pixels_averages_to_colorwc():
    client = _make_connected_client()
    pixels = np.array(
        [[255, 0, 0], [0, 255, 0], [0, 0, 255]],
        dtype=np.uint8,
    )

    await client.send_pixels(pixels)

    payloads = _sent_payloads(client)
    assert len(payloads) == 1
    assert payloads[0]["msg"]["cmd"] == "colorwc"
    # Average of (255,0,0), (0,255,0), (0,0,255) is (85, 85, 85)
    assert payloads[0]["msg"]["data"]["color"] == {"r": 85, "g": 85, "b": 85}
    assert payloads[0]["msg"]["data"]["colorTemInKelvin"] == 0


@pytest.mark.asyncio
async def test_send_pixels_scales_for_brightness():
    client = _make_connected_client()
    pixels = np.array([[200, 100, 50]], dtype=np.uint8)

    await client.send_pixels(pixels, brightness=128)

    payloads = _sent_payloads(client)
    color = payloads[0]["msg"]["data"]["color"]
    assert color["r"] == int(200 * 128 / 255)
    assert color["g"] == int(100 * 128 / 255)
    assert color["b"] == int(50 * 128 / 255)


@pytest.mark.asyncio
async def test_rate_limit_drops_subsequent_frames():
    client = _make_connected_client(min_interval_s=10.0)
    pixels = np.array([[10, 20, 30]], dtype=np.uint8)

    await client.send_pixels(pixels)
    await client.send_pixels(pixels)

    assert len(_sent_payloads(client)) == 1


@pytest.mark.asyncio
async def test_zero_interval_sends_every_frame():
    client = _make_connected_client(min_interval_s=0.0)
    pixels = np.array([[10, 20, 30]], dtype=np.uint8)

    await client.send_pixels(pixels)
    await client.send_pixels(pixels)
    await client.send_pixels(pixels)

    assert len(_sent_payloads(client)) == 3


@pytest.mark.asyncio
async def test_send_pixels_when_not_connected_raises():
    client = GoveeClient("govee://127.0.0.1", led_count=1)
    with pytest.raises(RuntimeError, match="not connected"):
        await client.send_pixels(np.array([[1, 2, 3]], dtype=np.uint8))


def test_send_pixels_fast_runs_synchronously():
    client = _make_connected_client(min_interval_s=0.0)
    pixels = np.array([[10, 20, 30]], dtype=np.uint8)

    client.send_pixels_fast(pixels)

    payloads = _sent_payloads(client)
    assert len(payloads) == 1
    assert payloads[0]["msg"]["cmd"] == "colorwc"


def test_supports_fast_send_is_true():
    assert GoveeClient("govee://127.0.0.1", led_count=1).supports_fast_send is True


@pytest.mark.asyncio
async def test_set_power_sends_turn_cmd():
    client = _make_connected_client()
    await client.set_power(True)
    await client.set_power(False)
    payloads = _sent_payloads(client)
    assert payloads[0] == {"msg": {"cmd": "turn", "data": {"value": 1}}}
    assert payloads[1] == {"msg": {"cmd": "turn", "data": {"value": 0}}}


@pytest.mark.asyncio
async def test_set_brightness_sends_brightness_cmd_clamped():
    client = _make_connected_client()
    await client.set_brightness(5)
    await client.set_brightness(50)
    await client.set_brightness(150)
    payloads = _sent_payloads(client)
    values = [p["msg"]["data"]["value"] for p in payloads]
    assert values == [5, 50, 100]


@pytest.mark.asyncio
async def test_set_color_sends_colorwc():
    client = _make_connected_client()
    await client.set_color(12, 34, 56)
    payloads = _sent_payloads(client)
    assert payloads[0]["msg"]["cmd"] == "colorwc"
    assert payloads[0]["msg"]["data"]["color"] == {"r": 12, "g": 34, "b": 56}


@pytest.mark.asyncio
async def test_close_releases_transport():
    client = _make_connected_client()
    transport = client._transport
    await client.close()
    transport.close.assert_called_once()
    assert client._transport is None
    assert client.is_connected is False


# ============================================================================
# Provider
# ============================================================================


def test_provider_device_type_and_capabilities():
    provider = GoveeDeviceProvider()
    assert provider.device_type == "govee"
    caps = provider.capabilities
    assert "manual_led_count" in caps
    assert "power_control" in caps
    assert "brightness_control" in caps
    assert "single_pixel" in caps


@pytest.mark.asyncio
async def test_provider_validate_accepts_bare_host():
    provider = GoveeDeviceProvider()
    assert await provider.validate_device("192.168.1.50") == {}


@pytest.mark.asyncio
async def test_provider_validate_rejects_empty():
    provider = GoveeDeviceProvider()
    with pytest.raises(ValueError, match="Invalid Govee URL"):
        await provider.validate_device("")


def test_provider_create_client_threads_config():
    provider = GoveeDeviceProvider()
    config = GoveeConfig(
        device_id="device_test",
        device_url="govee://192.168.1.50",
        led_count=30,
        govee_min_interval_ms=100,
    )

    client = provider.create_client(config, deps=ProviderDeps())

    assert isinstance(client, GoveeClient)
    assert client.host == "192.168.1.50"
    assert client.port == GOVEE_CONTROL_PORT
    assert client._led_count == 30
    assert client._min_interval_s == pytest.approx(0.1)


@pytest.mark.asyncio
async def test_provider_discover_returns_empty_on_failure(monkeypatch):
    async def _explode(timeout):
        raise OSError("multicast unreachable")

    monkeypatch.setattr("ledgrab.core.devices.govee_provider.discover_govee_devices", _explode)
    provider = GoveeDeviceProvider()
    assert await provider.discover() == []


@pytest.mark.asyncio
async def test_provider_discover_maps_replies_to_discovered_devices(monkeypatch):
    async def _fake(timeout):
        return [
            {
                "ip": "192.168.1.50",
                "device": "AA:BB:CC:DD:EE:FF",
                "sku": "H6076",
                "version": "2.3.4",
            },
            # Missing IP — should be skipped.
            {"ip": "", "device": "00:00", "sku": "H1234"},
        ]

    monkeypatch.setattr("ledgrab.core.devices.govee_provider.discover_govee_devices", _fake)
    provider = GoveeDeviceProvider()
    results = await provider.discover()

    assert len(results) == 1
    [bulb] = results
    assert bulb.device_type == "govee"
    assert bulb.url == "govee://192.168.1.50"
    assert bulb.ip == "192.168.1.50"
    assert bulb.mac == "AA:BB:CC:DD:EE:FF"
    assert bulb.version == "2.3.4"
    assert "h6076" in bulb.name.lower()


# ============================================================================
# Device.to_config() round-trip
# ============================================================================


def test_device_to_config_round_trip_govee():
    from ledgrab.storage.device_store import Device

    device = Device(
        device_id="device_abc12345",
        name="Govee Immersion",
        url="govee://192.168.1.42",
        led_count=30,
        device_type="govee",
        govee_min_interval_ms=100,
    )

    config = device.to_config()

    assert isinstance(config, GoveeConfig)
    assert config.device_url == "govee://192.168.1.42"
    assert config.led_count == 30
    assert config.govee_min_interval_ms == 100


def test_device_to_dict_omits_govee_default_interval():
    from ledgrab.storage.device_store import Device

    device = Device(
        device_id="device_abc12345",
        name="Default",
        url="govee://192.168.1.42",
        led_count=1,
        device_type="govee",
    )
    assert "govee_min_interval_ms" not in device.to_dict()


def test_device_to_dict_preserves_non_default_govee_interval():
    from ledgrab.storage.device_store import Device

    device = Device(
        device_id="device_abc12345",
        name="Custom",
        url="govee://192.168.1.42",
        led_count=1,
        device_type="govee",
        govee_min_interval_ms=200,
    )
    assert device.to_dict()["govee_min_interval_ms"] == 200


def test_device_from_dict_govee_round_trip():
    from ledgrab.storage.device_store import Device

    restored = Device.from_dict(
        {
            "id": "device_abc12345",
            "name": "Roundtrip",
            "url": "govee://10.0.0.1",
            "led_count": 1,
            "device_type": "govee",
            "govee_min_interval_ms": 150,
        }
    )
    assert restored.govee_min_interval_ms == 150