90 lines
3.0 KiB
Python
90 lines
3.0 KiB
Python
from __future__ import annotations
|
|
|
|
from ha_deconz_bridge.events import DeconzEvent, DeconzSensorEvent, HaCommandEvent
|
|
from ha_deconz_bridge.queueing import EventBroker
|
|
|
|
|
|
def test_event_broker_coalesces_latest_ha_command() -> None:
|
|
broker = EventBroker()
|
|
broker.put_ha(HaCommandEvent("light", on=True, brightness=10))
|
|
broker.put_ha(HaCommandEvent("light", on=True, brightness=200))
|
|
broker.put_deconz(DeconzEvent("1", {"on": True}))
|
|
|
|
first = broker.get()
|
|
second = broker.get()
|
|
|
|
assert isinstance(first, HaCommandEvent)
|
|
assert isinstance(second, DeconzEvent)
|
|
assert first.brightness == 200
|
|
|
|
|
|
def test_event_broker_merges_partial_ha_commands() -> None:
|
|
broker = EventBroker()
|
|
broker.put_ha(HaCommandEvent("light", on=True, brightness=120, color_temp=400))
|
|
broker.put_ha(HaCommandEvent("light", rgb=(10, 20, 30)))
|
|
|
|
event = broker.get()
|
|
|
|
assert isinstance(event, HaCommandEvent)
|
|
assert event.on is True
|
|
assert event.brightness == 120
|
|
assert event.rgb == (10, 20, 30)
|
|
assert event.color_temp is None
|
|
|
|
|
|
def test_event_broker_moves_merged_ha_command_to_back() -> None:
|
|
broker = EventBroker()
|
|
broker.put_ha(HaCommandEvent("first", brightness=1))
|
|
broker.put_ha(HaCommandEvent("second", brightness=2))
|
|
broker.put_ha(HaCommandEvent("first", rgb=(1, 2, 3)))
|
|
|
|
first = broker.get()
|
|
second = broker.get()
|
|
|
|
assert isinstance(first, HaCommandEvent)
|
|
assert isinstance(second, HaCommandEvent)
|
|
assert first.light_name == "second"
|
|
assert second.light_name == "first"
|
|
assert second.brightness == 1
|
|
assert second.rgb == (1, 2, 3)
|
|
|
|
|
|
def test_event_broker_deduplicates_pending_deconz_events_by_controller() -> None:
|
|
broker = EventBroker()
|
|
broker.put_deconz(DeconzEvent("1", {"bri": 10}))
|
|
broker.put_deconz(DeconzEvent("2", {"bri": 20}))
|
|
broker.put_deconz(DeconzEvent("1", {"bri": 30}))
|
|
|
|
first = broker.get()
|
|
second = broker.get()
|
|
|
|
assert isinstance(first, DeconzEvent)
|
|
assert isinstance(second, DeconzEvent)
|
|
assert first.controller_id == "2"
|
|
assert first.raw_state == {"bri": 20}
|
|
assert second.controller_id == "1"
|
|
assert second.raw_state == {"bri": 30}
|
|
|
|
|
|
def test_event_broker_prioritizes_sensor_then_ha_then_light_events() -> None:
|
|
broker = EventBroker()
|
|
broker.put_ha(HaCommandEvent("light", on=True, brightness=10))
|
|
broker.put_deconz(DeconzEvent("1", {"on": True}))
|
|
broker.put_sensor(DeconzSensorEvent("4", {"buttonevent": 1002}))
|
|
|
|
assert isinstance(broker.get(), DeconzSensorEvent)
|
|
assert isinstance(broker.get(), HaCommandEvent)
|
|
assert isinstance(broker.get(), DeconzEvent)
|
|
|
|
|
|
def test_event_broker_does_not_let_light_readbacks_starve_ha_commands() -> None:
|
|
broker = EventBroker()
|
|
broker.put_deconz(DeconzEvent("1", {"bri": 10}))
|
|
broker.put_deconz(DeconzEvent("2", {"bri": 20}))
|
|
broker.put_ha(HaCommandEvent("light", on=True, brightness=255))
|
|
|
|
event = broker.get()
|
|
|
|
assert isinstance(event, HaCommandEvent)
|
|
assert event.brightness == 255
|