343 lines
11 KiB
Python
343 lines
11 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.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 asyncio.Event().wait() # held open until the test drops it
|
|
|
|
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.
|
|
server.close_clients()
|
|
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()
|