Martin Bauer
a7fb56c9fe
Requiring 3.13 meant the device needed an interpreter the distribution does not
have, which is what dragged uv in, and uv then had to be matched to the Pi's
32-bit userland by hand and to build Pillow from source because no armv7 wheel
exists for a 3.13 ABI. Dropping to 3.11 removes all of that: apt provides the
interpreter and piwheels has prebuilt armhf wheels for the native dependencies.
The 3.13-only syntax was shallow - PEP 695 throughout, which converts back
mechanically:
type X = Y -> X: TypeAlias = Y
type Handler[E: Event] -> E = TypeVar("E", bound=Event) plus a plain alias,
which is generic anyway because it carries a TypeVar
def f[T: Bound](...) -> a module-level TypeVar
Also drop the one @override (3.12, and static-only), and stop the lirc test
harness calling Server.close_clients(), which is 3.13: the scripted handler now
releases its connection when asked, which is what that call was there to force.
Verified on 3.11.14 - 485 passed, mypy strict clean - and still 496 passed on
the 3.14 dev venv, which additionally has the analysis extra.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
353 lines
12 KiB
Python
353 lines
12 KiB
Python
"""The IR remote: line parsing, and the service against a scripted lircd."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import contextlib
|
|
from collections.abc import AsyncIterator
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from musicmouse.clock import FakeClock
|
|
from musicmouse.config import LircConfig, RemoteSlotConfig, load_config
|
|
from musicmouse.events import (
|
|
ConnectionChanged,
|
|
Event,
|
|
IntentEvent,
|
|
NextTrackRequested,
|
|
PauseRequested,
|
|
PlayAlbumRequested,
|
|
PlayRequested,
|
|
PlaySeriesLatestRequested,
|
|
PreviousTrackRequested,
|
|
VolumeChangeRequested,
|
|
)
|
|
from musicmouse.services.lirc import LircService
|
|
from musicmouse.services.lirc.protocol import LircButtonEvent, parse_line
|
|
from musicmouse.simulator.harness import Simulation, build_simulation
|
|
from tests.conftest import VALID_CONFIG, write_config
|
|
|
|
# ---------------------------------------------------------------------- parse_line
|
|
|
|
|
|
def test_a_valid_line_is_parsed() -> None:
|
|
assert parse_line("0000000000001781 00 BTN_1 Hauppauge") == LircButtonEvent(
|
|
code="0000000000001781", repeat=0, button="BTN_1", remote="Hauppauge"
|
|
)
|
|
|
|
|
|
def test_a_repeat_field_is_hexadecimal() -> None:
|
|
event = parse_line("0000000000001781 0a BTN_1 Hauppauge")
|
|
assert event is not None
|
|
assert event.repeat == 10
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"line",
|
|
[
|
|
"",
|
|
"0000000000001781 00 BTN_1", # missing remote
|
|
"0000000000001781 00 BTN_1 Hauppauge extra", # extra token
|
|
"0000000000001781 zz BTN_1 Hauppauge", # non-hex repeat
|
|
],
|
|
)
|
|
def test_malformed_lines_are_rejected(line: str) -> None:
|
|
assert parse_line(line) is None
|
|
|
|
|
|
def test_surrounding_whitespace_is_ignored() -> None:
|
|
assert parse_line(" 0000000000001781 00 BTN_1 Hauppauge \n") is not None
|
|
|
|
|
|
# --------------------------------------------------------------------------- service
|
|
|
|
|
|
class ScriptedLircd:
|
|
"""A tiny stand-in for lircd: accepts one connection at a time and lets the test
|
|
push lines to whoever is currently connected."""
|
|
|
|
def __init__(self) -> None:
|
|
self._writer: asyncio.StreamWriter | None = None
|
|
self._connected = asyncio.Event()
|
|
self._closing = asyncio.Event()
|
|
self.connection_count = 0
|
|
|
|
async def _handle(self, _reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None:
|
|
self._writer = writer
|
|
self.connection_count += 1
|
|
self._connected.set()
|
|
with contextlib.suppress(asyncio.CancelledError):
|
|
await self._closing.wait() # held open until the test drops it
|
|
writer.close()
|
|
with contextlib.suppress(OSError, asyncio.CancelledError):
|
|
await writer.wait_closed()
|
|
|
|
def shutdown(self) -> None:
|
|
"""Let go of every connection still being held open, so the server can close."""
|
|
self._closing.set()
|
|
|
|
async def wait_connected(self) -> None:
|
|
await self._connected.wait()
|
|
|
|
async def send(self, line: str) -> None:
|
|
await self._connected.wait()
|
|
assert self._writer is not None
|
|
self._writer.write(f"{line}\n".encode())
|
|
await self._writer.drain()
|
|
# Real socket I/O, not the fake clock: give the reader a moment to actually
|
|
# see the bytes before the caller checks what happened.
|
|
await asyncio.sleep(0.02)
|
|
|
|
def drop(self) -> None:
|
|
"""Simulate the link dying: close the socket, and wait for a new connect."""
|
|
assert self._writer is not None
|
|
self._writer.close()
|
|
self._writer = None
|
|
self._connected = asyncio.Event()
|
|
|
|
|
|
@pytest.fixture
|
|
async def lircd() -> AsyncIterator[tuple[ScriptedLircd, asyncio.base_events.Server]]:
|
|
station = ScriptedLircd()
|
|
server = await asyncio.start_server(station._handle, "127.0.0.1", 0)
|
|
try:
|
|
yield station, server
|
|
finally:
|
|
server.close()
|
|
# The scripted connection handler holds its connection open forever (it does
|
|
# not know the test is done), so `wait_closed()` alone would hang here. 3.13
|
|
# has Server.close_clients() for exactly this; ask the handlers to let go
|
|
# instead, because the device runs the Pi's Python 3.11.
|
|
station.shutdown()
|
|
await server.wait_closed()
|
|
|
|
|
|
@pytest.fixture
|
|
async def sim(config_dir: Path) -> AsyncIterator[Simulation]:
|
|
config = load_config(write_config(config_dir, VALID_CONFIG))
|
|
# Real time for the socket I/O itself, virtual time for the reconnect sleep - tests
|
|
# drive that explicitly via `sim.clock.advance()` and drain the bus after.
|
|
simulation = await build_simulation(config, clock=FakeClock())
|
|
try:
|
|
yield simulation
|
|
finally:
|
|
await simulation.aclose()
|
|
|
|
|
|
def _lirc_config(port: int, *, reconnect_interval: float = 5.0) -> LircConfig:
|
|
return LircConfig(host="127.0.0.1", port=port, reconnect_interval=reconnect_interval)
|
|
|
|
|
|
async def _run_service(sim: Simulation, config: LircConfig) -> asyncio.Task[None]:
|
|
service = LircService(sim.app, config, clock=sim.clock)
|
|
return asyncio.create_task(service.run())
|
|
|
|
|
|
async def test_transport_and_volume_buttons_emit_intents(
|
|
lircd: tuple[ScriptedLircd, asyncio.base_events.Server], sim: Simulation
|
|
) -> None:
|
|
station, server = lircd
|
|
port = server.sockets[0].getsockname()[1]
|
|
seen: list[Event] = []
|
|
sim.bus.subscribe_all(seen.append)
|
|
|
|
task = await _run_service(sim, _lirc_config(port))
|
|
try:
|
|
await station.wait_connected()
|
|
await station.send("0 00 KEY_PLAY Hauppauge")
|
|
await station.send("0 00 KEY_PAUSE Hauppauge")
|
|
await station.send("0 00 KEY_STOP Hauppauge")
|
|
await station.send("0 00 KEY_PREVIOUS Hauppauge")
|
|
await station.send("0 00 KEY_FORWARD Hauppauge")
|
|
await station.send("0 00 KEY_VOLUMEUP Hauppauge")
|
|
await sim.bus.drain()
|
|
finally:
|
|
task.cancel()
|
|
|
|
kinds = [type(event) for event in seen if isinstance(event, IntentEvent)]
|
|
assert kinds == [
|
|
PlayRequested,
|
|
PauseRequested,
|
|
PauseRequested,
|
|
PreviousTrackRequested,
|
|
NextTrackRequested,
|
|
VolumeChangeRequested,
|
|
]
|
|
|
|
|
|
async def test_events_from_another_remote_are_ignored(
|
|
lircd: tuple[ScriptedLircd, asyncio.base_events.Server], sim: Simulation
|
|
) -> None:
|
|
station, server = lircd
|
|
port = server.sockets[0].getsockname()[1]
|
|
seen: list[Event] = []
|
|
sim.bus.subscribe_all(seen.append)
|
|
|
|
task = await _run_service(sim, _lirc_config(port))
|
|
try:
|
|
await station.wait_connected()
|
|
await station.send("0 00 KEY_PLAY small_led_remote")
|
|
await sim.bus.drain()
|
|
finally:
|
|
task.cancel()
|
|
|
|
assert not any(isinstance(event, IntentEvent) for event in seen)
|
|
|
|
|
|
async def test_transport_buttons_only_act_on_the_first_press(
|
|
lircd: tuple[ScriptedLircd, asyncio.base_events.Server], sim: Simulation
|
|
) -> None:
|
|
station, server = lircd
|
|
port = server.sockets[0].getsockname()[1]
|
|
seen: list[Event] = []
|
|
sim.bus.subscribe(NextTrackRequested, seen.append)
|
|
|
|
task = await _run_service(sim, _lirc_config(port))
|
|
try:
|
|
await station.wait_connected()
|
|
await station.send("0 00 KEY_FORWARD Hauppauge")
|
|
await station.send("0 01 KEY_FORWARD Hauppauge")
|
|
await station.send("0 02 KEY_FORWARD Hauppauge")
|
|
await sim.bus.drain()
|
|
finally:
|
|
task.cancel()
|
|
|
|
assert len(seen) == 1
|
|
|
|
|
|
async def test_volume_buttons_act_on_every_repeat(
|
|
lircd: tuple[ScriptedLircd, asyncio.base_events.Server], sim: Simulation
|
|
) -> None:
|
|
station, server = lircd
|
|
port = server.sockets[0].getsockname()[1]
|
|
seen: list[VolumeChangeRequested] = []
|
|
sim.bus.subscribe(VolumeChangeRequested, seen.append)
|
|
|
|
task = await _run_service(sim, _lirc_config(port))
|
|
try:
|
|
await station.wait_connected()
|
|
await station.send("0 00 KEY_VOLUMEDOWN Hauppauge")
|
|
await station.send("0 01 KEY_VOLUMEDOWN Hauppauge")
|
|
await station.send("0 02 KEY_VOLUMEDOWN Hauppauge")
|
|
await sim.bus.drain()
|
|
finally:
|
|
task.cancel()
|
|
|
|
assert len(seen) == 3
|
|
assert all(event.delta < 0 for event in seen)
|
|
|
|
|
|
async def test_mute_toggles_and_restores(
|
|
lircd: tuple[ScriptedLircd, asyncio.base_events.Server], sim: Simulation
|
|
) -> None:
|
|
station, server = lircd
|
|
port = server.sockets[0].getsockname()[1]
|
|
|
|
task = await _run_service(sim, _lirc_config(port))
|
|
try:
|
|
await station.wait_connected()
|
|
sim.player.set_volume(37)
|
|
await sim.bus.drain()
|
|
|
|
await station.send("0 00 KEY_MUTE Hauppauge")
|
|
await sim.bus.drain()
|
|
assert sim.player.volume == 0
|
|
|
|
await station.send("0 00 KEY_MUTE Hauppauge")
|
|
await sim.bus.drain()
|
|
assert sim.player.volume == 37
|
|
finally:
|
|
task.cancel()
|
|
|
|
|
|
async def test_a_digit_with_no_assignment_does_nothing(
|
|
lircd: tuple[ScriptedLircd, asyncio.base_events.Server], sim: Simulation
|
|
) -> None:
|
|
station, server = lircd
|
|
port = server.sockets[0].getsockname()[1]
|
|
seen: list[Event] = []
|
|
sim.bus.subscribe(PlayAlbumRequested, seen.append)
|
|
sim.bus.subscribe(PlaySeriesLatestRequested, seen.append)
|
|
|
|
task = await _run_service(sim, _lirc_config(port))
|
|
try:
|
|
await station.wait_connected()
|
|
await station.send("0 00 BTN_5 Hauppauge")
|
|
await sim.bus.drain()
|
|
finally:
|
|
task.cancel()
|
|
|
|
assert seen == []
|
|
|
|
|
|
async def test_an_album_digit_plays_from_the_beginning(
|
|
lircd: tuple[ScriptedLircd, asyncio.base_events.Server], sim: Simulation
|
|
) -> None:
|
|
station, server = lircd
|
|
port = server.sockets[0].getsockname()[1]
|
|
album = next(a for a in sim.app.library.albums if a.title == "Kinderparty Lieder")
|
|
sim.app.config.remote = {"3": RemoteSlotConfig(target_kind="album", target=album.id)}
|
|
|
|
task = await _run_service(sim, _lirc_config(port))
|
|
try:
|
|
await station.wait_connected()
|
|
await station.send("0 00 BTN_3 Hauppauge")
|
|
await sim.bus.drain()
|
|
finally:
|
|
task.cancel()
|
|
|
|
assert sim.player.playlist is not None
|
|
assert sim.player.playlist.album_id == album.id
|
|
assert sim.player.track_index == 0
|
|
|
|
|
|
async def test_a_series_digit_plays_the_latest_episode(
|
|
lircd: tuple[ScriptedLircd, asyncio.base_events.Server], sim: Simulation
|
|
) -> None:
|
|
station, server = lircd
|
|
port = server.sockets[0].getsockname()[1]
|
|
sim.app.config.remote = {"7": RemoteSlotConfig(target_kind="series", target="Wissen macht Ah")}
|
|
|
|
task = await _run_service(sim, _lirc_config(port))
|
|
try:
|
|
await station.wait_connected()
|
|
await station.send("0 00 BTN_7 Hauppauge")
|
|
await sim.bus.drain()
|
|
finally:
|
|
task.cancel()
|
|
|
|
assert sim.player.playlist is not None
|
|
played = sim.app.library.get(sim.player.playlist.album_id)
|
|
assert played is not None
|
|
assert played.title == "Neu"
|
|
|
|
|
|
async def test_a_dropped_link_reconnects(
|
|
lircd: tuple[ScriptedLircd, asyncio.base_events.Server], sim: Simulation
|
|
) -> None:
|
|
station, server = lircd
|
|
port = server.sockets[0].getsockname()[1]
|
|
seen: list[ConnectionChanged] = []
|
|
sim.bus.subscribe(ConnectionChanged, seen.append)
|
|
|
|
task = await _run_service(sim, _lirc_config(port, reconnect_interval=10.0))
|
|
try:
|
|
await station.wait_connected()
|
|
await sim.bus.drain()
|
|
assert station.connection_count == 1
|
|
|
|
station.drop()
|
|
await asyncio.sleep(0.05) # let the client notice EOF
|
|
await sim.bus.drain()
|
|
|
|
assert isinstance(sim.clock, FakeClock)
|
|
await sim.clock.advance(10.0)
|
|
await station.wait_connected()
|
|
await sim.bus.drain()
|
|
|
|
assert station.connection_count == 2
|
|
assert [event.connected for event in seen] == [True, False, True]
|
|
finally:
|
|
task.cancel()
|