music-assistant-server

1"""Tests for the shared RecencyEngine (music/recency.py)."""
2
3from __future__ import annotations
4
5import time
6
7from music_assistant_models.enums import MediaType
8from music_assistant_models.media_items import ItemMapping, ProviderMapping, Track
9from music_assistant_models.unique_list import UniqueList
10
11from music_assistant.constants import DB_TABLE_PLAYLOG
12from music_assistant.controllers.music.recency import RecencyWindows
13from music_assistant.helpers.json import serialize_to_json
14from music_assistant.mass import MusicAssistant
15
16
17async def _add_playlog_row(
18    mass: MusicAssistant,
19    *,
20    item_id: str,
21    provider: str,
22    media_type: MediaType,
23    name: str,
24    timestamp: int,
25    userid: str = "user-a",
26    artist_names: list[str] | None = None,
27    fully_played: bool = True,
28) -> None:
29    """Insert a single row into the playlog."""
30    await mass.music.database.insert(
31        DB_TABLE_PLAYLOG,
32        {
33            "item_id": item_id,
34            "provider": provider,
35            "media_type": media_type.value,
36            "name": name,
37            "artists": serialize_to_json([{"name": artist_name} for artist_name in artist_names])
38            if artist_names
39            else None,
40            "timestamp": timestamp,
41            "fully_played": fully_played,
42            "seconds_played": 180 if fully_played else 30,
43            "userid": userid,
44            "user_initiated": True,
45        },
46    )
47
48
49def _track(item_id: str, provider: str, *, mapping: ProviderMapping | None = None) -> Track:
50    """Build a minimal Track with an optional single provider mapping."""
51    mappings = {mapping} if mapping else set()
52    return Track(
53        item_id=item_id,
54        provider=provider,
55        name=f"Track {item_id}",
56        uri=f"{provider}://track/{item_id}",
57        provider_mappings=mappings,
58    )
59
60
61def _named_track(item_id: str, name: str, artist_names: list[str]) -> Track:
62    """Build a Track with the given title and artist names (no provider mappings)."""
63    return Track(
64        item_id=item_id,
65        provider="library",
66        name=name,
67        uri=f"library://track/{item_id}",
68        provider_mappings=set(),
69        artists=UniqueList(
70            [
71                ItemMapping(
72                    item_id=artist_name.lower(),
73                    provider="library",
74                    name=artist_name,
75                    media_type=MediaType.ARTIST,
76                )
77                for artist_name in artist_names
78            ]
79        ),
80    )
81
82
83async def test_snapshot_builds_song_and_artist_maps(mass: MusicAssistant) -> None:
84    """Track rows populate song_ts; artist rows populate artist_ts (lowercased)."""
85    now = int(time.time())
86    await _add_playlog_row(
87        mass,
88        item_id="t1",
89        provider="library",
90        media_type=MediaType.TRACK,
91        name="T1",
92        timestamp=now - 60,
93    )
94    await _add_playlog_row(
95        mass,
96        item_id="a1",
97        provider="library",
98        media_type=MediaType.ARTIST,
99        name="The Beatles",
100        timestamp=now - 60,
101    )
102
103    snapshot = await mass.music.recency.snapshot(
104        RecencyWindows(song_seconds=3600, artist_seconds=3600), userid="user-a"
105    )
106
107    assert ("library", "t1") in snapshot.song_ts
108    assert "the beatles" in snapshot.artist_ts
109
110
111async def test_snapshot_can_include_partially_played_tracks(mass: MusicAssistant) -> None:
112    """Include interrupted tracks only when the caller opts in."""
113    now = int(time.time())
114    await _add_playlog_row(
115        mass,
116        item_id="partial",
117        provider="library",
118        media_type=MediaType.TRACK,
119        name="Partial",
120        timestamp=now - 60,
121        fully_played=False,
122    )
123
124    default_snapshot = await mass.music.recency.snapshot(
125        RecencyWindows(song_seconds=3600), userid="user-a"
126    )
127    inclusive_snapshot = await mass.music.recency.snapshot(
128        RecencyWindows(song_seconds=3600),
129        userid="user-a",
130        include_partially_played=True,
131    )
132
133    assert ("library", "partial") not in default_snapshot.song_ts
134    assert ("library", "partial") in inclusive_snapshot.song_ts
135
136
137async def test_snapshot_is_user_scoped(mass: MusicAssistant) -> None:
138    """A snapshot for one user does not see another user's plays."""
139    now = int(time.time())
140    await _add_playlog_row(
141        mass,
142        item_id="mine",
143        provider="library",
144        media_type=MediaType.TRACK,
145        name="Mine",
146        timestamp=now - 60,
147        userid="user-a",
148    )
149    await _add_playlog_row(
150        mass,
151        item_id="theirs",
152        provider="library",
153        media_type=MediaType.TRACK,
154        name="Theirs",
155        timestamp=now - 60,
156        userid="user-b",
157    )
158
159    snapshot = await mass.music.recency.snapshot(RecencyWindows(song_seconds=3600), userid="user-a")
160
161    assert ("library", "mine") in snapshot.song_ts
162    assert ("library", "theirs") not in snapshot.song_ts
163
164
165async def test_track_recent_matches_own_id_and_provider_mapping(mass: MusicAssistant) -> None:
166    """track_recent matches both a library play (own id) and a streaming play (via mapping)."""
167    now = int(time.time())
168    # a library play and a streaming play, recorded under different provider keys
169    await _add_playlog_row(
170        mass,
171        item_id="lib1",
172        provider="library",
173        media_type=MediaType.TRACK,
174        name="Lib",
175        timestamp=now - 60,
176    )
177    await _add_playlog_row(
178        mass,
179        item_id="sp1",
180        provider="spotify--x",
181        media_type=MediaType.TRACK,
182        name="Stream",
183        timestamp=now - 60,
184    )
185    snapshot = await mass.music.recency.snapshot(RecencyWindows(song_seconds=3600), userid="user-a")
186    stream_mapping = ProviderMapping(
187        item_id="sp1", provider_domain="spotify", provider_instance="spotify--x"
188    )
189
190    # matched via the track's own (library) id
191    assert snapshot.track_recent(_track("lib1", "library"), 3600) is True
192    # matched via a streaming provider mapping while the track itself is the library item
193    assert snapshot.track_recent(_track("libZ", "library", mapping=stream_mapping), 3600) is True
194    # a track that matches neither key is not recent
195    other_mapping = ProviderMapping(
196        item_id="other", provider_domain="spotify", provider_instance="spotify--x"
197    )
198    assert snapshot.track_recent(_track("nope", "library", mapping=other_mapping), 3600) is False
199
200
201async def test_track_recent_respects_window(mass: MusicAssistant) -> None:
202    """A play older than the window is not considered recent."""
203    now = int(time.time())
204    await _add_playlog_row(
205        mass,
206        item_id="t1",
207        provider="library",
208        media_type=MediaType.TRACK,
209        name="T1",
210        timestamp=now - 7200,
211    )
212    snapshot = await mass.music.recency.snapshot(
213        RecencyWindows(song_seconds=10800), userid="user-a"
214    )
215
216    assert snapshot.track_recent(_track("t1", "library"), 10800) is True  # within 3h
217    assert snapshot.track_recent(_track("t1", "library"), 3600) is False  # but not within 1h
218
219
220async def test_snapshot_disabled_windows_skips_query(mass: MusicAssistant) -> None:
221    """With every window disabled the snapshot is empty even when plays exist."""
222    now = int(time.time())
223    await _add_playlog_row(
224        mass,
225        item_id="t1",
226        provider="library",
227        media_type=MediaType.TRACK,
228        name="T1",
229        timestamp=now - 60,
230    )
231
232    snapshot = await mass.music.recency.snapshot(RecencyWindows(), userid="user-a")
233
234    assert not snapshot.song_ts
235    assert not snapshot.artist_ts
236
237
238async def test_song_lookback_uses_duplicate_gap_when_song_window_unset(
239    mass: MusicAssistant,
240) -> None:
241    """When only the duplicate gap is set, track rows are still read back to that gap."""
242    now = int(time.time())
243    await _add_playlog_row(
244        mass,
245        item_id="recent",
246        provider="library",
247        media_type=MediaType.TRACK,
248        name="R",
249        timestamp=now - 1800,
250    )
251    await _add_playlog_row(
252        mass,
253        item_id="old",
254        provider="library",
255        media_type=MediaType.TRACK,
256        name="O",
257        timestamp=now - 7200,
258    )
259
260    snapshot = await mass.music.recency.snapshot(
261        RecencyWindows(song_seconds=None, duplicate_gap_seconds=3600), userid="user-a"
262    )
263
264    assert ("library", "recent") in snapshot.song_ts
265    assert ("library", "old") not in snapshot.song_ts
266
267
268async def test_track_recent_fuzzy_matches_other_version(mass: MusicAssistant) -> None:
269    """A different release/version of a played song (same title + shared artist) is recent too."""
270    now = int(time.time())
271    await _add_playlog_row(
272        mass,
273        item_id="amber-1",
274        provider="spotify--x",
275        media_type=MediaType.TRACK,
276        name="Amber",
277        timestamp=now - 60,
278        artist_names=["The Thrillseekers"],
279    )
280    snapshot = await mass.music.recency.snapshot(RecencyWindows(song_seconds=3600), userid="user-a")
281
282    # same song, different catalog item with differing artist credits: matched fuzzily
283    other_version = _named_track("amber-2", "Amber", ["The Thrillseekers", "Hydra"])
284    assert snapshot.track_recent(other_version, 3600) is True
285    # a bracketed version suffix on the candidate title is stripped before matching
286    remaster = _named_track("amber-3", "Amber (Remastered 2019)", ["The Thrillseekers"])
287    assert snapshot.track_recent(remaster, 3600) is True
288    # same title by an unrelated artist is NOT matched
289    unrelated = _named_track("amber-4", "Amber", ["Someone Else"])
290    assert snapshot.track_recent(unrelated, 3600) is False
291
292
293async def test_track_recent_no_fuzzy_match_for_legacy_rows(mass: MusicAssistant) -> None:
294    """Playlog rows without stored artists (legacy) only match on exact ids."""
295    now = int(time.time())
296    await _add_playlog_row(
297        mass,
298        item_id="amber-1",
299        provider="spotify--x",
300        media_type=MediaType.TRACK,
301        name="Amber",
302        timestamp=now - 60,
303    )
304    snapshot = await mass.music.recency.snapshot(RecencyWindows(song_seconds=3600), userid="user-a")
305
306    other_version = _named_track("amber-2", "Amber", ["The Thrillseekers"])
307    assert snapshot.track_recent(other_version, 3600) is False
308