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).
421 lines
13 KiB
Python
421 lines
13 KiB
Python
"""Tests for the Yeelight LAN LED client + provider."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import json
|
|
from unittest.mock import AsyncMock, MagicMock
|
|
|
|
import numpy as np
|
|
import pytest
|
|
|
|
from ledgrab.core.devices.device_config import YeelightConfig
|
|
from ledgrab.core.devices.led_client import ProviderDeps
|
|
from ledgrab.core.devices.yeelight_client import (
|
|
YeelightClient,
|
|
_average_color,
|
|
_pack_rgb,
|
|
_parse_ssdp_response,
|
|
parse_yeelight_url,
|
|
)
|
|
from ledgrab.core.devices.yeelight_provider import YeelightDeviceProvider
|
|
|
|
# ============================================================================
|
|
# parse_yeelight_url
|
|
# ============================================================================
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"url,expected",
|
|
[
|
|
("yeelight://192.168.1.50", "192.168.1.50"),
|
|
("yeelight://192.168.1.50:55443", "192.168.1.50"),
|
|
("192.168.1.50", "192.168.1.50"),
|
|
("192.168.1.50:55443", "192.168.1.50"),
|
|
("bulb.local", "bulb.local"),
|
|
],
|
|
)
|
|
def test_parse_yeelight_url(url, expected):
|
|
assert parse_yeelight_url(url) == expected
|
|
|
|
|
|
@pytest.mark.parametrize("url", ["", " ", "yeelight://", "://192.168.1.1"])
|
|
def test_parse_yeelight_url_rejects_empty(url):
|
|
with pytest.raises(ValueError):
|
|
parse_yeelight_url(url)
|
|
|
|
|
|
# ============================================================================
|
|
# Helpers
|
|
# ============================================================================
|
|
|
|
|
|
def test_pack_rgb_packs_24_bit_int():
|
|
assert _pack_rgb(255, 0, 0) == 0xFF0000
|
|
assert _pack_rgb(0, 255, 0) == 0x00FF00
|
|
assert _pack_rgb(0, 0, 255) == 0x0000FF
|
|
assert _pack_rgb(0x12, 0x34, 0x56) == 0x123456
|
|
# Clamps to a byte
|
|
assert _pack_rgb(300, -5, 256) & 0xFFFFFF == _pack_rgb(300 & 0xFF, -5 & 0xFF, 256 & 0xFF)
|
|
|
|
|
|
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():
|
|
assert _average_color([(10, 0, 0), (20, 0, 0), (30, 0, 0)]) == (20, 0, 0)
|
|
|
|
|
|
def test_average_color_empty():
|
|
assert _average_color([]) == (0, 0, 0)
|
|
assert _average_color(np.array([], dtype=np.uint8)) == (0, 0, 0)
|
|
|
|
|
|
# ============================================================================
|
|
# YeelightClient (mocked transport)
|
|
# ============================================================================
|
|
|
|
|
|
def _make_connected_client(min_interval_s: float = 0.0) -> YeelightClient:
|
|
client = YeelightClient("yeelight://127.0.0.1", led_count=10, min_interval_s=min_interval_s)
|
|
writer = MagicMock()
|
|
writer.write = MagicMock()
|
|
writer.drain = AsyncMock()
|
|
writer.close = MagicMock()
|
|
writer.wait_closed = AsyncMock()
|
|
client._writer = writer
|
|
client._reader = MagicMock()
|
|
client._connected = True
|
|
return client
|
|
|
|
|
|
def _sent_payloads(client: YeelightClient) -> list[dict]:
|
|
"""Decode every JSON-RPC body the client has written."""
|
|
payloads = []
|
|
for call in client._writer.write.call_args_list:
|
|
raw = call.args[0].decode("utf-8").strip()
|
|
payloads.append(json.loads(raw))
|
|
return payloads
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_pixels_averages_and_packs_rgb():
|
|
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]["method"] == "set_rgb"
|
|
# Average of (255,0,0), (0,255,0), (0,0,255) is (85, 85, 85) → 0x555555
|
|
assert payloads[0]["params"][0] == _pack_rgb(85, 85, 85)
|
|
# Effect & duration: ambilight needs sudden + 0 ms
|
|
assert payloads[0]["params"][1] == "sudden"
|
|
assert payloads[0]["params"][2] == 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)
|
|
# Scaled: 200*0.501..→100, 100→50, 50→25
|
|
expected = _pack_rgb(int(200 * 128 / 255), int(100 * 128 / 255), int(50 * 128 / 255))
|
|
assert payloads[0]["params"][0] == expected
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_pixels_full_brightness_passthrough():
|
|
client = _make_connected_client()
|
|
pixels = np.array([[200, 100, 50]], dtype=np.uint8)
|
|
|
|
await client.send_pixels(pixels, brightness=255)
|
|
|
|
payloads = _sent_payloads(client)
|
|
assert payloads[0]["params"][0] == _pack_rgb(200, 100, 50)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_pixels_rate_limit_drops_subsequent_calls():
|
|
"""Within the min interval, the second call no-ops without TX."""
|
|
client = _make_connected_client(min_interval_s=10.0) # huge gate
|
|
pixels = np.array([[10, 20, 30]], dtype=np.uint8)
|
|
|
|
await client.send_pixels(pixels)
|
|
await client.send_pixels(pixels)
|
|
await client.send_pixels(pixels)
|
|
|
|
payloads = _sent_payloads(client)
|
|
assert len(payloads) == 1 # only the first one made it through
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_pixels_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)
|
|
|
|
payloads = _sent_payloads(client)
|
|
assert len(payloads) == 3
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_pixels_when_not_connected_raises():
|
|
client = YeelightClient("yeelight://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))
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_set_power_sends_set_power_command():
|
|
client = _make_connected_client()
|
|
await client.set_power(True)
|
|
await client.set_power(False)
|
|
payloads = _sent_payloads(client)
|
|
assert [p["method"] for p in payloads] == ["set_power", "set_power"]
|
|
assert payloads[0]["params"][0] == "on"
|
|
assert payloads[1]["params"][0] == "off"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_set_brightness_clamps_to_1_100():
|
|
client = _make_connected_client()
|
|
await client.set_brightness(0)
|
|
await client.set_brightness(50)
|
|
await client.set_brightness(200)
|
|
payloads = _sent_payloads(client)
|
|
# Yeelight bulbs reject brightness 0 (use set_power off instead) — clamp to 1.
|
|
assert payloads[0]["params"][0] == 1
|
|
assert payloads[1]["params"][0] == 50
|
|
assert payloads[2]["params"][0] == 100
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_close_releases_transport():
|
|
client = _make_connected_client()
|
|
writer = client._writer
|
|
await client.close()
|
|
writer.close.assert_called_once()
|
|
assert client._writer is None
|
|
assert client.is_connected is False
|
|
|
|
|
|
# ============================================================================
|
|
# SSDP response parsing
|
|
# ============================================================================
|
|
|
|
|
|
_SAMPLE_RESPONSE = (
|
|
b"HTTP/1.1 200 OK\r\n"
|
|
b"Cache-Control: max-age=3600\r\n"
|
|
b"Date: \r\n"
|
|
b"Ext: \r\n"
|
|
b"Location: yeelight://192.168.1.50:55443\r\n"
|
|
b"Server: POSIX UPnP/1.0 YGLC/1\r\n"
|
|
b"id: 0x000000000035cb01\r\n"
|
|
b"model: color\r\n"
|
|
b"fw_ver: 18\r\n"
|
|
b"support: get_prop set_default set_power set_bright\r\n"
|
|
b"power: off\r\n"
|
|
b"bright: 100\r\n"
|
|
b"color_mode: 2\r\n"
|
|
b"ct: 4000\r\n"
|
|
b"rgb: 16711680\r\n"
|
|
)
|
|
|
|
|
|
def test_parse_ssdp_response_extracts_headers():
|
|
headers = _parse_ssdp_response(_SAMPLE_RESPONSE)
|
|
assert headers is not None
|
|
assert headers["location"] == "yeelight://192.168.1.50:55443"
|
|
assert headers["id"] == "0x000000000035cb01"
|
|
assert headers["model"] == "color"
|
|
assert headers["fw_ver"] == "18"
|
|
|
|
|
|
def test_parse_ssdp_response_rejects_non_yeelight():
|
|
"""A stray HTTP response from another SSDP service should be ignored."""
|
|
other = b"HTTP/1.1 200 OK\r\nLocation: upnp://something/else\r\n"
|
|
assert _parse_ssdp_response(other) is None
|
|
|
|
|
|
# ============================================================================
|
|
# Provider
|
|
# ============================================================================
|
|
|
|
|
|
def test_provider_device_type_and_capabilities():
|
|
provider = YeelightDeviceProvider()
|
|
assert provider.device_type == "yeelight"
|
|
caps = provider.capabilities
|
|
assert "manual_led_count" in caps
|
|
assert "power_control" in caps
|
|
assert "brightness_control" in caps
|
|
assert "static_color" in caps
|
|
assert "single_pixel" in caps
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_provider_validate_device_accepts_bare_host():
|
|
provider = YeelightDeviceProvider()
|
|
assert await provider.validate_device("192.168.1.50") == {}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_provider_validate_device_rejects_empty():
|
|
provider = YeelightDeviceProvider()
|
|
with pytest.raises(ValueError, match="Invalid Yeelight URL"):
|
|
await provider.validate_device("")
|
|
|
|
|
|
def test_provider_create_client_threads_config():
|
|
provider = YeelightDeviceProvider()
|
|
config = YeelightConfig(
|
|
device_id="device_test",
|
|
device_url="yeelight://192.168.1.50",
|
|
led_count=30,
|
|
yeelight_min_interval_ms=750,
|
|
)
|
|
|
|
client = provider.create_client(config, deps=ProviderDeps())
|
|
|
|
assert isinstance(client, YeelightClient)
|
|
assert client.host == "192.168.1.50"
|
|
assert client._led_count == 30
|
|
assert client._min_interval_s == pytest.approx(0.75)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_provider_discover_returns_empty_on_failure(monkeypatch):
|
|
"""Multicast failures (no network, firewall) must yield [], not raise."""
|
|
|
|
async def _explode(timeout):
|
|
raise OSError("no route to host")
|
|
|
|
monkeypatch.setattr(
|
|
"ledgrab.core.devices.yeelight_provider.discover_yeelight_bulbs",
|
|
_explode,
|
|
)
|
|
provider = YeelightDeviceProvider()
|
|
assert await provider.discover() == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_provider_discover_maps_ssdp_to_discovered_device(monkeypatch):
|
|
async def _fake(timeout):
|
|
return [
|
|
{
|
|
"location": "yeelight://192.168.1.50:55443",
|
|
"id": "0x0035cb01",
|
|
"model": "color",
|
|
"fw_ver": "18",
|
|
},
|
|
# Missing location should be skipped, not crash.
|
|
{"id": "0xff", "model": "stripe"},
|
|
]
|
|
|
|
monkeypatch.setattr(
|
|
"ledgrab.core.devices.yeelight_provider.discover_yeelight_bulbs",
|
|
_fake,
|
|
)
|
|
provider = YeelightDeviceProvider()
|
|
results = await provider.discover()
|
|
|
|
assert len(results) == 1
|
|
[bulb] = results
|
|
assert bulb.device_type == "yeelight"
|
|
assert bulb.url == "yeelight://192.168.1.50"
|
|
assert bulb.ip == "192.168.1.50"
|
|
assert bulb.mac == "0x0035cb01"
|
|
assert bulb.version == "18"
|
|
assert "color" in bulb.name.lower()
|
|
|
|
|
|
# ============================================================================
|
|
# Device.to_config() round-trip
|
|
# ============================================================================
|
|
|
|
|
|
def test_device_to_config_round_trip_yeelight():
|
|
from ledgrab.storage.device_store import Device
|
|
|
|
device = Device(
|
|
device_id="device_abc12345",
|
|
name="Bedroom bulb",
|
|
url="yeelight://192.168.1.42",
|
|
led_count=30,
|
|
device_type="yeelight",
|
|
yeelight_min_interval_ms=750,
|
|
)
|
|
|
|
config = device.to_config()
|
|
|
|
assert isinstance(config, YeelightConfig)
|
|
assert config.device_url == "yeelight://192.168.1.42"
|
|
assert config.led_count == 30
|
|
assert config.yeelight_min_interval_ms == 750
|
|
|
|
|
|
def test_device_to_dict_omits_yeelight_default_interval():
|
|
from ledgrab.storage.device_store import Device
|
|
|
|
device = Device(
|
|
device_id="device_abc12345",
|
|
name="Default",
|
|
url="yeelight://192.168.1.42",
|
|
led_count=1,
|
|
device_type="yeelight",
|
|
)
|
|
assert "yeelight_min_interval_ms" not in device.to_dict()
|
|
|
|
|
|
def test_device_to_dict_preserves_non_default_yeelight_interval():
|
|
from ledgrab.storage.device_store import Device
|
|
|
|
device = Device(
|
|
device_id="device_abc12345",
|
|
name="Custom",
|
|
url="yeelight://192.168.1.42",
|
|
led_count=1,
|
|
device_type="yeelight",
|
|
yeelight_min_interval_ms=1000,
|
|
)
|
|
assert device.to_dict()["yeelight_min_interval_ms"] == 1000
|
|
|
|
|
|
def test_device_from_dict_yeelight_round_trip():
|
|
from ledgrab.storage.device_store import Device
|
|
|
|
restored = Device.from_dict(
|
|
{
|
|
"id": "device_abc12345",
|
|
"name": "Roundtrip",
|
|
"url": "yeelight://10.0.0.1",
|
|
"led_count": 1,
|
|
"device_type": "yeelight",
|
|
"yeelight_min_interval_ms": 250,
|
|
}
|
|
)
|
|
assert restored.yeelight_min_interval_ms == 250
|
|
|
|
|
|
# Suppress the asyncio CancelledError that asyncio raises on
|
|
# garbage-collected mock writers — they're swallowed in close() but
|
|
# pytest-asyncio still warns about them when MagicMock is used.
|
|
@pytest.fixture(autouse=True)
|
|
def _suppress_asyncio_warnings():
|
|
yield
|
|
asyncio.get_event_loop_policy()
|