music-assistant-server

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