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).
372 lines
11 KiB
Python
372 lines
11 KiB
Python
"""Tests for the WiZ Connected UDP 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 WiZConfig
|
|
from ledgrab.core.devices.led_client import ProviderDeps
|
|
from ledgrab.core.devices.wiz_client import (
|
|
WIZ_PORT,
|
|
WiZClient,
|
|
_average_color,
|
|
_extract_mac,
|
|
parse_wiz_url,
|
|
)
|
|
from ledgrab.core.devices.wiz_provider import WiZDeviceProvider
|
|
|
|
# ============================================================================
|
|
# parse_wiz_url
|
|
# ============================================================================
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"url,expected",
|
|
[
|
|
("wiz://192.168.1.50", ("192.168.1.50", WIZ_PORT)),
|
|
("wiz://192.168.1.50:38899", ("192.168.1.50", 38899)),
|
|
("wiz://192.168.1.50:40000", ("192.168.1.50", 40000)),
|
|
("192.168.1.50", ("192.168.1.50", WIZ_PORT)),
|
|
("192.168.1.50:38899", ("192.168.1.50", 38899)),
|
|
("bulb.local", ("bulb.local", WIZ_PORT)),
|
|
],
|
|
)
|
|
def test_parse_wiz_url(url, expected):
|
|
assert parse_wiz_url(url) == expected
|
|
|
|
|
|
@pytest.mark.parametrize("url", ["", " ", "wiz://", "://192.168.1.1"])
|
|
def test_parse_wiz_url_rejects_empty(url):
|
|
with pytest.raises(ValueError):
|
|
parse_wiz_url(url)
|
|
|
|
|
|
# ============================================================================
|
|
# 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)
|
|
|
|
|
|
def test_extract_mac_from_registration_reply():
|
|
assert _extract_mac({"result": {"mac": "AABBCCDDEEFF"}}) == "aabbccddeeff"
|
|
assert _extract_mac({"result": {}}) == ""
|
|
assert _extract_mac({"error": "oops"}) == ""
|
|
assert _extract_mac({}) == ""
|
|
|
|
|
|
# ============================================================================
|
|
# WiZClient (mocked transport)
|
|
# ============================================================================
|
|
|
|
|
|
def _make_connected_client(min_interval_s: float = 0.0) -> WiZClient:
|
|
client = WiZClient("wiz://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: WiZClient) -> list[dict]:
|
|
return [
|
|
json.loads(call.args[0].decode("utf-8")) for call in client._transport.sendto.call_args_list
|
|
]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_pixels_averages_to_set_pilot_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"] == "setPilot"
|
|
# Average of (255,0,0), (0,255,0), (0,0,255) is (85, 85, 85)
|
|
assert payloads[0]["params"] == {"r": 85, "g": 85, "b": 85}
|
|
|
|
|
|
@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)
|
|
expected_r = int(200 * 128 / 255)
|
|
expected_g = int(100 * 128 / 255)
|
|
expected_b = int(50 * 128 / 255)
|
|
assert payloads[0]["params"] == {"r": expected_r, "g": expected_g, "b": expected_b}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_pixels_zero_brightness_blacks_out():
|
|
client = _make_connected_client()
|
|
pixels = np.array([[200, 100, 50]], dtype=np.uint8)
|
|
|
|
await client.send_pixels(pixels, brightness=0)
|
|
|
|
payloads = _sent_payloads(client)
|
|
assert payloads[0]["params"] == {"r": 0, "g": 0, "b": 0}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_rate_limit_drops_subsequent_frames_within_window():
|
|
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)
|
|
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 = WiZClient("wiz://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():
|
|
"""Hot path: synchronous fast-send must dispatch over the UDP transport."""
|
|
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 payloads[0]["params"] == {"r": 10, "g": 20, "b": 30}
|
|
|
|
|
|
def test_send_pixels_fast_when_not_connected_raises():
|
|
client = WiZClient("wiz://127.0.0.1", led_count=1)
|
|
with pytest.raises(RuntimeError, match="not connected"):
|
|
client.send_pixels_fast([(1, 2, 3)])
|
|
|
|
|
|
def test_supports_fast_send_is_true():
|
|
assert WiZClient("wiz://127.0.0.1", led_count=1).supports_fast_send is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_set_power_sends_state_param():
|
|
client = _make_connected_client()
|
|
await client.set_power(True)
|
|
await client.set_power(False)
|
|
payloads = _sent_payloads(client)
|
|
assert payloads[0]["params"] == {"state": True}
|
|
assert payloads[1]["params"] == {"state": False}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_set_brightness_clamps_to_10_100():
|
|
"""WiZ rejects dimming values below ~10."""
|
|
client = _make_connected_client()
|
|
await client.set_brightness(5)
|
|
await client.set_brightness(50)
|
|
await client.set_brightness(200)
|
|
payloads = _sent_payloads(client)
|
|
assert payloads[0]["params"] == {"dimming": 10}
|
|
assert payloads[1]["params"] == {"dimming": 50}
|
|
assert payloads[2]["params"] == {"dimming": 100}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_set_color_sends_rgb_set_pilot():
|
|
client = _make_connected_client()
|
|
await client.set_color(12, 34, 56)
|
|
payloads = _sent_payloads(client)
|
|
assert payloads[0] == {"method": "setPilot", "params": {"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 = WiZDeviceProvider()
|
|
assert provider.device_type == "wiz"
|
|
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 = WiZDeviceProvider()
|
|
assert await provider.validate_device("192.168.1.50") == {}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_provider_validate_rejects_empty():
|
|
provider = WiZDeviceProvider()
|
|
with pytest.raises(ValueError, match="Invalid WiZ URL"):
|
|
await provider.validate_device("")
|
|
|
|
|
|
def test_provider_create_client_threads_config():
|
|
provider = WiZDeviceProvider()
|
|
config = WiZConfig(
|
|
device_id="device_test",
|
|
device_url="wiz://192.168.1.50",
|
|
led_count=30,
|
|
wiz_min_interval_ms=100,
|
|
)
|
|
|
|
client = provider.create_client(config, deps=ProviderDeps())
|
|
|
|
assert isinstance(client, WiZClient)
|
|
assert client.host == "192.168.1.50"
|
|
assert client.port == WIZ_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("network unreachable")
|
|
|
|
monkeypatch.setattr("ledgrab.core.devices.wiz_provider.discover_wiz_bulbs", _explode)
|
|
provider = WiZDeviceProvider()
|
|
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", "mac": "aabbccddeeff", "raw": {}},
|
|
# Missing IP should be skipped silently.
|
|
{"ip": "", "mac": "1234567890ab", "raw": {}},
|
|
]
|
|
|
|
monkeypatch.setattr("ledgrab.core.devices.wiz_provider.discover_wiz_bulbs", _fake)
|
|
provider = WiZDeviceProvider()
|
|
results = await provider.discover()
|
|
|
|
assert len(results) == 1
|
|
[bulb] = results
|
|
assert bulb.device_type == "wiz"
|
|
assert bulb.url == "wiz://192.168.1.50"
|
|
assert bulb.ip == "192.168.1.50"
|
|
assert bulb.mac == "aabbccddeeff"
|
|
# Last 6 chars of the MAC end up in the surface name for easy ID
|
|
assert "ddeeff" in bulb.name.lower()
|
|
|
|
|
|
# ============================================================================
|
|
# Device.to_config() round-trip
|
|
# ============================================================================
|
|
|
|
|
|
def test_device_to_config_round_trip_wiz():
|
|
from ledgrab.storage.device_store import Device
|
|
|
|
device = Device(
|
|
device_id="device_abc12345",
|
|
name="Bedroom WiZ",
|
|
url="wiz://192.168.1.42",
|
|
led_count=30,
|
|
device_type="wiz",
|
|
wiz_min_interval_ms=100,
|
|
)
|
|
|
|
config = device.to_config()
|
|
|
|
assert isinstance(config, WiZConfig)
|
|
assert config.device_url == "wiz://192.168.1.42"
|
|
assert config.led_count == 30
|
|
assert config.wiz_min_interval_ms == 100
|
|
|
|
|
|
def test_device_to_dict_omits_wiz_default_interval():
|
|
from ledgrab.storage.device_store import Device
|
|
|
|
device = Device(
|
|
device_id="device_abc12345",
|
|
name="Default",
|
|
url="wiz://192.168.1.42",
|
|
led_count=1,
|
|
device_type="wiz",
|
|
)
|
|
assert "wiz_min_interval_ms" not in device.to_dict()
|
|
|
|
|
|
def test_device_to_dict_preserves_non_default_wiz_interval():
|
|
from ledgrab.storage.device_store import Device
|
|
|
|
device = Device(
|
|
device_id="device_abc12345",
|
|
name="Custom",
|
|
url="wiz://192.168.1.42",
|
|
led_count=1,
|
|
device_type="wiz",
|
|
wiz_min_interval_ms=200,
|
|
)
|
|
assert device.to_dict()["wiz_min_interval_ms"] == 200
|
|
|
|
|
|
def test_device_from_dict_wiz_round_trip():
|
|
from ledgrab.storage.device_store import Device
|
|
|
|
restored = Device.from_dict(
|
|
{
|
|
"id": "device_abc12345",
|
|
"name": "Roundtrip",
|
|
"url": "wiz://10.0.0.1",
|
|
"led_count": 1,
|
|
"device_type": "wiz",
|
|
"wiz_min_interval_ms": 150,
|
|
}
|
|
)
|
|
assert restored.wiz_min_interval_ms == 150
|