Typing lessons like duolingo & musicmouse cleanup

python-backend/tests/test_lirc.py

@@ -20,7 +20,7 @@ from musicmouse.events import (
     PlayAlbumRequested,
     PlayRequested,
     PlaySeriesLatestRequested,
-    PrevTrackRequested,
+    PreviousTrackRequested,
     VolumeChangeRequested,
 )
 from musicmouse.services.lirc import LircService
@@ -160,7 +160,7 @@ async def test_transport_and_volume_buttons_emit_intents(
         PlayRequested,
         PauseRequested,
         PauseRequested,
-        PrevTrackRequested,
+        PreviousTrackRequested,
         NextTrackRequested,
         VolumeChangeRequested,
     ]

python-backend/tests/test_mqtt.py

@@ -419,14 +419,18 @@ async def test_transport_buttons_emit_intents(
 
 
 async def test_the_player_sensor_reports_the_current_track(
-    bus: EventBus, mqtt_config: MqttConfig, player: FakePlayer, publisher: RecordingPublisher
+    bus: EventBus,
+    mqtt_config: MqttConfig,
+    mouse: MusicMouseDevice,
+    player: FakePlayer,
+    publisher: RecordingPublisher,
 ) -> None:
     from pathlib import Path
 
     from musicmouse.media import Playlist, Track
     from musicmouse.services.mqtt.player import PlayerSensor
 
-    entity = PlayerSensor(bus, mqtt_config, player)
+    entity = PlayerSensor(bus, mqtt_config, mouse, player)
     entity.attach(publisher)
     player.set_playlist(Playlist(name="fuchs", tracks=(Track(Path("/m/01 - Song.mp3")),)))
     player.play_from_start()
@@ -439,6 +443,29 @@ async def test_the_player_sensor_reports_the_current_track(
     assert attributes["volume"] == 40
 
 
+async def test_the_player_sensor_reports_the_mouse_s_active_figure(
+    bus: EventBus,
+    mqtt_config: MqttConfig,
+    transport: FakeTransport,
+    mouse: MusicMouseDevice,
+    player: FakePlayer,
+    publisher: RecordingPublisher,
+) -> None:
+    """The sensor reads ``mouse.active_figure`` rather than keeping its own copy of it,
+    so it agrees with the mouse even if it missed the event that changed it."""
+    from musicmouse.services.mqtt.player import PlayerSensor
+
+    entity = PlayerSensor(bus, mqtt_config, mouse, player)
+    entity.attach(publisher)
+
+    transport.inject(RfidTokenRead(tag_id=bytes.fromhex("04a1b2c3d4"), source="device"))
+    await bus.drain()
+
+    attributes = publisher.last_json(f"{entity.base_topic}/attributes")
+    assert attributes["figure"] == "fuchs"
+    assert mouse.active_figure == "fuchs"
+
+
 # --------------------------------------------------------------------- triggers
 
 

python-backend/tests/test_player.py

@@ -334,20 +334,76 @@ class _FakeMedia:
 
 
 class _FakeMediaPlayer:
-    """Just enough of libVLC's media player to exercise index syncing."""
+    """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 _vlc_like(bus: EventBus, playlist: Playlist) -> VlcPlayer:
+    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 = _FakeMediaPlayer()  # type: ignore[assignment]
+    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
@@ -365,7 +421,7 @@ async def test_the_track_index_is_read_back_off_the_player(bus: EventBus) -> Non
     events: list[Event] = []
     bus.subscribe(TrackChanged, events.append)
 
-    player._media_player.media = _FakeMedia("file:///music/2.mp3")  # type: ignore[attr-defined]
+    player._media_player.media = _FakeMedia("file:///music/2.mp3")
     player._on_next_item(object())
     await bus.drain()
 
@@ -379,10 +435,193 @@ 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  # type: ignore[attr-defined]
+    player._media_player.media = None
     player._on_next_item(object())
-    player._media_player.media = _FakeMedia("file:///elsewhere/x.mp3")  # type: ignore[attr-defined]
+    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")]