alexei.dolgolyov
888f8fd16e
ruff --select UP007,UP045 --fix converted ~1760 sites across the backend: `Optional[T]` → `T | None`, `Union[X, Y]` → `X | Y`. The remaining module-level alias targets that ruff conservatively skips (BindableFloatInput, ColorList, DeviceConfig) were converted by hand earlier in the pass. black -formatted the result so the wider unions fit cleanly under the 100-char line budget. pyproject.toml now sets [tool.ruff.lint] extend-select = ["UP007", "UP045"] so future legacy imports fire CI on every push. The pre-commit ruff hook was bumped from v0.8.0 -> v0.15.12 to recognise UP045 (split off from UP007 in v0.13).
434 lines
13 KiB
Python
434 lines
13 KiB
Python
"""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
|