/
/
/
1"""Tests for playlist import track matching and scoring logic."""
2
3from __future__ import annotations
4
5from typing import Any, cast
6from unittest.mock import AsyncMock, MagicMock
7
8from music_assistant_models.enums import ExternalID, ImageType, MediaType
9from music_assistant_models.errors import MediaNotFoundError
10from music_assistant_models.media_items import (
11 Artist,
12 ItemMapping,
13 MediaItemImage,
14 MediaItemMetadata,
15 Playlist,
16 ProviderMapping,
17 Track,
18 UniqueList,
19)
20
21from music_assistant.helpers.playlists import (
22 PlaylistItem,
23 ProviderMappingInfo,
24 generate_m3u,
25)
26from music_assistant.providers.builtin import BuiltinProvider
27
28
29def _make_provider() -> BuiltinProvider:
30 """Create a minimal BuiltinProvider with mocked mass."""
31 mass = MagicMock()
32 prov = object.__new__(BuiltinProvider)
33 prov.mass = mass
34 prov.logger = MagicMock()
35 return prov
36
37
38def _make_track(
39 name: str,
40 artists: list[str] | None = None,
41 duration: int = 0,
42 album_name: str | None = None,
43 version: str = "",
44 isrc: str | None = None,
45 mbid: str | None = None,
46 media_type: MediaType = MediaType.TRACK,
47) -> Track:
48 """Build a Track for matching tests."""
49 artist_list: UniqueList[Artist | ItemMapping] = UniqueList()
50 for a in artists or []:
51 artist_list.append(
52 ItemMapping(item_id=a, provider="test", name=a, media_type=MediaType.ARTIST)
53 )
54 external_ids: set[tuple[ExternalID, str]] = set()
55 if isrc:
56 external_ids.add((ExternalID.ISRC, isrc))
57 if mbid:
58 external_ids.add((ExternalID.MB_RECORDING, mbid))
59 album_mapping = None
60 if album_name:
61 album_mapping = ItemMapping(
62 item_id=album_name, provider="test", name=album_name, media_type=MediaType.ALBUM
63 )
64 track = Track(
65 item_id="test123",
66 provider="test",
67 name=name,
68 version=version,
69 duration=duration,
70 artists=artist_list,
71 album=album_mapping,
72 external_ids=external_ids,
73 provider_mappings={
74 ProviderMapping(
75 item_id="test123",
76 provider_domain="test",
77 provider_instance="test",
78 )
79 },
80 )
81 track.media_type = media_type
82 return track
83
84
85def _make_playlist_item(
86 title: str | None = None,
87 length: str | None = None,
88 metadata: dict[str, str] | None = None,
89 providers: list[ProviderMappingInfo] | None = None,
90) -> PlaylistItem:
91 """Build a PlaylistItem for matching tests."""
92 return PlaylistItem(
93 path="spotify://track/original",
94 title=title,
95 length=length,
96 metadata=metadata,
97 providers=providers or [],
98 )
99
100
101def _make_playlist(name: str, image_url: str | None = None) -> Playlist:
102 """Build a Playlist for builtin provider tests."""
103 metadata = MediaItemMetadata()
104 if image_url:
105 metadata.images = UniqueList(
106 [
107 MediaItemImage(
108 type=ImageType.THUMB,
109 path=image_url,
110 provider="builtin",
111 remotely_accessible=True,
112 )
113 ]
114 )
115 return Playlist(
116 item_id="playlist_1",
117 provider="builtin",
118 name=name,
119 metadata=metadata,
120 provider_mappings={
121 ProviderMapping(
122 item_id="playlist_1",
123 provider_domain="builtin",
124 provider_instance="builtin",
125 )
126 },
127 )
128
129
130# --------------------------------------------------------------------------- #
131# _score_track_match #
132# --------------------------------------------------------------------------- #
133
134
135class TestScoreTrackMatch:
136 """Tests for the _score_track_match scoring method."""
137
138 def setup_method(self) -> None:
139 """Set up test fixtures."""
140 self.prov = _make_provider()
141
142 def test_isrc_match_returns_max_score(self) -> None:
143 """ISRC match should return 10 (maximum score)."""
144 candidate = _make_track("Song", artists=["Artist"], isrc="USRC17607839")
145 item = _make_playlist_item(
146 title="Artist - Song",
147 metadata={"isrc": "USRC17607839"},
148 )
149 assert self.prov._score_track_match(candidate, item) == 10
150
151 def test_isrc_match_case_insensitive(self) -> None:
152 """ISRC matching should be case-insensitive."""
153 candidate = _make_track("Song", artists=["Artist"], isrc="usrc17607839")
154 item = _make_playlist_item(
155 title="Artist - Song",
156 metadata={"isrc": "USRC17607839"},
157 )
158 assert self.prov._score_track_match(candidate, item) == 10
159
160 def test_mbid_match_returns_max_score(self) -> None:
161 """MusicBrainz recording ID match should return 10."""
162 candidate = _make_track(
163 "Song", artists=["Artist"], mbid="a1b2c3d4-e5f6-7890-abcd-ef1234567890"
164 )
165 item = _make_playlist_item(
166 title="Artist - Song",
167 metadata={"mbid": "A1B2C3D4-E5F6-7890-ABCD-EF1234567890"},
168 )
169 assert self.prov._score_track_match(candidate, item) == 10
170
171 def test_title_match_only(self) -> None:
172 """Title-only match (no artist) should score 1."""
173 candidate = _make_track("Song Title")
174 item = _make_playlist_item(title="Song Title")
175 assert self.prov._score_track_match(candidate, item) == 1
176
177 def test_title_mismatch_returns_zero(self) -> None:
178 """Non-matching title should return 0."""
179 candidate = _make_track("Completely Different")
180 item = _make_playlist_item(title="Song Title")
181 assert self.prov._score_track_match(candidate, item) == 0
182
183 def test_title_and_artist_match(self) -> None:
184 """Title + artist match should score 3 (1 title + 2 artist)."""
185 candidate = _make_track("Song", artists=["Radiohead"])
186 item = _make_playlist_item(title="Radiohead - Song")
187 assert self.prov._score_track_match(candidate, item) == 3
188
189 def test_artist_mismatch_returns_zero(self) -> None:
190 """When artist is provided but doesn't match, return 0."""
191 candidate = _make_track("Song", artists=["Coldplay"])
192 item = _make_playlist_item(title="Radiohead - Song")
193 assert self.prov._score_track_match(candidate, item) == 0
194
195 def test_album_bonus(self) -> None:
196 """Matching album adds 1 to the score."""
197 candidate = _make_track("Song", artists=["Artist"], album_name="OK Computer")
198 item = _make_playlist_item(
199 title="Artist - Song",
200 metadata={"album": "OK Computer"},
201 )
202 # 1 (title) + 2 (artist) + 1 (album) = 4
203 assert self.prov._score_track_match(candidate, item) == 4
204
205 def test_duration_near_exact_bonus(self) -> None:
206 """Duration within 2 seconds adds 2 to the score."""
207 candidate = _make_track("Song", artists=["Artist"], duration=241)
208 item = _make_playlist_item(title="Artist - Song", length="240")
209 # 1 (title) + 2 (artist) + 2 (duration) = 5
210 assert self.prov._score_track_match(candidate, item) == 5
211
212 def test_duration_close_bonus(self) -> None:
213 """Duration within 5 seconds adds 1 to the score."""
214 candidate = _make_track("Song", artists=["Artist"], duration=245)
215 item = _make_playlist_item(title="Artist - Song", length="240")
216 # 1 (title) + 2 (artist) + 1 (duration close) = 4
217 assert self.prov._score_track_match(candidate, item) == 4
218
219 def test_duration_too_far_no_bonus(self) -> None:
220 """Duration beyond 5 seconds gets no bonus."""
221 candidate = _make_track("Song", artists=["Artist"], duration=260)
222 item = _make_playlist_item(title="Artist - Song", length="240")
223 # 1 (title) + 2 (artist) = 3
224 assert self.prov._score_track_match(candidate, item) == 3
225
226 def test_version_match_bonus(self) -> None:
227 """Matching version adds 1."""
228 candidate = _make_track("Song", artists=["Artist"], version="Remastered")
229 item = _make_playlist_item(
230 title="Artist - Song",
231 metadata={"version": "Remastered"},
232 )
233 # 1 (title) + 2 (artist) + 1 (version) = 4
234 assert self.prov._score_track_match(candidate, item) == 4
235
236 def test_version_mismatch_penalty(self) -> None:
237 """Mismatched version subtracts 1."""
238 candidate = _make_track("Song", artists=["Artist"], version="Live")
239 item = _make_playlist_item(
240 title="Artist - Song",
241 metadata={"version": "Remastered"},
242 )
243 # 1 (title) + 2 (artist) - 1 (version mismatch) = 2
244 assert self.prov._score_track_match(candidate, item) == 2
245
246 def test_media_type_gate(self) -> None:
247 """Mismatched media type should return 0."""
248 candidate = _make_track("Song", artists=["Artist"])
249 item = _make_playlist_item(
250 title="Artist - Song",
251 metadata={"media_type": "podcast_episode"},
252 )
253 assert self.prov._score_track_match(candidate, item) == 0
254
255 def test_none_title_returns_zero(self) -> None:
256 """PlaylistItem with no title should score 0."""
257 candidate = _make_track("Song")
258 item = _make_playlist_item(title=None)
259 assert self.prov._score_track_match(candidate, item) == 0
260
261 def test_full_metadata_high_score(self) -> None:
262 """Track with all metadata matching should get a high score."""
263 candidate = _make_track(
264 "Everything In Its Right Place",
265 artists=["Radiohead"],
266 duration=240,
267 album_name="Kid A",
268 version="Remastered",
269 )
270 item = _make_playlist_item(
271 title="Radiohead - Everything In Its Right Place",
272 length="240",
273 metadata={
274 "media_type": "track",
275 "album": "Kid A",
276 "version": "Remastered",
277 },
278 )
279 # 1 (title) + 2 (artist) + 1 (album) + 1 (version) + 2 (duration exact) = 7
280 assert self.prov._score_track_match(candidate, item) == 7
281
282
283async def test_import_playlist_preserves_playlist_image() -> None:
284 """Test that importing an M3U keeps the playlist-level image."""
285 prov = _make_provider()
286 prov_any = cast("Any", prov)
287 prov_any.create_playlist = AsyncMock(return_value=_make_playlist("Imported Playlist"))
288 prov_any.get_playlist = AsyncMock(
289 return_value=_make_playlist("Imported Playlist", "https://img.example.com/cover.jpg")
290 )
291 prov_any._write_m3u_file = AsyncMock()
292
293 m3u_data = generate_m3u(
294 "Imported Playlist",
295 [PlaylistItem(path="spotify://track/abc123", title="Test", length="120")],
296 "https://img.example.com/cover.jpg",
297 )
298
299 result = await prov.import_playlist(m3u_data)
300
301 assert prov_any._write_m3u_file.await_args is not None
302 args = prov_any._write_m3u_file.await_args.args
303 assert args[0] == "playlist_1"
304 assert args[1] == "Imported Playlist"
305 assert args[3] == "https://img.example.com/cover.jpg"
306 assert result.image is not None
307 assert result.image.path == "https://img.example.com/cover.jpg"
308
309
310async def test_match_imported_tracks_enriches_matched_entries() -> None:
311 """Test that a matched entry is enriched with provider metadata, not just its URI."""
312 prov = _make_provider()
313 prov_any = cast("Any", prov)
314 prov_any._read_m3u_file = AsyncMock(
315 return_value=(
316 "#EXTM3U\n"
317 "#PLAYLIST:Imported\n"
318 "#EXTMA:media_type=track||name=Song||mbid=a1b2c3d4-e5f6-7890-abcd-ef1234567890\n"
319 "#EXTINF:294,Artist - Song\n"
320 "track-1\n"
321 )
322 )
323 matched_uri = "opensubsonic--abc123://track/xyz789"
324 enriched_entry = PlaylistItem(
325 path=matched_uri,
326 title="Artist - Song",
327 length="294",
328 metadata={"media_type": "track", "name": "Song"},
329 providers=[
330 ProviderMappingInfo(
331 domain="opensubsonic", item_id="xyz789", instance_id="opensubsonic--abc123"
332 )
333 ],
334 )
335 prov_any._match_track_by_metadata = AsyncMock(return_value=matched_uri)
336 prov_any._build_m3u_entry_from_uri = AsyncMock(return_value=enriched_entry)
337 prov_any.get_playlist = AsyncMock(return_value=_make_playlist("Imported"))
338 prov_any._write_m3u_file = AsyncMock()
339
340 await prov.match_imported_playlist_tracks("playlist_1")
341
342 prov_any._build_m3u_entry_from_uri.assert_awaited_once_with(matched_uri)
343 assert prov_any._write_m3u_file.await_args is not None
344 written_items = prov_any._write_m3u_file.await_args.args[2]
345 assert written_items == [enriched_entry]
346 assert written_items[0].providers
347
348
349async def test_match_imported_tracks_falls_back_to_uri_when_enrich_fails() -> None:
350 """Test that the matched URI is still stored when enrichment fails."""
351 prov = _make_provider()
352 prov_any = cast("Any", prov)
353 prov_any._read_m3u_file = AsyncMock(
354 return_value="#EXTM3U\n#PLAYLIST:Imported\n#EXTINF:294,Artist - Song\ntrack-1\n"
355 )
356 matched_uri = "opensubsonic--abc123://track/xyz789"
357 prov_any._match_track_by_metadata = AsyncMock(return_value=matched_uri)
358 prov_any._build_m3u_entry_from_uri = AsyncMock(side_effect=MediaNotFoundError("gone"))
359 prov_any.get_playlist = AsyncMock(return_value=_make_playlist("Imported"))
360 prov_any._write_m3u_file = AsyncMock()
361
362 await prov.match_imported_playlist_tracks("playlist_1")
363
364 assert prov_any._write_m3u_file.await_args is not None
365 written_items = prov_any._write_m3u_file.await_args.args[2]
366 assert written_items[0].path == matched_uri
367 assert not written_items[0].providers
368
369
370async def test_remove_playlist_tracks_preserves_playlist_image() -> None:
371 """Test that rewriting a playlist after track removal keeps the playlist image."""
372 prov = _make_provider()
373 prov._playlist_locks = {}
374 prov_any = cast("Any", prov)
375 prov_any._read_m3u_file = AsyncMock(
376 return_value=generate_m3u(
377 "My Playlist",
378 [
379 PlaylistItem(path="spotify://track/one", title="One", length="120"),
380 PlaylistItem(path="spotify://track/two", title="Two", length="180"),
381 ],
382 "https://img.example.com/cover.jpg",
383 )
384 )
385 prov_any.get_playlist = AsyncMock(
386 return_value=_make_playlist("My Playlist", "https://img.example.com/cover.jpg")
387 )
388 prov_any._write_m3u_file = AsyncMock()
389
390 await prov.remove_playlist_tracks("playlist_1", (1,))
391
392 assert prov_any._write_m3u_file.await_args is not None
393 args = prov_any._write_m3u_file.await_args.args
394 assert args[0] == "playlist_1"
395 assert args[1] == "My Playlist"
396 assert len(args[2]) == 1
397 assert args[3] == "https://img.example.com/cover.jpg"
398