feat: roadmap round two (2026-06-23) — per-pixel smart-lights + integrations
A) Per-pixel smart-lights - LIFX multizone (SetExtendedColorZones msg 510, <=82 zones) + Tile (SetTileState64 715), auto-detected on connect with single-colour fallback; lifx_per_zone threaded like nanoleaf_per_panel - Hue gradient-lightstrip mapping: Entertainment v2 frame now keyed by channel id (was 1 light=1 LED), channels discovered on connect; hue_gradient_mode toggle (default on) B) Integrations bundle - Outbound webhook automation action (Discord/IFTTT/Zapier/Node-RED), SSRF-gated via validate_polling_url at both save and fire time; fires on activate/deactivate, best-effort, audited - Home Assistant MQTT auto-discovery: read-only binary_sensors per automation + connectivity, availability via birth/will, cleanup on disable/delete, live state from the engine Shared: pixel_reduce.resample_to_n nearest-neighbour helper. 57 new tests (lifx_multizone, hue_segment, webhook_action, ha_discovery). Gate: ruff + tsc + build clean, pytest 2719 passed / 2 skipped.
This commit is contained in:
@@ -0,0 +1,145 @@
|
||||
"""Tests for Home Assistant MQTT auto-discovery.
|
||||
|
||||
The end-to-end entity appearance, retained-config survival across HA restart,
|
||||
and availability flips need a live broker + HA. Here we lock down the parts
|
||||
that DON'T: the discovery config payloads, publish_all / remove_all topic sets,
|
||||
the MQTTSource field round-trip, and the manager's state fan-out.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime, timezone
|
||||
|
||||
import pytest
|
||||
|
||||
from ledgrab.core.mqtt.ha_discovery import HADiscoveryPublisher
|
||||
from ledgrab.core.mqtt.mqtt_manager import MQTTManager
|
||||
from ledgrab.storage.mqtt_source import MQTTSource
|
||||
|
||||
|
||||
class _FakeRuntime:
|
||||
def __init__(self) -> None:
|
||||
self.published: list[tuple[str, str, bool]] = []
|
||||
self.states: list[tuple[str, str]] = []
|
||||
|
||||
async def publish(self, topic: str, payload: str, retain: bool = False, qos: int = 0) -> None:
|
||||
self.published.append((topic, payload, retain))
|
||||
|
||||
async def publish_automation_state(self, automation_id: str, action: str) -> None:
|
||||
self.states.append((automation_id, action))
|
||||
|
||||
|
||||
class _Auto:
|
||||
def __init__(self, id: str, name: str) -> None:
|
||||
self.id = id
|
||||
self.name = name
|
||||
|
||||
|
||||
class _FakeAutomationStore:
|
||||
def __init__(self, autos) -> None:
|
||||
self._autos = autos
|
||||
|
||||
def get_all(self):
|
||||
return self._autos
|
||||
|
||||
|
||||
def _source(**kw) -> MQTTSource:
|
||||
now = datetime.now(timezone.utc)
|
||||
defaults = dict(id="mqs_1", name="Broker", created_at=now, updated_at=now)
|
||||
defaults.update(kw)
|
||||
return MQTTSource(**defaults)
|
||||
|
||||
|
||||
def _publisher(autos=None, **src_kw):
|
||||
runtime = _FakeRuntime()
|
||||
source = _source(**src_kw)
|
||||
store = _FakeAutomationStore(autos if autos is not None else [_Auto("auto_1", "Movie Night")])
|
||||
return HADiscoveryPublisher(runtime, source, store, version="9.9.9"), runtime
|
||||
|
||||
|
||||
class TestConfigBuilders:
|
||||
def test_connectivity_config_shape(self):
|
||||
pub, _ = _publisher(base_topic="ledgrab")
|
||||
topic, payload = pub.build_connectivity_config()
|
||||
assert topic == "homeassistant/binary_sensor/ledgrab_mqs_1/connectivity/config"
|
||||
assert payload["device_class"] == "connectivity"
|
||||
assert payload["state_topic"] == "ledgrab/status"
|
||||
assert payload["unique_id"] == "ledgrab_mqs_1_connectivity"
|
||||
assert payload["device"]["identifiers"] == ["ledgrab_mqs_1"]
|
||||
assert payload["device"]["sw_version"] == "9.9.9"
|
||||
|
||||
def test_automation_config_shape(self):
|
||||
pub, _ = _publisher(base_topic="lg")
|
||||
topic, payload = pub.build_automation_config(_Auto("auto_7", "Night"))
|
||||
assert topic == "homeassistant/binary_sensor/ledgrab_mqs_1/automation_auto_7/config"
|
||||
assert payload["state_topic"] == "lg/automation/auto_7/state"
|
||||
assert payload["value_template"] == "{{ value_json.action }}"
|
||||
assert payload["payload_on"] == "active" and payload["payload_off"] == "inactive"
|
||||
assert payload["availability_topic"] == "lg/status"
|
||||
assert payload["name"] == "Night"
|
||||
|
||||
def test_custom_discovery_prefix(self):
|
||||
pub, _ = _publisher(discovery_prefix="ha")
|
||||
topic, _ = pub.build_connectivity_config()
|
||||
assert topic.startswith("ha/binary_sensor/")
|
||||
|
||||
|
||||
class TestPublishRemove:
|
||||
@pytest.mark.asyncio
|
||||
async def test_publish_all_publishes_retained_configs_and_state(self):
|
||||
pub, runtime = _publisher(autos=[_Auto("a1", "One"), _Auto("a2", "Two")])
|
||||
await pub.publish_all()
|
||||
config_topics = [t for (t, p, r) in runtime.published if t.endswith("/config")]
|
||||
assert any("connectivity" in t for t in config_topics)
|
||||
assert sum("automation_" in t for t in config_topics) == 2
|
||||
assert all(r for (_t, _p, r) in runtime.published) # all retained
|
||||
# Seeded initial states.
|
||||
assert ("a1", "inactive") in runtime.states
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_remove_all_clears_with_empty_payload(self):
|
||||
pub, runtime = _publisher(autos=[_Auto("a1", "One")])
|
||||
await pub.publish_all()
|
||||
runtime.published.clear()
|
||||
await pub.remove_all()
|
||||
# Every clear is an empty retained payload to a config topic.
|
||||
assert runtime.published
|
||||
assert all(p == "" and r for (_t, p, r) in runtime.published)
|
||||
assert all(t.endswith("/config") for (t, _p, _r) in runtime.published)
|
||||
|
||||
|
||||
class TestSourceRoundTrip:
|
||||
def test_fields_round_trip(self):
|
||||
s = _source(publish_ha_discovery=True, discovery_prefix="ha")
|
||||
back = MQTTSource.from_dict(s.to_dict())
|
||||
assert back.publish_ha_discovery is True
|
||||
assert back.discovery_prefix == "ha"
|
||||
|
||||
def test_defaults_when_absent(self):
|
||||
back = MQTTSource.from_dict({"id": "x", "name": "n", "created_at": "", "updated_at": ""})
|
||||
assert back.publish_ha_discovery is False
|
||||
assert back.discovery_prefix == "homeassistant"
|
||||
|
||||
|
||||
class TestManagerStateFanout:
|
||||
@pytest.mark.asyncio
|
||||
async def test_publish_state_only_to_discovery_sources(self):
|
||||
mgr = MQTTManager(store=None, automation_store=None)
|
||||
runtime = _FakeRuntime()
|
||||
# Inject a fake runtime + mark it discovery-enabled.
|
||||
mgr._runtimes["mqs_1"] = (runtime, 1)
|
||||
mgr._discovery_sources.add("mqs_1")
|
||||
await mgr.publish_automation_state_all("auto_1", True)
|
||||
assert runtime.states == [("auto_1", "active")]
|
||||
await mgr.publish_automation_state_all("auto_1", False)
|
||||
assert runtime.states[-1] == ("auto_1", "inactive")
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_no_discovery_sources_is_noop(self):
|
||||
mgr = MQTTManager(store=None, automation_store=None)
|
||||
# Should not raise with no discovery sources.
|
||||
await mgr.publish_automation_state_all("auto_1", True)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
pytest.main([__file__, "-v"])
|
||||
@@ -0,0 +1,140 @@
|
||||
"""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"])
|
||||
@@ -0,0 +1,215 @@
|
||||
"""Tests for LIFX multizone (Z/Beam) + tile (Canvas) per-pixel streaming.
|
||||
|
||||
The device-side handshake (zone/tile auto-detection over UDP, firmware
|
||||
fallback) needs real hardware to validate; here we lock down the parts that
|
||||
DON'T: the exact packet framing for SetExtendedColorZones (510) and
|
||||
SetTileState64 (715), the StateMultiZone / StateDeviceChain reply parsers,
|
||||
the strip→element resample, the per-mode emit path, and the ``lifx_per_zone``
|
||||
config round-trip through the device store.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import struct
|
||||
|
||||
import pytest
|
||||
|
||||
from ledgrab.core.devices.lifx_client import (
|
||||
MSG_SET_COLOR,
|
||||
MSG_SET_EXTENDED_COLOR_ZONES,
|
||||
MSG_SET_TILE_STATE_64,
|
||||
MSG_STATE_DEVICE_CHAIN,
|
||||
MSG_STATE_MULTIZONE,
|
||||
LIFXClient,
|
||||
_build_packet,
|
||||
_build_set_extended_color_zones_payload,
|
||||
_build_set_tile_state64_payload,
|
||||
_parse_multizone_reply,
|
||||
_parse_state_device_chain,
|
||||
rgb_to_hsbk,
|
||||
)
|
||||
from ledgrab.core.devices.pixel_reduce import resample_to_n
|
||||
from ledgrab.storage.device_store import Device
|
||||
|
||||
_EXT_ZONE_MAX = 82
|
||||
_TILE_PIXELS = 64
|
||||
_TILE_STRUCT_SIZE = 55
|
||||
_TILE_CHAIN_MAX = 16
|
||||
|
||||
|
||||
class _FakeTransport:
|
||||
"""Captures every datagram the client emits via ``_send``."""
|
||||
|
||||
def __init__(self) -> None:
|
||||
self.sent: list[bytes] = []
|
||||
|
||||
def sendto(self, data: bytes) -> None:
|
||||
self.sent.append(bytes(data))
|
||||
|
||||
|
||||
def _client(mode: str = "single", *, zone_count: int = 0, tiles=None) -> LIFXClient:
|
||||
c = LIFXClient("lifx://1.2.3.4", led_count=10, per_zone=True)
|
||||
c._connected = True
|
||||
c._transport = _FakeTransport() # type: ignore[assignment]
|
||||
c._mode = mode
|
||||
c._zone_count = zone_count
|
||||
c._tiles = tiles or []
|
||||
return c
|
||||
|
||||
|
||||
def _msg_type(packet: bytes) -> int:
|
||||
"""LIFX protocol-header message type lives at byte offset 32."""
|
||||
return struct.unpack_from("<H", packet, 32)[0]
|
||||
|
||||
|
||||
class TestResampleToN:
|
||||
def test_nearest_neighbour_downsample(self):
|
||||
pixels = [[10, 10, 10], [20, 20, 20], [30, 30, 30], [40, 40, 40]]
|
||||
assert resample_to_n(pixels, 2) == [(10, 10, 10), (30, 30, 30)]
|
||||
|
||||
def test_empty_strip_is_black(self):
|
||||
assert resample_to_n([], 3) == [(0, 0, 0), (0, 0, 0), (0, 0, 0)]
|
||||
|
||||
def test_upsample_repeats(self):
|
||||
assert resample_to_n([[255, 0, 0]], 3) == [(255, 0, 0), (255, 0, 0), (255, 0, 0)]
|
||||
|
||||
def test_zero_n_is_empty(self):
|
||||
assert resample_to_n([[1, 2, 3]], 0) == []
|
||||
|
||||
|
||||
class TestExtendedColorZonesPayload:
|
||||
def test_framing_is_byte_exact(self):
|
||||
hsbk = [(100, 200, 300, 3500), (400, 500, 600, 3500)]
|
||||
payload = _build_set_extended_color_zones_payload(hsbk, duration_ms=0, zone_index=0)
|
||||
# header(8) + 82 fixed HSBK slots * 8 bytes
|
||||
assert len(payload) == 8 + _EXT_ZONE_MAX * 8
|
||||
duration, apply, zone_index, count = struct.unpack_from("<IBHB", payload, 0)
|
||||
assert (duration, apply, zone_index, count) == (0, 1, 0, 2)
|
||||
h, s, b, k = struct.unpack_from("<HHHH", payload, 8)
|
||||
assert (h, s, b, k) == (100, 200, 300, 3500)
|
||||
|
||||
def test_unused_slots_are_zero_padded(self):
|
||||
payload = _build_set_extended_color_zones_payload([(1, 2, 3, 4)])
|
||||
# Third slot (index 2) onward must be zero.
|
||||
assert payload[8 + 8 : 8 + _EXT_ZONE_MAX * 8] == b"\x00" * (8 * (_EXT_ZONE_MAX - 1))
|
||||
|
||||
def test_overflow_is_capped_at_82(self):
|
||||
payload = _build_set_extended_color_zones_payload([(1, 1, 1, 1)] * 200)
|
||||
_, _, _, count = struct.unpack_from("<IBHB", payload, 0)
|
||||
assert count == _EXT_ZONE_MAX
|
||||
assert len(payload) == 8 + _EXT_ZONE_MAX * 8
|
||||
|
||||
|
||||
class TestTileState64Payload:
|
||||
def test_framing_is_byte_exact(self):
|
||||
hsbk = [(11, 22, 33, 3500)]
|
||||
payload = _build_set_tile_state64_payload(hsbk, tile_index=2, x=0, y=0, width=8)
|
||||
# header(10) + 64 fixed HSBK slots * 8 bytes
|
||||
assert len(payload) == 10 + _TILE_PIXELS * 8
|
||||
tile_index, length, reserved, x, y, width, duration = struct.unpack_from(
|
||||
"<BBBBBBI", payload, 0
|
||||
)
|
||||
assert (tile_index, length, reserved, x, y, width, duration) == (2, 1, 0, 0, 0, 8, 0)
|
||||
h, s, b, k = struct.unpack_from("<HHHH", payload, 10)
|
||||
assert (h, s, b, k) == (11, 22, 33, 3500)
|
||||
|
||||
|
||||
class TestMultizoneReplyParser:
|
||||
def test_parses_state_multizone(self):
|
||||
body = struct.pack("<BB", 16, 0) + b"\x00" * (8 * 8) # count=16, index=0, 8 HSBK
|
||||
raw = _build_packet(msg_type=MSG_STATE_MULTIZONE, payload=body)
|
||||
parsed = _parse_multizone_reply(raw)
|
||||
assert parsed == {"count": 16, "index": 0}
|
||||
|
||||
def test_other_message_type_returns_none(self):
|
||||
raw = _build_packet(msg_type=MSG_SET_COLOR, payload=b"\x00" * 16)
|
||||
assert _parse_multizone_reply(raw) is None
|
||||
|
||||
|
||||
class TestDeviceChainParser:
|
||||
def _chain_packet(self, tile_specs: list[tuple[int, int]]) -> bytes:
|
||||
payload = bytearray(1 + _TILE_STRUCT_SIZE * _TILE_CHAIN_MAX + 1)
|
||||
payload[0] = 0 # start_index
|
||||
for i, (w, h) in enumerate(tile_specs):
|
||||
base = 1 + i * _TILE_STRUCT_SIZE
|
||||
payload[base + 16] = w
|
||||
payload[base + 17] = h
|
||||
payload[1 + _TILE_STRUCT_SIZE * _TILE_CHAIN_MAX] = len(tile_specs) # tile count
|
||||
return _build_packet(msg_type=MSG_STATE_DEVICE_CHAIN, payload=bytes(payload))
|
||||
|
||||
def test_parses_tile_widths_and_heights(self):
|
||||
raw = self._chain_packet([(8, 8), (8, 8)])
|
||||
parsed = _parse_state_device_chain(raw)
|
||||
assert parsed == {"start_index": 0, "tiles": [(8, 8), (8, 8)]}
|
||||
|
||||
def test_short_packet_returns_none(self):
|
||||
assert _parse_state_device_chain(b"\x00" * 40) is None
|
||||
|
||||
|
||||
class TestEmitPixels:
|
||||
def test_multizone_emits_one_extended_packet(self):
|
||||
c = _client("multizone", zone_count=4)
|
||||
c._emit_pixels([(255, 0, 0)] * 8, 255)
|
||||
sent = c._transport.sent # type: ignore[attr-defined]
|
||||
assert len(sent) == 1
|
||||
assert _msg_type(sent[0]) == MSG_SET_EXTENDED_COLOR_ZONES
|
||||
|
||||
def test_tile_emits_one_packet_per_tile(self):
|
||||
c = _client("tile", tiles=[(2, 2), (2, 2)])
|
||||
c._emit_pixels([(0, 255, 0)] * 8, 255)
|
||||
sent = c._transport.sent # type: ignore[attr-defined]
|
||||
assert len(sent) == 2
|
||||
assert all(_msg_type(p) == MSG_SET_TILE_STATE_64 for p in sent)
|
||||
|
||||
def test_single_mode_falls_back_to_set_color(self):
|
||||
c = _client("single")
|
||||
c._emit_pixels([(10, 20, 30)] * 4, 255)
|
||||
sent = c._transport.sent # type: ignore[attr-defined]
|
||||
assert len(sent) == 1
|
||||
assert _msg_type(sent[0]) == MSG_SET_COLOR
|
||||
|
||||
def test_device_led_count_reflects_mode(self):
|
||||
assert _client("multizone", zone_count=16).device_led_count == 16
|
||||
assert _client("tile", tiles=[(8, 8), (8, 8)]).device_led_count == 128
|
||||
assert _client("single").device_led_count == 10
|
||||
|
||||
|
||||
class TestBrightnessScaling:
|
||||
def test_brightness_scales_zone_colours(self):
|
||||
c = _client("multizone", zone_count=1)
|
||||
c._emit_pixels([(200, 100, 50)], 128)
|
||||
payload = c._transport.sent[0][36:] # type: ignore[index]
|
||||
h, s, b, k = struct.unpack_from("<HHHH", payload, 8)
|
||||
scale = 128 / 255.0
|
||||
exp_h, exp_s, exp_b, exp_k = rgb_to_hsbk(
|
||||
int(200 * scale), int(100 * scale), int(50 * scale)
|
||||
)
|
||||
assert (h, s, b) == (exp_h, exp_s, exp_b)
|
||||
|
||||
|
||||
class TestConfigRoundTrip:
|
||||
def _device(self, per_zone: bool) -> Device:
|
||||
return Device(
|
||||
device_id="d1",
|
||||
name="Beam",
|
||||
url="lifx://1.2.3.4",
|
||||
led_count=10,
|
||||
device_type="lifx",
|
||||
lifx_per_zone=per_zone,
|
||||
)
|
||||
|
||||
def test_per_zone_round_trips_through_store(self):
|
||||
d = self._device(True)
|
||||
assert d.to_dict().get("lifx_per_zone") is True
|
||||
back = Device.from_dict(d.to_dict())
|
||||
assert back.lifx_per_zone is True
|
||||
assert back.to_config().lifx_per_zone is True
|
||||
|
||||
def test_default_off_is_omitted(self):
|
||||
d = self._device(False)
|
||||
assert "lifx_per_zone" not in d.to_dict()
|
||||
assert d.to_config().lifx_per_zone is False
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
pytest.main([__file__, "-v"])
|
||||
@@ -0,0 +1,136 @@
|
||||
"""Tests for the outbound webhook automation action.
|
||||
|
||||
Covers the pure logic (template rendering, fire_on filtering, model
|
||||
round-trip) and the fire path with a mocked transport (success, non-2xx,
|
||||
SSRF-blocked). Live delivery to a real Discord/Zapier endpoint is out of
|
||||
scope for CI and is verified manually.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime, timezone
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
import respx
|
||||
|
||||
from ledgrab.core.automations.webhook_action import (
|
||||
fire_webhook_action,
|
||||
render_template,
|
||||
should_fire,
|
||||
)
|
||||
from ledgrab.storage.automation import Action, Automation, WebhookAction
|
||||
|
||||
|
||||
def _automation(actions=None) -> Automation:
|
||||
now = datetime.now(timezone.utc)
|
||||
return Automation(
|
||||
id="auto_1",
|
||||
name="Movie Night",
|
||||
enabled=True,
|
||||
rule_logic="or",
|
||||
rules=[],
|
||||
scene_preset_id=None,
|
||||
deactivation_mode="none",
|
||||
deactivation_scene_preset_id=None,
|
||||
created_at=now,
|
||||
updated_at=now,
|
||||
actions=actions or [],
|
||||
)
|
||||
|
||||
|
||||
class TestRenderTemplate:
|
||||
def test_substitutes_all_tokens(self):
|
||||
out = render_template(
|
||||
'{"name":"{{automation_name}}","id":"{{automation_id}}","ev":"{{event}}"}',
|
||||
_automation(),
|
||||
"activate",
|
||||
)
|
||||
assert '"name":"Movie Night"' in out
|
||||
assert '"id":"auto_1"' in out
|
||||
assert '"ev":"activate"' in out
|
||||
|
||||
def test_leaves_unknown_tokens(self):
|
||||
assert render_template("{{unknown}}", _automation(), "activate") == "{{unknown}}"
|
||||
|
||||
|
||||
class TestShouldFire:
|
||||
def test_matches_event_or_both(self):
|
||||
assert should_fire(WebhookAction(fire_on="activate"), "activate")
|
||||
assert not should_fire(WebhookAction(fire_on="activate"), "deactivate")
|
||||
assert should_fire(WebhookAction(fire_on="both"), "activate")
|
||||
assert should_fire(WebhookAction(fire_on="both"), "deactivate")
|
||||
|
||||
|
||||
class TestModelRoundTrip:
|
||||
def test_webhook_action_round_trips(self):
|
||||
a = WebhookAction(
|
||||
webhook_url="https://example.com/hook",
|
||||
method="PUT",
|
||||
body_template="hi {{event}}",
|
||||
content_type="text/plain",
|
||||
fire_on="both",
|
||||
)
|
||||
back = Action.from_dict(a.to_dict())
|
||||
assert isinstance(back, WebhookAction)
|
||||
assert back.webhook_url == "https://example.com/hook"
|
||||
assert back.method == "PUT"
|
||||
assert back.fire_on == "both"
|
||||
|
||||
def test_unknown_action_type_raises(self):
|
||||
with pytest.raises(ValueError):
|
||||
Action.from_dict({"action_type": "nope"})
|
||||
|
||||
def test_from_dict_normalises_bad_method_and_fire_on(self):
|
||||
a = WebhookAction.from_dict({"method": "delete", "fire_on": "whenever"})
|
||||
assert a.method == "POST"
|
||||
assert a.fire_on == "activate"
|
||||
|
||||
def test_automation_actions_survive_serialization(self):
|
||||
auto = _automation([WebhookAction(webhook_url="https://x.test/h", fire_on="both")])
|
||||
back = Automation.from_dict(auto.to_dict())
|
||||
assert len(back.actions) == 1
|
||||
assert isinstance(back.actions[0], WebhookAction)
|
||||
assert back.actions[0].webhook_url == "https://x.test/h"
|
||||
|
||||
def test_no_actions_omitted_from_dict(self):
|
||||
assert "actions" not in _automation().to_dict()
|
||||
|
||||
|
||||
class TestFire:
|
||||
@respx.mock
|
||||
@pytest.mark.asyncio
|
||||
async def test_success_returns_true(self):
|
||||
route = respx.post("http://93.184.216.34/hook").mock(return_value=httpx.Response(204))
|
||||
action = WebhookAction(webhook_url="http://93.184.216.34/hook", body_template="{{event}}")
|
||||
ok, err = await fire_webhook_action(action, _automation(), "activate")
|
||||
assert ok is True and err is None
|
||||
assert route.called
|
||||
# Body template was rendered and sent.
|
||||
assert route.calls.last.request.content == b"activate"
|
||||
|
||||
@respx.mock
|
||||
@pytest.mark.asyncio
|
||||
async def test_non_2xx_returns_error(self):
|
||||
respx.post("http://93.184.216.34/hook").mock(return_value=httpx.Response(500))
|
||||
action = WebhookAction(webhook_url="http://93.184.216.34/hook")
|
||||
ok, err = await fire_webhook_action(action, _automation(), "activate")
|
||||
assert ok is False and err == "HTTP 500"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_ssrf_blocked_loopback(self):
|
||||
# validate_polling_url must reject loopback — no HTTP call is made.
|
||||
action = WebhookAction(webhook_url="http://127.0.0.1:8080/admin")
|
||||
ok, err = await fire_webhook_action(action, _automation(), "activate")
|
||||
assert ok is False and "SSRF" in (err or "")
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_empty_url_returns_error(self):
|
||||
ok, err = await fire_webhook_action(
|
||||
WebhookAction(webhook_url=""), _automation(), "activate"
|
||||
)
|
||||
assert ok is False and err == "no URL configured"
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
pytest.main([__file__, "-v"])
|
||||
Reference in New Issue
Block a user