/
/
/
1"""Tests the event handling for LastFM Plugin Provider."""
2
3import logging
4
5from music_assistant_models.enums import EventType, MediaType
6from music_assistant_models.event import MassEvent
7from music_assistant_models.playback_progress_report import MediaItemPlaybackProgressReport
8
9from music_assistant.helpers.scrobbler import ScrobblerConfig, ScrobblerHelper
10
11
12class DummyHandler(ScrobblerHelper):
13 """Spy version of a ScrobblerHelper to allow easy testing."""
14
15 _tracked = 0
16 _now_playing = 0
17
18 def __init__(self, logger: logging.Logger, config: ScrobblerConfig | None = None) -> None:
19 """Initialize."""
20 super().__init__(logger, config)
21
22 def _is_configured(self) -> bool:
23 return True
24
25 async def _update_now_playing(self, report: MediaItemPlaybackProgressReport) -> None:
26 self._now_playing += 1
27
28 async def _scrobble(self, report: MediaItemPlaybackProgressReport) -> None:
29 self._tracked += 1
30
31
32async def test_it_does_not_scrobble_the_same_track_twice() -> None:
33 """While songs are playing we get updates every 30 seconds.
34
35 Here we test that songs only get scrobbled once during each play.
36 """
37 handler = DummyHandler(logging.getLogger())
38
39 # not fully played yet
40 await handler._on_mass_media_item_played(create_report(duration=180, seconds_played=30))
41 assert handler._tracked == 0
42
43 # fully played near the end
44 await handler._on_mass_media_item_played(create_report(duration=180, seconds_played=176))
45 assert handler._tracked == 1
46
47 # fully played on track change should not scrobble again
48 await handler._on_mass_media_item_played(create_report(duration=180, seconds_played=180))
49 assert handler._tracked == 1
50
51 # single song is on repeat and started playing again
52 await handler._on_mass_media_item_played(create_report(duration=180, seconds_played=30))
53 assert handler._tracked == 1
54
55 # fully played for the second time
56 await handler._on_mass_media_item_played(create_report(duration=180, seconds_played=179))
57 assert handler._tracked == 2
58
59
60async def test_it_resets_now_playing_when_songs_are_on_loop() -> None:
61 """When a song starts playing we update the 'now playing' endpoint.
62
63 This ends automatically, so if a single song is on repeat, we need to send the request again
64 """
65 handler = DummyHandler(logging.getLogger())
66
67 # started playing, should update now_playing
68 await handler._on_mass_media_item_played(create_report(duration=180, seconds_played=30))
69 assert handler._now_playing == 1
70
71 # fully played on track change should not update again
72 await handler._on_mass_media_item_played(create_report(duration=180, seconds_played=180))
73 assert handler._now_playing == 1
74
75 # restarted same song, should scrobble again
76 await handler._on_mass_media_item_played(create_report(duration=180, seconds_played=30))
77 assert handler._now_playing == 2
78
79
80async def test_it_does_not_update_now_playing_on_pause() -> None:
81 """Don't update now_playing when pausing the player early in the song."""
82 handler = DummyHandler(logging.getLogger())
83
84 await handler._on_mass_media_item_played(
85 create_report(duration=180, seconds_played=20, is_playing=False)
86 )
87 assert handler._now_playing == 0
88
89
90async def test_it_suffixes_the_version_if_enabled_and_available() -> None:
91 """Test that the track version is suffixed to the track name when enabled."""
92 report_with_version = create_report(version="Deluxe Edition").data
93 report_without_version = create_report(version=None).data
94
95 handler = DummyHandler(logging.getLogger(), ScrobblerConfig(suffix_version=True))
96 assert handler.get_name(report_with_version) == "track (Deluxe Edition)"
97 assert handler.get_name(report_without_version) == "track"
98
99 handler = DummyHandler(logging.getLogger(), ScrobblerConfig(suffix_version=False))
100 assert handler.get_name(report_with_version) == "track"
101 assert handler.get_name(report_without_version) == "track"
102
103
104def create_report(
105 duration: int = 148,
106 seconds_played: int = 59,
107 is_playing: bool = True,
108 uri: str = "filesystem://track/1",
109 version: str | None = None,
110) -> MassEvent:
111 """Create the MediaItemPlaybackProgressReport and wrap it in a MassEvent."""
112 return wrap_event(
113 MediaItemPlaybackProgressReport(
114 uri=uri,
115 media_type=MediaType.TRACK,
116 name="track",
117 artist=None,
118 artist_mbids=None,
119 album=None,
120 album_mbid=None,
121 image_url=None,
122 duration=duration,
123 mbid="",
124 seconds_played=seconds_played,
125 fully_played=duration - seconds_played < 5,
126 is_playing=is_playing,
127 version=version,
128 )
129 )
130
131
132def wrap_event(data: MediaItemPlaybackProgressReport) -> MassEvent:
133 """Create a MEDIA_ITEM_PLAYED event."""
134 return MassEvent(EventType.MEDIA_ITEM_PLAYED, data.uri, data)
135