music-assistant-server

1"""Tests the event handling for LastFM Plugin Provider."""
2
3import logging
4
5import pytest
6from music_assistant_models.enums import EventType, MediaType
7from music_assistant_models.event import MassEvent
8from music_assistant_models.playback_progress_report import MediaItemPlaybackProgressReport
9
10from music_assistant.helpers.scrobbler import ScrobblerConfig, ScrobblerHelper
11
12
13class DummyHandler(ScrobblerHelper):
14    """Spy version of a ScrobblerHelper to allow easy testing."""
15
16    _tracked = 0
17    _now_playing = 0
18
19    def __init__(
20        self,
21        logger: logging.Logger,
22        config: ScrobblerConfig | None = None,
23        supported_media_types: frozenset[MediaType] | None = None,
24    ) -> None:
25        """Initialize."""
26        super().__init__(logger, config, supported_media_types)
27
28    def _is_configured(self) -> bool:
29        return True
30
31    async def _update_now_playing(self, report: MediaItemPlaybackProgressReport) -> None:
32        self._now_playing += 1
33
34    async def _scrobble(self, report: MediaItemPlaybackProgressReport) -> None:
35        self._tracked += 1
36
37
38async def test_it_does_not_scrobble_the_same_track_twice() -> None:
39    """
40    While songs are playing we get updates every 30 seconds.
41
42    Here we test that songs only get scrobbled once during each play.
43    """
44    handler = DummyHandler(logging.getLogger())
45
46    # not fully played yet
47    await handler._on_mass_media_item_played(create_report(duration=180, seconds_played=30))
48    assert handler._tracked == 0
49
50    # fully played near the end
51    await handler._on_mass_media_item_played(create_report(duration=180, seconds_played=176))
52    assert handler._tracked == 1
53
54    # fully played on track change should not scrobble again
55    await handler._on_mass_media_item_played(create_report(duration=180, seconds_played=180))
56    assert handler._tracked == 1
57
58    # single song is on repeat and started playing again
59    await handler._on_mass_media_item_played(create_report(duration=180, seconds_played=30))
60    assert handler._tracked == 1
61
62    # fully played for the second time
63    await handler._on_mass_media_item_played(create_report(duration=180, seconds_played=179))
64    assert handler._tracked == 2
65
66
67async def test_it_resets_now_playing_when_songs_are_on_loop() -> None:
68    """
69    When a song starts playing we update the 'now playing' endpoint.
70
71    This ends automatically, so if a single song is on repeat, we need to send the request again
72    """
73    handler = DummyHandler(logging.getLogger())
74
75    # started playing, should update now_playing
76    await handler._on_mass_media_item_played(create_report(duration=180, seconds_played=30))
77    assert handler._now_playing == 1
78
79    # fully played on track change should not update again
80    await handler._on_mass_media_item_played(create_report(duration=180, seconds_played=180))
81    assert handler._now_playing == 1
82
83    # restarted same song, should scrobble again
84    await handler._on_mass_media_item_played(create_report(duration=180, seconds_played=30))
85    assert handler._now_playing == 2
86
87
88async def test_it_does_not_update_now_playing_on_pause() -> None:
89    """Don't update now_playing when pausing the player early in the song."""
90    handler = DummyHandler(logging.getLogger())
91
92    await handler._on_mass_media_item_played(
93        create_report(duration=180, seconds_played=20, is_playing=False)
94    )
95    assert handler._now_playing == 0
96
97
98async def test_it_filters_scrobbles_by_player() -> None:
99    """Only scrobble tracks from configured players."""
100    handler = DummyHandler(
101        logging.getLogger(),
102        ScrobblerConfig(suffix_version=False, mass_playerids=["living_room"]),
103    )
104
105    await handler._on_mass_media_item_played(
106        create_report(duration=180, seconds_played=176, player_id="kitchen")
107    )
108    assert handler._now_playing == 0
109    assert handler._tracked == 0
110
111    await handler._on_mass_media_item_played(
112        create_report(duration=180, seconds_played=176, player_id="living_room")
113    )
114    assert handler._now_playing == 1
115    assert handler._tracked == 1
116
117
118async def test_it_filters_scrobbles_without_player_context() -> None:
119    """Skip scrobbling if a player filter is configured and the event has no player context."""
120    handler = DummyHandler(
121        logging.getLogger(),
122        ScrobblerConfig(suffix_version=False, mass_playerids=["living_room"]),
123    )
124
125    await handler._on_mass_media_item_played(
126        create_report(duration=180, seconds_played=176, player_id=None)
127    )
128    assert handler._now_playing == 0
129    assert handler._tracked == 0
130
131
132async def test_it_filters_unsupported_media_types() -> None:
133    """Only provider supported media types should be scrobbled."""
134    handler = DummyHandler(logging.getLogger(), supported_media_types=frozenset({MediaType.TRACK}))
135
136    await handler._on_mass_media_item_played(
137        create_report(
138            duration=180,
139            seconds_played=176,
140            uri="filesystem://audiobook/1",
141            media_type=MediaType.AUDIOBOOK,
142        )
143    )
144    assert handler._now_playing == 0
145    assert handler._tracked == 0
146
147
148async def test_it_allows_provider_supported_media_types() -> None:
149    """Providers can opt in to scrobbling additional media types."""
150    handler = DummyHandler(
151        logging.getLogger(),
152        supported_media_types=frozenset({MediaType.TRACK, MediaType.AUDIOBOOK}),
153    )
154
155    await handler._on_mass_media_item_played(
156        create_report(
157            duration=180,
158            seconds_played=176,
159            uri="filesystem://audiobook/1",
160            media_type=MediaType.AUDIOBOOK,
161        )
162    )
163    assert handler._now_playing == 1
164    assert handler._tracked == 1
165
166
167async def test_it_suffixes_the_version_if_enabled_and_available() -> None:
168    """Test that the track version is suffixed to the track name when enabled."""
169    report_with_version = create_report(version="Deluxe Edition").data
170    report_without_version = create_report(version=None).data
171
172    handler = DummyHandler(logging.getLogger(), ScrobblerConfig(suffix_version=True))
173    assert handler.get_name(report_with_version) == "track (Deluxe Edition)"
174    assert handler.get_name(report_without_version) == "track"
175
176    handler = DummyHandler(logging.getLogger(), ScrobblerConfig(suffix_version=False))
177    assert handler.get_name(report_with_version) == "track"
178    assert handler.get_name(report_without_version) == "track"
179
180
181class _ServiceError(Exception):
182    """Stand-in for a scrobble client's expected service/network error."""
183
184
185class FailingHandler(DummyHandler):
186    """Handler whose submissions always raise, to test exception handling."""
187
188    scrobble_exceptions = (_ServiceError,)
189
190    def __init__(self, logger: logging.Logger, error: Exception) -> None:
191        """Initialize with the error to raise on every submission."""
192        super().__init__(logger)
193        self._error = error
194
195    async def _update_now_playing(self, report: MediaItemPlaybackProgressReport) -> None:
196        raise self._error
197
198    async def _scrobble(self, report: MediaItemPlaybackProgressReport) -> None:
199        raise self._error
200
201
202async def test_it_swallows_expected_scrobble_exceptions() -> None:
203    """Errors listed in scrobble_exceptions are logged and swallowed, leaving state untouched."""
204    handler = FailingHandler(logging.getLogger(), _ServiceError("service unavailable"))
205
206    # a fully played, playing report drives both the now_playing and scrobble paths
207    await handler._on_mass_media_item_played(create_report(duration=180, seconds_played=176))
208
209    # neither marker advances because both submissions failed before assignment
210    assert handler.currently_playing is None
211    assert handler.last_scrobbled is None
212
213
214async def test_it_propagates_unexpected_scrobble_exceptions() -> None:
215    """Errors outside scrobble_exceptions surface instead of being silently swallowed."""
216    handler = FailingHandler(logging.getLogger(), ValueError("unexpected bug"))
217
218    with pytest.raises(ValueError, match="unexpected bug"):
219        await handler._on_mass_media_item_played(create_report(duration=180, seconds_played=176))
220
221
222def create_report(
223    duration: int = 148,
224    seconds_played: int = 59,
225    is_playing: bool = True,
226    uri: str = "filesystem://track/1",
227    version: str | None = None,
228    player_id: str | None = "test_player",
229    media_type: MediaType = MediaType.TRACK,
230) -> MassEvent:
231    """Create the MediaItemPlaybackProgressReport and wrap it in a MassEvent."""
232    return wrap_event(
233        MediaItemPlaybackProgressReport(
234            uri=uri,
235            media_type=media_type,
236            name="track",
237            artist=None,
238            artist_mbids=None,
239            album=None,
240            album_mbid=None,
241            image_url=None,
242            duration=duration,
243            mbid="",
244            seconds_played=seconds_played,
245            fully_played=duration - seconds_played < 5,
246            is_playing=is_playing,
247            version=version,
248            player_id=player_id,
249        )
250    )
251
252
253def wrap_event(data: MediaItemPlaybackProgressReport) -> MassEvent:
254    """Create a MEDIA_ITEM_PLAYED event."""
255    return MassEvent(EventType.MEDIA_ITEM_PLAYED, data.uri, data)
256