Martin Bauer
d44c24ec97
- event bus systen - all components are independent - preparation for web frontend
179 lines
4.6 KiB
Python
179 lines
4.6 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import logging
|
|
from collections.abc import AsyncIterator
|
|
|
|
import pytest
|
|
|
|
from musicmouse.bus import EventBus
|
|
from musicmouse.events import (
|
|
ButtonEvent,
|
|
Event,
|
|
InputEvent,
|
|
NextTrackRequested,
|
|
RfidTokenRead,
|
|
VolumeChanged,
|
|
)
|
|
from musicmouse.hardware import Button, ButtonAction
|
|
|
|
|
|
@pytest.fixture
|
|
async def bus() -> AsyncIterator[EventBus]:
|
|
async with EventBus() as running_bus:
|
|
yield running_bus
|
|
|
|
|
|
def _rfid(tag: str = "04a1b2c3d4") -> RfidTokenRead:
|
|
return RfidTokenRead(tag_id=bytes.fromhex(tag), figure="fuchs", source="device")
|
|
|
|
|
|
async def test_handler_receives_its_event(bus: EventBus) -> None:
|
|
seen: list[Event] = []
|
|
bus.subscribe(RfidTokenRead, seen.append)
|
|
|
|
event = _rfid()
|
|
await bus.emit_and_wait(event)
|
|
|
|
assert seen == [event]
|
|
|
|
|
|
async def test_unrelated_handlers_are_not_called(bus: EventBus) -> None:
|
|
seen: list[Event] = []
|
|
bus.subscribe(NextTrackRequested, seen.append)
|
|
|
|
await bus.emit_and_wait(_rfid())
|
|
|
|
assert seen == []
|
|
|
|
|
|
async def test_subscribing_to_a_base_class_catches_subclasses(bus: EventBus) -> None:
|
|
seen: list[Event] = []
|
|
bus.subscribe(InputEvent, seen.append)
|
|
|
|
await bus.emit_and_wait(_rfid())
|
|
await bus.emit_and_wait(VolumeChanged(volume=30))
|
|
|
|
assert [type(e) for e in seen] == [RfidTokenRead]
|
|
|
|
|
|
async def test_subscribe_all_sees_everything(bus: EventBus) -> None:
|
|
seen: list[Event] = []
|
|
bus.subscribe_all(seen.append)
|
|
|
|
await bus.emit_and_wait(_rfid())
|
|
await bus.emit_and_wait(VolumeChanged(volume=30))
|
|
|
|
assert [type(e) for e in seen] == [RfidTokenRead, VolumeChanged]
|
|
|
|
|
|
async def test_async_handlers_are_awaited(bus: EventBus) -> None:
|
|
seen: list[str] = []
|
|
|
|
async def slow(_: Event) -> None:
|
|
await asyncio.sleep(0.01)
|
|
seen.append("slow")
|
|
|
|
bus.subscribe(RfidTokenRead, slow)
|
|
bus.subscribe(RfidTokenRead, lambda _: seen.append("fast"))
|
|
|
|
await bus.emit_and_wait(_rfid())
|
|
|
|
assert seen == ["slow", "fast"]
|
|
|
|
|
|
async def test_events_are_dispatched_in_emission_order(bus: EventBus) -> None:
|
|
"""Ordering is what makes 'last event wins' meaningful for LED arbitration."""
|
|
seen: list[int] = []
|
|
|
|
async def record(event: VolumeChanged) -> None:
|
|
await asyncio.sleep(0)
|
|
seen.append(event.volume)
|
|
|
|
bus.subscribe(VolumeChanged, record)
|
|
|
|
for volume in range(5):
|
|
bus.emit(VolumeChanged(volume=volume))
|
|
await bus.drain()
|
|
|
|
assert seen == [0, 1, 2, 3, 4]
|
|
|
|
|
|
async def test_events_emitted_from_a_handler_are_handled_before_drain_returns(
|
|
bus: EventBus,
|
|
) -> None:
|
|
seen: list[str] = []
|
|
|
|
def on_button(_: ButtonEvent) -> None:
|
|
seen.append("button")
|
|
bus.emit(NextTrackRequested(source="device"))
|
|
|
|
bus.subscribe(ButtonEvent, on_button)
|
|
bus.subscribe(NextTrackRequested, lambda _: seen.append("next"))
|
|
|
|
await bus.emit_and_wait(
|
|
ButtonEvent(button=Button.RIGHT, action=ButtonAction.PRESSED, source="device")
|
|
)
|
|
|
|
assert seen == ["button", "next"]
|
|
|
|
|
|
async def test_a_raising_handler_does_not_stop_the_others(
|
|
bus: EventBus, caplog: pytest.LogCaptureFixture
|
|
) -> None:
|
|
seen: list[str] = []
|
|
|
|
def boom(_: Event) -> None:
|
|
raise RuntimeError("handler is broken")
|
|
|
|
bus.subscribe(RfidTokenRead, boom)
|
|
bus.subscribe(RfidTokenRead, lambda _: seen.append("survivor"))
|
|
|
|
with caplog.at_level(logging.ERROR):
|
|
await bus.emit_and_wait(_rfid())
|
|
await bus.emit_and_wait(_rfid())
|
|
|
|
assert seen == ["survivor", "survivor"]
|
|
assert "handler is broken" in caplog.text
|
|
|
|
|
|
async def test_unsubscribe(bus: EventBus) -> None:
|
|
seen: list[Event] = []
|
|
unsubscribe = bus.subscribe(RfidTokenRead, seen.append)
|
|
|
|
await bus.emit_and_wait(_rfid())
|
|
unsubscribe()
|
|
await bus.emit_and_wait(_rfid())
|
|
|
|
assert len(seen) == 1
|
|
|
|
|
|
async def test_emit_is_safe_from_another_thread(bus: EventBus) -> None:
|
|
"""libVLC fires its callbacks off-loop; this is the crossing that must be safe."""
|
|
seen: list[VolumeChanged] = []
|
|
done = asyncio.Event()
|
|
|
|
def record(event: VolumeChanged) -> None:
|
|
seen.append(event)
|
|
done.set()
|
|
|
|
bus.subscribe(VolumeChanged, record)
|
|
|
|
await asyncio.to_thread(bus.emit, VolumeChanged(volume=42, source="player"))
|
|
async with asyncio.timeout(2):
|
|
await done.wait()
|
|
|
|
assert [e.volume for e in seen] == [42]
|
|
|
|
|
|
async def test_emit_before_start_is_an_error() -> None:
|
|
with pytest.raises(RuntimeError, match="before start"):
|
|
EventBus().emit(VolumeChanged(volume=1))
|
|
|
|
|
|
async def test_stop_is_idempotent() -> None:
|
|
stopped = EventBus()
|
|
await stopped.start()
|
|
await stopped.stop()
|
|
await stopped.stop()
|