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>
632 lines
18 KiB
Python
632 lines
18 KiB
Python
"""Tests for the shared player behaviour, exercised through FakePlayer.
|
|
|
|
VlcPlayer adds only the libVLC bindings on top of PlayerBase; it needs a real audio
|
|
device and is covered by the on-device checklist, not here - except for how it works
|
|
out which track is playing, which is stubbed out at the bottom of this file because
|
|
getting it wrong is invisible until a front-end displays it.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from collections.abc import AsyncIterator
|
|
from pathlib import Path
|
|
from typing import TypeVar
|
|
|
|
import pytest
|
|
|
|
from musicmouse.bus import EventBus
|
|
from musicmouse.clock import FakeClock
|
|
from musicmouse.devices.player import Player, PlayerBase, VlcPlayer
|
|
from musicmouse.events import (
|
|
Event,
|
|
PlaybackChanged,
|
|
PlaylistFinished,
|
|
TrackChanged,
|
|
VolumeChanged,
|
|
)
|
|
from musicmouse.media import Playlist, Track
|
|
from musicmouse.simulator.fake_player import FakePlayer
|
|
|
|
|
|
def playlist(name: str = "fuchs", count: int = 3) -> Playlist:
|
|
tracks = tuple(Track(Path(f"/music/{name}/{i}.mp3")) for i in range(count))
|
|
return Playlist(name=name, tracks=tracks)
|
|
|
|
|
|
@pytest.fixture
|
|
async def bus() -> AsyncIterator[EventBus]:
|
|
async with EventBus() as running:
|
|
yield running
|
|
|
|
|
|
@pytest.fixture
|
|
def clock(bus: EventBus) -> FakeClock:
|
|
return FakeClock(idle=bus.drain)
|
|
|
|
|
|
@pytest.fixture
|
|
def player(bus: EventBus, clock: FakeClock) -> FakePlayer:
|
|
return FakePlayer(bus, clock=clock, track_duration=10.0, initial_volume=50)
|
|
|
|
|
|
@pytest.fixture
|
|
def seen(bus: EventBus) -> list[Event]:
|
|
events: list[Event] = []
|
|
bus.subscribe_all(events.append)
|
|
return events
|
|
|
|
|
|
T = TypeVar("T", bound=Event)
|
|
|
|
|
|
def only(events: list[Event], event_type: type[T]) -> list[T]:
|
|
return [e for e in events if isinstance(e, event_type)]
|
|
|
|
|
|
# ------------------------------------------------------------------------ volume
|
|
|
|
|
|
async def test_volume_starts_at_the_configured_value(player: FakePlayer) -> None:
|
|
assert player.volume == 50
|
|
|
|
|
|
async def test_setting_volume_announces_it(
|
|
bus: EventBus, player: FakePlayer, seen: list[Event]
|
|
) -> None:
|
|
player.set_volume(30, source="mqtt")
|
|
await bus.drain()
|
|
|
|
assert player.volume == 30
|
|
assert only(seen, VolumeChanged) == [VolumeChanged(volume=30, source="mqtt")]
|
|
|
|
|
|
async def test_setting_the_same_volume_is_not_announced(
|
|
bus: EventBus, player: FakePlayer, seen: list[Event]
|
|
) -> None:
|
|
player.set_volume(50)
|
|
await bus.drain()
|
|
assert only(seen, VolumeChanged) == []
|
|
|
|
|
|
async def test_change_volume_is_relative(bus: EventBus, player: FakePlayer) -> None:
|
|
player.change_volume(-20)
|
|
await bus.drain()
|
|
assert player.volume == 30
|
|
|
|
|
|
async def test_volume_is_clamped_to_the_configured_range(bus: EventBus) -> None:
|
|
limited = FakePlayer(bus, min_volume=20, max_volume=60, initial_volume=40)
|
|
|
|
limited.set_volume(100)
|
|
assert limited.volume == 60
|
|
limited.set_volume(0)
|
|
assert limited.volume == 20
|
|
|
|
|
|
async def test_min_volume_of_zero_is_honoured(bus: EventBus) -> None:
|
|
"""Regression: `if self.volume_min and ...` treated a configured 0 as unset."""
|
|
limited = FakePlayer(bus, min_volume=0, max_volume=100, initial_volume=10)
|
|
|
|
limited.set_volume(-5)
|
|
|
|
assert limited.volume == 0
|
|
|
|
|
|
# --------------------------------------------------------------------- playback
|
|
|
|
|
|
async def test_play_from_start_starts_the_first_track(
|
|
bus: EventBus, player: FakePlayer, seen: list[Event]
|
|
) -> None:
|
|
player.set_playlist(playlist())
|
|
player.play_from_start()
|
|
await bus.drain()
|
|
|
|
assert player.is_playing
|
|
assert player.track_index == 0
|
|
assert player.current_track is not None
|
|
assert player.current_track.title == "0"
|
|
assert only(seen, PlaybackChanged)[-1].playing is True
|
|
|
|
|
|
async def test_switching_albums_while_playing_announces_the_new_track(
|
|
bus: EventBus, player: FakePlayer, seen: list[Event]
|
|
) -> None:
|
|
"""Regression: starting a new playlist from track 0 while already playing track 0
|
|
of the previous one used to look like a no-op index change, so no ``TrackChanged``
|
|
went out and front-ends never learned the album had switched."""
|
|
player.set_playlist(playlist(name="fuchs"))
|
|
player.play_from_start()
|
|
await bus.drain()
|
|
|
|
player.set_playlist(playlist(name="eule"))
|
|
player.play_from_start()
|
|
await bus.drain()
|
|
|
|
assert player.is_playing
|
|
assert player.playlist is not None
|
|
assert player.playlist.name == "eule"
|
|
changes = only(seen, TrackChanged)
|
|
assert changes[-1].index == 0
|
|
assert changes[-1].track is not None
|
|
assert changes[-1].track.path.parent.name == "eule"
|
|
|
|
|
|
async def test_playing_an_empty_playlist_does_nothing(
|
|
bus: EventBus, player: FakePlayer, caplog: pytest.LogCaptureFixture
|
|
) -> None:
|
|
player.set_playlist(Playlist(name="leer", tracks=()))
|
|
player.play_from_start()
|
|
await bus.drain()
|
|
|
|
assert not player.is_playing
|
|
assert "playlist is empty" in caplog.text
|
|
|
|
|
|
async def test_pause_and_resume(bus: EventBus, player: FakePlayer) -> None:
|
|
player.set_playlist(playlist())
|
|
player.play_from_start()
|
|
player.pause()
|
|
await bus.drain()
|
|
assert not player.is_playing
|
|
|
|
player.play()
|
|
await bus.drain()
|
|
assert player.is_playing
|
|
|
|
|
|
async def test_tracks_advance_as_time_passes(
|
|
bus: EventBus, player: FakePlayer, clock: FakeClock, seen: list[Event]
|
|
) -> None:
|
|
player.set_playlist(playlist(count=3))
|
|
player.play_from_start()
|
|
await bus.drain()
|
|
|
|
await clock.advance(10.0)
|
|
assert player.track_index == 1
|
|
|
|
await clock.advance(10.0)
|
|
assert player.track_index == 2
|
|
|
|
assert [e.index for e in only(seen, TrackChanged)] == [1, 2]
|
|
|
|
|
|
async def test_playlist_end_stops_playback_and_is_announced(
|
|
bus: EventBus, player: FakePlayer, clock: FakeClock, seen: list[Event]
|
|
) -> None:
|
|
player.set_playlist(playlist(count=2))
|
|
player.play_from_start()
|
|
|
|
await clock.advance(25.0)
|
|
|
|
assert only(seen, PlaylistFinished) == [PlaylistFinished(source="player")]
|
|
assert not player.is_playing
|
|
|
|
|
|
async def test_a_paused_player_does_not_advance(
|
|
bus: EventBus, player: FakePlayer, clock: FakeClock
|
|
) -> None:
|
|
player.set_playlist(playlist())
|
|
player.play_from_start()
|
|
await bus.drain()
|
|
player.pause()
|
|
|
|
await clock.advance(100.0)
|
|
|
|
assert player.track_index == 0
|
|
assert not player.is_playing
|
|
|
|
|
|
async def test_resuming_continues_the_remainder_of_the_track(
|
|
bus: EventBus, player: FakePlayer, clock: FakeClock
|
|
) -> None:
|
|
player.set_playlist(playlist())
|
|
player.play_from_start()
|
|
await clock.advance(7.0)
|
|
player.pause()
|
|
await clock.advance(100.0)
|
|
player.play()
|
|
|
|
await clock.advance(2.0)
|
|
assert player.track_index == 0 # 3s of the track were still left
|
|
|
|
await clock.advance(2.0)
|
|
assert player.track_index == 1
|
|
|
|
|
|
async def test_next_and_previous(bus: EventBus, player: FakePlayer) -> None:
|
|
player.set_playlist(playlist(count=3))
|
|
player.play_from_start()
|
|
await bus.drain()
|
|
|
|
player.next_track()
|
|
player.next_track()
|
|
await bus.drain()
|
|
assert player.track_index == 2
|
|
|
|
player.previous_track()
|
|
await bus.drain()
|
|
assert player.track_index == 1
|
|
|
|
|
|
async def test_previous_on_the_first_track_stays_there(bus: EventBus, player: FakePlayer) -> None:
|
|
player.set_playlist(playlist())
|
|
player.play_from_start()
|
|
player.previous_track()
|
|
await bus.drain()
|
|
|
|
assert player.track_index == 0
|
|
|
|
|
|
async def test_next_past_the_last_track_ends_the_playlist(
|
|
bus: EventBus, player: FakePlayer, seen: list[Event]
|
|
) -> None:
|
|
player.set_playlist(playlist(count=2))
|
|
player.play_from_start()
|
|
player.next_track()
|
|
player.next_track()
|
|
await bus.drain()
|
|
|
|
assert only(seen, PlaylistFinished) == [PlaylistFinished(source="player")]
|
|
assert not player.is_playing
|
|
|
|
|
|
async def test_skipping_restarts_the_track_timer(
|
|
bus: EventBus, player: FakePlayer, clock: FakeClock
|
|
) -> None:
|
|
player.set_playlist(playlist(count=3))
|
|
player.play_from_start()
|
|
await clock.advance(9.0)
|
|
player.next_track()
|
|
await bus.drain()
|
|
|
|
await clock.advance(9.0)
|
|
assert player.track_index == 1 # a fresh 10s, not the 1s left over
|
|
|
|
await clock.advance(2.0)
|
|
assert player.track_index == 2
|
|
|
|
|
|
async def test_stop_resets_playback(bus: EventBus, player: FakePlayer, clock: FakeClock) -> None:
|
|
player.set_playlist(playlist())
|
|
player.play_from_start()
|
|
player.stop()
|
|
await bus.drain()
|
|
|
|
assert not player.is_playing
|
|
await clock.advance(100.0)
|
|
assert player.track_index == 0
|
|
|
|
|
|
async def test_setting_a_new_playlist_resets_the_index(bus: EventBus, player: FakePlayer) -> None:
|
|
player.set_playlist(playlist(count=3))
|
|
player.play_from_start()
|
|
player.next_track()
|
|
await bus.drain()
|
|
|
|
player.set_playlist(playlist(name="eule", count=2))
|
|
|
|
assert player.track_index == 0
|
|
assert player.playlist is not None
|
|
assert player.playlist.name == "eule"
|
|
|
|
|
|
async def test_current_track_is_none_without_a_playlist(player: FakePlayer) -> None:
|
|
assert player.current_track is None
|
|
assert player.playlist is None
|
|
|
|
|
|
def test_fake_player_satisfies_the_player_protocol(player: FakePlayer) -> None:
|
|
check: Player = player
|
|
assert check.volume == player.volume
|
|
|
|
|
|
def _vlc_player_satisfies_the_player_protocol(real: VlcPlayer) -> Player:
|
|
"""Checked by mypy, not at runtime: VlcPlayer needs libVLC to instantiate."""
|
|
return real
|
|
|
|
|
|
# --------------------------------------------------------- track index without libVLC
|
|
|
|
|
|
class _FakeMedia:
|
|
def __init__(self, mrl: str) -> None:
|
|
self._mrl = mrl
|
|
|
|
def get_mrl(self) -> str:
|
|
return self._mrl
|
|
|
|
|
|
class _FakeMediaPlayer:
|
|
"""Just enough of libVLC's media player to exercise index syncing, pause/seek and
|
|
the volume/playing poll."""
|
|
|
|
def __init__(self) -> None:
|
|
self.media: _FakeMedia | None = None
|
|
self.paused = False
|
|
self.time_ms = 0
|
|
self.length_ms = 0
|
|
self.volume = 50
|
|
|
|
def get_media(self) -> _FakeMedia | None:
|
|
return self.media
|
|
|
|
def set_pause(self, flag: int) -> None:
|
|
self.paused = bool(flag)
|
|
|
|
def set_time(self, ms: int) -> None:
|
|
self.time_ms = ms
|
|
|
|
def get_time(self) -> int:
|
|
return self.time_ms
|
|
|
|
def get_length(self) -> int:
|
|
return self.length_ms
|
|
|
|
def audio_set_volume(self, volume: int) -> None:
|
|
self.volume = volume
|
|
|
|
def audio_get_volume(self) -> int:
|
|
return self.volume
|
|
|
|
|
|
class _FakeListPlayer:
|
|
"""Just enough of libVLC's list player to exercise play/stop and the is-playing
|
|
half of the poll."""
|
|
|
|
def __init__(self) -> None:
|
|
self.playing = False
|
|
self.play_calls = 0
|
|
self.stop_calls = 0
|
|
self.played_index: int | None = None
|
|
|
|
def play(self) -> None:
|
|
self.play_calls += 1
|
|
self.playing = True
|
|
|
|
def stop(self) -> None:
|
|
self.stop_calls += 1
|
|
self.playing = False
|
|
|
|
def is_playing(self) -> bool:
|
|
return self.playing
|
|
|
|
def play_item_at_index(self, index: int) -> None:
|
|
self.played_index = index
|
|
self.playing = True
|
|
|
|
|
|
def _vlc_like(
|
|
bus: EventBus,
|
|
playlist: Playlist,
|
|
*,
|
|
media_player: _FakeMediaPlayer | None = None,
|
|
list_player: _FakeListPlayer | None = None,
|
|
) -> VlcPlayer:
|
|
"""A VlcPlayer with its libVLC parts stubbed out, without calling __init__."""
|
|
player = object.__new__(VlcPlayer)
|
|
PlayerBase.__init__(player, bus)
|
|
player._media_player = media_player or _FakeMediaPlayer()
|
|
player._list_player = list_player or _FakeListPlayer()
|
|
player._playlist = playlist
|
|
player._mrl_to_index = {f"file://{track.path}": i for i, track in enumerate(playlist.tracks)}
|
|
return player
|
|
|
|
|
|
async def test_the_track_index_is_read_back_off_the_player(bus: EventBus) -> None:
|
|
"""Not taken from the event payload.
|
|
|
|
``MediaListPlayerNextItemSet`` carries a bare int rather than a Media on some
|
|
libVLC builds, and trusting it left the index pinned at zero: the UI showed track 1
|
|
of an album that was audibly on track 3.
|
|
"""
|
|
playlist = Playlist("test", tuple(Track(Path(f"/music/{i}.mp3")) for i in range(3)))
|
|
player = _vlc_like(bus, playlist)
|
|
events: list[Event] = []
|
|
bus.subscribe(TrackChanged, events.append)
|
|
|
|
player._media_player.media = _FakeMedia("file:///music/2.mp3")
|
|
player._on_next_item(object())
|
|
await bus.drain()
|
|
|
|
assert player.track_index == 2
|
|
assert player.current_track is not None
|
|
assert player.current_track.path.name == "2.mp3"
|
|
assert len(events) == 1
|
|
|
|
|
|
async def test_an_unknown_media_leaves_the_index_alone(bus: EventBus) -> None:
|
|
playlist = Playlist("test", tuple(Track(Path(f"/music/{i}.mp3")) for i in range(3)))
|
|
player = _vlc_like(bus, playlist)
|
|
|
|
player._media_player.media = None
|
|
player._on_next_item(object())
|
|
player._media_player.media = _FakeMedia("file:///elsewhere/x.mp3")
|
|
player._on_next_item(object())
|
|
await bus.drain()
|
|
|
|
assert player.track_index == 0
|
|
|
|
|
|
# ------------------------------------------------------ VlcPlayer actions, stubbed
|
|
|
|
|
|
async def test_vlc_play_starts_playback_when_idle(bus: EventBus) -> None:
|
|
list_player = _FakeListPlayer()
|
|
player = _vlc_like(bus, playlist(), list_player=list_player)
|
|
|
|
player.play()
|
|
|
|
assert list_player.play_calls == 1
|
|
|
|
|
|
async def test_vlc_play_is_a_no_op_once_already_playing(bus: EventBus) -> None:
|
|
"""Mirrors `FakePlayer.play()`'s own guard, so the two `Player` implementations
|
|
agree on what a repeated play() does instead of leaving it to libVLC."""
|
|
list_player = _FakeListPlayer()
|
|
player = _vlc_like(bus, playlist(), list_player=list_player)
|
|
player._set_playing(True)
|
|
|
|
player.play()
|
|
|
|
assert list_player.play_calls == 0
|
|
|
|
|
|
async def test_vlc_pause_pauses_when_playing(bus: EventBus) -> None:
|
|
media_player = _FakeMediaPlayer()
|
|
player = _vlc_like(bus, playlist(), media_player=media_player)
|
|
player._set_playing(True)
|
|
|
|
player.pause()
|
|
|
|
assert media_player.paused is True
|
|
|
|
|
|
async def test_vlc_pause_is_a_no_op_when_not_playing(bus: EventBus) -> None:
|
|
media_player = _FakeMediaPlayer()
|
|
player = _vlc_like(bus, playlist(), media_player=media_player)
|
|
|
|
player.pause()
|
|
|
|
assert media_player.paused is False
|
|
|
|
|
|
async def test_vlc_play_track_clamps_to_the_playlist_bounds(bus: EventBus) -> None:
|
|
list_player = _FakeListPlayer()
|
|
player = _vlc_like(bus, playlist(count=3), list_player=list_player)
|
|
|
|
player.play_track(99)
|
|
|
|
assert list_player.played_index == 2
|
|
|
|
|
|
async def test_vlc_play_track_on_an_empty_playlist_does_nothing(
|
|
bus: EventBus, caplog: pytest.LogCaptureFixture
|
|
) -> None:
|
|
list_player = _FakeListPlayer()
|
|
player = _vlc_like(bus, Playlist(name="leer", tracks=()), list_player=list_player)
|
|
|
|
with caplog.at_level("WARNING"):
|
|
player.play_track(0)
|
|
|
|
assert list_player.played_index is None
|
|
assert "playlist is empty" in caplog.text
|
|
|
|
|
|
async def test_vlc_seek_sets_the_time_in_milliseconds(bus: EventBus) -> None:
|
|
media_player = _FakeMediaPlayer()
|
|
player = _vlc_like(bus, playlist(), media_player=media_player)
|
|
|
|
player.seek(12.5)
|
|
|
|
assert media_player.time_ms == 12500
|
|
|
|
|
|
async def test_vlc_seek_clamps_negative_positions_to_zero(bus: EventBus) -> None:
|
|
media_player = _FakeMediaPlayer()
|
|
player = _vlc_like(bus, playlist(), media_player=media_player)
|
|
|
|
player.seek(-5.0)
|
|
|
|
assert media_player.time_ms == 0
|
|
|
|
|
|
async def test_vlc_position_and_duration_read_the_media_player(bus: EventBus) -> None:
|
|
media_player = _FakeMediaPlayer()
|
|
media_player.time_ms = 4_000
|
|
media_player.length_ms = 180_000
|
|
player = _vlc_like(bus, playlist(), media_player=media_player)
|
|
|
|
assert player.position == 4.0
|
|
assert player.duration == 180.0
|
|
|
|
|
|
async def test_vlc_position_floors_libvlcs_minus_one_at_zero(bus: EventBus) -> None:
|
|
"""libVLC reports -1 for both until a media is actually opened."""
|
|
media_player = _FakeMediaPlayer()
|
|
media_player.time_ms = -1
|
|
media_player.length_ms = -1
|
|
player = _vlc_like(bus, playlist(), media_player=media_player)
|
|
|
|
assert player.position == 0.0
|
|
assert player.duration == 0.0
|
|
|
|
|
|
async def test_vlc_apply_volume_pushes_to_the_media_player(bus: EventBus) -> None:
|
|
media_player = _FakeMediaPlayer()
|
|
player = _vlc_like(bus, playlist(), media_player=media_player)
|
|
|
|
player.set_volume(77)
|
|
|
|
assert media_player.volume == 77
|
|
|
|
|
|
async def test_vlc_poll_announces_a_volume_change_made_outside_the_app(
|
|
bus: EventBus, seen: list[Event]
|
|
) -> None:
|
|
"""The physical volume can move without going through `set_volume` - a hardware
|
|
knob on the amp, or another process - so the poll is what catches libVLC drifting
|
|
from what `PlayerBase` last announced."""
|
|
media_player = _FakeMediaPlayer()
|
|
media_player.volume = 50
|
|
player = _vlc_like(bus, playlist(), media_player=media_player)
|
|
|
|
media_player.volume = 65
|
|
player._poll()
|
|
await bus.drain()
|
|
|
|
assert player.volume == 65
|
|
assert only(seen, VolumeChanged) == [VolumeChanged(volume=65, source="player")]
|
|
|
|
|
|
async def test_vlc_poll_ignores_a_negative_volume_reading(bus: EventBus) -> None:
|
|
media_player = _FakeMediaPlayer()
|
|
player = _vlc_like(bus, playlist(), media_player=media_player)
|
|
media_player.volume = -1
|
|
|
|
player._poll()
|
|
|
|
assert player.volume == 50
|
|
|
|
|
|
async def test_vlc_poll_syncs_is_playing_from_the_list_player(
|
|
bus: EventBus, seen: list[Event]
|
|
) -> None:
|
|
list_player = _FakeListPlayer()
|
|
player = _vlc_like(bus, playlist(), list_player=list_player)
|
|
list_player.playing = True
|
|
|
|
player._poll()
|
|
await bus.drain()
|
|
|
|
assert player.is_playing
|
|
assert only(seen, PlaybackChanged)[-1].playing is True
|
|
|
|
|
|
async def test_vlc_on_playing_and_on_stopped_announce_playback_changed(
|
|
bus: EventBus, seen: list[Event]
|
|
) -> None:
|
|
player = _vlc_like(bus, playlist())
|
|
|
|
player._on_playing(object())
|
|
await bus.drain()
|
|
assert player.is_playing
|
|
|
|
player._on_stopped(object())
|
|
await bus.drain()
|
|
assert not player.is_playing
|
|
assert [e.playing for e in only(seen, PlaybackChanged)] == [True, False]
|
|
|
|
|
|
async def test_vlc_on_playlist_end_stops_and_announces_playlist_finished(
|
|
bus: EventBus, seen: list[Event]
|
|
) -> None:
|
|
player = _vlc_like(bus, playlist())
|
|
player._set_playing(True)
|
|
|
|
player._on_playlist_end(object())
|
|
await bus.drain()
|
|
|
|
assert not player.is_playing
|
|
assert only(seen, PlaylistFinished) == [PlaylistFinished(source="player")]
|