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