/
/
/
1"""Tests for the playlist_metadata metadata provider."""
2
3from __future__ import annotations
4
5import os
6from collections.abc import AsyncGenerator
7from pathlib import Path
8from typing import Any
9from unittest.mock import AsyncMock, MagicMock, patch
10
11import pytest
12from music_assistant_models.enums import ImageType
13from music_assistant_models.media_items import MediaItemImage, Playlist, ProviderMapping, Track
14from music_assistant_models.media_items.metadata import MediaItemMetadata
15from music_assistant_models.unique_list import UniqueList
16
17from music_assistant.constants import CONF_LOG_LEVEL
18from music_assistant.providers.playlist_metadata import PlaylistMetadataProvider
19
20
21def _make_provider(tmp_path: Any) -> PlaylistMetadataProvider:
22 """Construct a PlaylistMetadataProvider with mocked MA infrastructure."""
23 mass = MagicMock()
24 mass.storage_path = str(tmp_path)
25 mass.cache_path = str(tmp_path / "cache")
26 mass.music.get_library_item = AsyncMock()
27 mass.music.get_library_item_by_prov_mappings = AsyncMock()
28
29 manifest = MagicMock()
30 manifest.domain = "playlist_metadata"
31
32 config = MagicMock()
33 config.instance_id = "playlist_metadata"
34 config.get_value = MagicMock(
35 side_effect=lambda key: {
36 CONF_LOG_LEVEL: "GLOBAL",
37 "template": "album_grid",
38 "skip_provider_playlists": False,
39 }.get(key, "album_grid")
40 )
41
42 provider = PlaylistMetadataProvider(mass, manifest, config, set())
43 provider._images_dir = str(tmp_path / "playlist_images")
44 Path(provider._images_dir).mkdir(parents=True, exist_ok=True)
45
46 return provider
47
48
49def _make_playlist() -> Playlist:
50 """Create a test playlist with tracks."""
51 return Playlist(
52 item_id="test_playlist_1",
53 provider="test_provider",
54 name="Test Playlist",
55 provider_mappings={
56 ProviderMapping(
57 item_id="test_playlist_1",
58 provider_domain="test_provider",
59 provider_instance="test",
60 )
61 },
62 metadata=MediaItemMetadata(),
63 )
64
65
66def _make_track_with_image(track_id: str, image_url: str) -> Track:
67 """Create a test track with an album image."""
68 return Track(
69 item_id=track_id,
70 provider="test_provider",
71 name=f"Track {track_id}",
72 provider_mappings={
73 ProviderMapping(
74 item_id=track_id,
75 provider_domain="test_provider",
76 provider_instance="test",
77 )
78 },
79 metadata=MediaItemMetadata(
80 images=UniqueList(
81 [
82 MediaItemImage(
83 type=ImageType.THUMB,
84 path=image_url,
85 provider="test_provider",
86 remotely_accessible=True,
87 )
88 ]
89 )
90 ),
91 )
92
93
94def _make_track_with_genres(track_id: str, genres: set[str]) -> Track:
95 """Create a test track with genres."""
96 return Track(
97 item_id=track_id,
98 provider="test_provider",
99 name=f"Track {track_id}",
100 provider_mappings={
101 ProviderMapping(
102 item_id=track_id,
103 provider_domain="test_provider",
104 provider_instance="test",
105 )
106 },
107 metadata=MediaItemMetadata(genres=genres),
108 )
109
110
111@pytest.mark.asyncio
112async def test_get_playlist_metadata_returns_none_when_insufficient_images(
113 tmp_path: Any,
114) -> None:
115 """Provider should return None when playlist has insufficient unique images."""
116 provider = _make_provider(tmp_path)
117 await provider.handle_async_init()
118
119 playlist = _make_playlist()
120
121 # Mock empty playlist (no tracks = no images)
122 async def mock_tracks_iter(
123 _item_id: str,
124 _provider: str,
125 _force_refresh: bool = False,
126 _allow_dynamic_tracks: bool = False,
127 ) -> AsyncGenerator[Track]:
128 if False: # pragma: no cover
129 yield # type: ignore[unreachable]
130
131 with patch.object(provider.mass.music.playlists, "tracks", side_effect=mock_tracks_iter):
132 result = await provider.get_playlist_metadata(playlist)
133
134 assert result is None
135
136
137@pytest.mark.asyncio
138async def test_get_playlist_metadata_returns_metadata_when_sufficient_images(
139 tmp_path: Any,
140) -> None:
141 """Provider should return MediaItemMetadata with images when there are sufficient images."""
142 provider = _make_provider(tmp_path)
143 await provider.handle_async_init()
144
145 playlist = _make_playlist()
146
147 # Mock getting tracks with multiple unique images
148 tracks = [
149 _make_track_with_image(f"track{i}", f"http://example.com/img{i}.jpg") for i in range(10)
150 ]
151
152 async def mock_tracks_iter(
153 _item_id: str,
154 _provider: str,
155 _force_refresh: bool = False,
156 _allow_dynamic_tracks: bool = False,
157 ) -> AsyncGenerator[Track]:
158 for track in tracks:
159 yield track
160
161 # Mock _render to return fake image data
162 with (
163 patch.object(provider.mass.music.playlists, "tracks", side_effect=mock_tracks_iter),
164 patch.object(provider, "_render", new_callable=AsyncMock) as mock_render,
165 ):
166 mock_render.return_value = b"fake_image_data"
167
168 result = await provider.get_playlist_metadata(playlist)
169
170 assert result is not None
171 assert isinstance(result, MediaItemMetadata)
172 assert result.images is not None
173 assert len(result.images) == 2 # Both THUMB and FANART
174
175 # Check THUMB image
176 thumb_image = next((img for img in result.images if img.type == ImageType.THUMB), None)
177 assert thumb_image is not None
178 assert thumb_image.provider == "playlist_metadata"
179 assert Path(thumb_image.path).exists()
180 assert "_thumb.jpg" in thumb_image.path
181
182 # Check FANART image
183 fanart_image = next((img for img in result.images if img.type == ImageType.FANART), None)
184 assert fanart_image is not None
185 assert fanart_image.provider == "playlist_metadata"
186 assert Path(fanart_image.path).exists()
187 assert "_fanart.jpg" in fanart_image.path
188
189
190@pytest.mark.asyncio
191async def test_get_playlist_metadata_handles_exception(
192 tmp_path: Any,
193) -> None:
194 """Provider should return None and log when rendering fails."""
195 provider = _make_provider(tmp_path)
196 await provider.handle_async_init()
197
198 playlist = _make_playlist()
199
200 # Mock _render to raise an exception
201 with patch.object(provider, "_render", new_callable=AsyncMock) as mock_render:
202 mock_render.side_effect = RuntimeError("Rendering failed")
203
204 result = await provider.get_playlist_metadata(playlist)
205
206 assert result is None
207
208
209@pytest.mark.asyncio
210async def test_get_playlist_metadata_cleans_up_old_files(
211 tmp_path: Any,
212) -> None:
213 """Provider should clean up old artwork files when regenerating."""
214 provider = _make_provider(tmp_path)
215 await provider.handle_async_init()
216
217 playlist = _make_playlist()
218
219 # Create existing artwork files (simulating previous generations)
220 old_file1 = os.path.join(provider._images_dir, f"{playlist.item_id}_1234567890_thumb.jpg")
221 old_file2 = os.path.join(provider._images_dir, f"{playlist.item_id}_9876543210_thumb.jpg")
222 with open(old_file1, "wb") as f: # noqa: ASYNC230
223 f.write(b"old_image_data_1")
224 with open(old_file2, "wb") as f: # noqa: ASYNC230
225 f.write(b"old_image_data_2")
226
227 tracks = [
228 _make_track_with_image(f"track{i}", f"http://example.com/img{i}.jpg") for i in range(10)
229 ]
230
231 async def mock_tracks_iter(
232 _item_id: str,
233 _provider: str,
234 _force_refresh: bool = False,
235 _allow_dynamic_tracks: bool = False,
236 ) -> AsyncGenerator[Track]:
237 for track in tracks:
238 yield track
239
240 with (
241 patch.object(provider.mass.music.playlists, "tracks", side_effect=mock_tracks_iter),
242 patch.object(provider, "_render", new_callable=AsyncMock) as mock_render,
243 ):
244 mock_render.return_value = b"new_image_data"
245
246 result = await provider.get_playlist_metadata(playlist)
247
248 assert result is not None
249 assert result.images is not None
250 thumb_image = next((img for img in result.images if img.type == ImageType.THUMB), None)
251 assert thumb_image is not None
252 # New file should have timestamp in filename
253 assert playlist.item_id in thumb_image.path
254 assert "_thumb.jpg" in thumb_image.path
255 # Old files are cleaned up asynchronously by _cleanup_stale_images, not inline here.
256 assert Path(thumb_image.path).exists()
257
258
259@pytest.mark.asyncio
260async def test_get_playlist_metadata_always_creates_unique_filename(
261 tmp_path: Any,
262) -> None:
263 """Provider should always create unique filename with timestamp."""
264 provider = _make_provider(tmp_path)
265 await provider.handle_async_init()
266
267 playlist = _make_playlist()
268
269 tracks = [
270 _make_track_with_image(f"track{i}", f"http://example.com/img{i}.jpg") for i in range(10)
271 ]
272
273 async def mock_tracks_iter(
274 _item_id: str,
275 _provider: str,
276 _force_refresh: bool = False,
277 _allow_dynamic_tracks: bool = False,
278 ) -> AsyncGenerator[Track]:
279 for track in tracks:
280 yield track
281
282 with (
283 patch.object(provider.mass.music.playlists, "tracks", side_effect=mock_tracks_iter),
284 patch.object(provider, "_render", new_callable=AsyncMock) as mock_render,
285 ):
286 mock_render.return_value = b"new_image_data"
287
288 result = await provider.get_playlist_metadata(playlist)
289
290 assert result is not None
291 assert result.images is not None
292 thumb_image = next((img for img in result.images if img.type == ImageType.THUMB), None)
293 assert thumb_image is not None
294 # Filename should include timestamp
295 assert playlist.item_id in thumb_image.path
296 assert "_thumb.jpg" in thumb_image.path
297 # Should have timestamp in filename (longer than base name)
298 assert len(Path(thumb_image.path).name) > len(f"{playlist.item_id}_thumb.jpg")
299
300
301@pytest.mark.asyncio
302async def test_is_our_image_recognizes_own_images(
303 tmp_path: Any,
304) -> None:
305 """_is_our_image should correctly identify images generated by this provider."""
306 provider = _make_provider(tmp_path)
307 await provider.handle_async_init()
308
309 # Own image with correct provider
310 own_image = MediaItemImage(
311 type=ImageType.THUMB,
312 path=os.path.join(provider._images_dir, "test_thumb.jpg"),
313 provider="playlist_metadata",
314 remotely_accessible=False,
315 )
316 assert provider._is_our_image(own_image) is True
317
318 # Image in our directory but provider changed to "builtin"
319 builtin_image = MediaItemImage(
320 type=ImageType.THUMB,
321 path=os.path.join(provider._images_dir, "test_thumb.jpg"),
322 provider="builtin",
323 remotely_accessible=False,
324 )
325 assert provider._is_our_image(builtin_image) is True
326
327 # Remote URL with different provider
328 remote_image = MediaItemImage(
329 type=ImageType.THUMB,
330 path="http://example.com/image.jpg",
331 provider="other_provider",
332 remotely_accessible=True,
333 )
334 assert provider._is_our_image(remote_image) is False
335
336 # Bare filename (builtin asset)
337 builtin_asset = MediaItemImage(
338 type=ImageType.THUMB,
339 path="logo.png",
340 provider="builtin",
341 remotely_accessible=False,
342 )
343 assert provider._is_our_image(builtin_asset) is False
344
345
346@pytest.mark.asyncio
347async def test_analyze_playlist_genres_returns_most_common_genres(
348 tmp_path: Any,
349) -> None:
350 """Provider should return the most common genres from playlist tracks."""
351 provider = _make_provider(tmp_path)
352 await provider.handle_async_init()
353
354 # Override config to enable genre detection
355 provider.config.get_value = MagicMock( # type: ignore[method-assign]
356 side_effect=lambda key: {
357 CONF_LOG_LEVEL: "GLOBAL",
358 "template": "album_grid",
359 "skip_provider_playlists": False,
360 "enable_genre_detection": True,
361 "genre_min_threshold": 10,
362 "genre_max_count": 3,
363 }.get(key)
364 )
365
366 playlist = _make_playlist()
367
368 # Create tracks with genres:
369 # Rock: 15/26 = 57.7%
370 # Pop: 5/26 = 19.2%
371 # Jazz: 3/26 = 11.5% (above threshold)
372 # Electronic: 2/26 = 7.7% (below 10% threshold)
373 # Classical: 1/26 = 3.8% (below threshold)
374 tracks = (
375 [_make_track_with_genres(f"track{i}", {"Rock"}) for i in range(15)]
376 + [_make_track_with_genres(f"track{i}", {"Pop"}) for i in range(15, 20)]
377 + [_make_track_with_genres(f"track{i}", {"Jazz"}) for i in range(20, 23)]
378 + [_make_track_with_genres(f"track{i}", {"Electronic"}) for i in range(23, 25)]
379 + [_make_track_with_genres(f"track{i}", {"Classical"}) for i in range(25, 26)]
380 )
381
382 async def mock_tracks_iter(
383 _item_id: str,
384 _provider: str,
385 _force_refresh: bool = False,
386 _allow_dynamic_tracks: bool = False,
387 ) -> AsyncGenerator[Track]:
388 for track in tracks:
389 yield track
390
391 with patch.object(provider.mass.music.playlists, "tracks", side_effect=mock_tracks_iter):
392 result = await provider._analyze_playlist_genres(playlist)
393
394 assert result is not None
395 # Should return top 3 genres above 10% threshold
396 assert result == {"Rock", "Pop", "Jazz"}
397
398
399@pytest.mark.asyncio
400async def test_analyze_playlist_genres_respects_threshold(
401 tmp_path: Any,
402) -> None:
403 """Provider should filter out genres below the minimum threshold."""
404 provider = _make_provider(tmp_path)
405 await provider.handle_async_init()
406
407 # Override config with higher threshold
408 provider.config.get_value = MagicMock( # type: ignore[method-assign]
409 side_effect=lambda key: {
410 CONF_LOG_LEVEL: "GLOBAL",
411 "enable_genre_detection": True,
412 "genre_min_threshold": 30, # 30% threshold
413 "genre_max_count": 5,
414 }.get(key)
415 )
416
417 playlist = _make_playlist()
418
419 # Rock: 10/20 = 50% (above threshold)
420 # Pop: 5/20 = 25% (below threshold)
421 tracks = (
422 [_make_track_with_genres(f"track{i}", {"Rock"}) for i in range(10)]
423 + [_make_track_with_genres(f"track{i}", {"Pop"}) for i in range(10, 15)]
424 + [_make_track_with_genres(f"track{i}", set()) for i in range(15, 20)]
425 )
426
427 async def mock_tracks_iter(
428 _item_id: str,
429 _provider: str,
430 _force_refresh: bool = False,
431 _allow_dynamic_tracks: bool = False,
432 ) -> AsyncGenerator[Track]:
433 for track in tracks:
434 yield track
435
436 with patch.object(provider.mass.music.playlists, "tracks", side_effect=mock_tracks_iter):
437 result = await provider._analyze_playlist_genres(playlist)
438
439 assert result is not None
440 assert result == {"Rock"} # Only Rock meets the 30% threshold
441
442
443@pytest.mark.asyncio
444async def test_analyze_playlist_genres_returns_none_for_empty_playlist(
445 tmp_path: Any,
446) -> None:
447 """Provider should return None when playlist has no tracks."""
448 provider = _make_provider(tmp_path)
449 await provider.handle_async_init()
450
451 provider.config.get_value = MagicMock( # type: ignore[method-assign]
452 side_effect=lambda key: {
453 "enable_genre_detection": True,
454 "genre_min_threshold": 10,
455 "genre_max_count": 3,
456 }.get(key)
457 )
458
459 playlist = _make_playlist()
460
461 async def mock_tracks_iter(
462 _item_id: str,
463 _provider: str,
464 _force_refresh: bool = False,
465 _allow_dynamic_tracks: bool = False,
466 ) -> AsyncGenerator[Track]:
467 if False: # pragma: no cover
468 yield # type: ignore[unreachable]
469
470 with patch.object(provider.mass.music.playlists, "tracks", side_effect=mock_tracks_iter):
471 result = await provider._analyze_playlist_genres(playlist)
472
473 assert result is None
474
475
476@pytest.mark.asyncio
477async def test_analyze_playlist_genres_returns_none_when_no_genres_meet_threshold(
478 tmp_path: Any,
479) -> None:
480 """Provider should return None when no genres meet the minimum threshold."""
481 provider = _make_provider(tmp_path)
482 await provider.handle_async_init()
483
484 provider.config.get_value = MagicMock( # type: ignore[method-assign]
485 side_effect=lambda key: {
486 "enable_genre_detection": True,
487 "genre_min_threshold": 50, # 50% threshold
488 "genre_max_count": 3,
489 }.get(key)
490 )
491
492 playlist = _make_playlist()
493
494 # All genres below 50%
495 tracks = (
496 [_make_track_with_genres(f"track{i}", {"Rock"}) for i in range(4)]
497 + [_make_track_with_genres(f"track{i}", {"Pop"}) for i in range(4, 7)]
498 + [_make_track_with_genres(f"track{i}", set()) for i in range(7, 10)]
499 )
500
501 async def mock_tracks_iter(
502 _item_id: str,
503 _provider: str,
504 _force_refresh: bool = False,
505 _allow_dynamic_tracks: bool = False,
506 ) -> AsyncGenerator[Track]:
507 for track in tracks:
508 yield track
509
510 with patch.object(provider.mass.music.playlists, "tracks", side_effect=mock_tracks_iter):
511 result = await provider._analyze_playlist_genres(playlist)
512
513 assert result is None
514
515
516@pytest.mark.asyncio
517async def test_analyze_playlist_genres_handles_exception(
518 tmp_path: Any,
519) -> None:
520 """Provider should handle exceptions gracefully and return None."""
521 provider = _make_provider(tmp_path)
522 await provider.handle_async_init()
523
524 provider.config.get_value = MagicMock( # type: ignore[method-assign]
525 side_effect=lambda key: {
526 "enable_genre_detection": True,
527 "genre_min_threshold": 10,
528 "genre_max_count": 3,
529 }.get(key)
530 )
531
532 playlist = _make_playlist()
533
534 # Mock tracks to raise exception
535 async def mock_tracks_iter(
536 _item_id: str,
537 _provider: str,
538 _force_refresh: bool = False,
539 _allow_dynamic_tracks: bool = False,
540 ) -> AsyncGenerator[Track]:
541 if False: # pragma: no cover
542 yield # type: ignore[unreachable]
543 raise AttributeError("Failed to get tracks")
544
545 with patch.object(provider.mass.music.playlists, "tracks", side_effect=mock_tracks_iter):
546 result = await provider._analyze_playlist_genres(playlist)
547
548 assert result is None
549
550
551@pytest.mark.asyncio
552async def test_get_playlist_metadata_includes_genres_when_enabled(
553 tmp_path: Any,
554) -> None:
555 """Provider should include genres in metadata when genre detection is enabled."""
556 provider = _make_provider(tmp_path)
557 await provider.handle_async_init()
558
559 provider.config.get_value = MagicMock( # type: ignore[method-assign]
560 side_effect=lambda key: {
561 CONF_LOG_LEVEL: "GLOBAL",
562 "template": "album_grid",
563 "skip_provider_playlists": False,
564 "enable_genre_detection": True,
565 "genre_min_threshold": 10,
566 "genre_max_count": 3,
567 }.get(key)
568 )
569
570 playlist = _make_playlist()
571
572 tracks_with_images = [
573 _make_track_with_image(f"track{i}", f"http://example.com/img{i}.jpg") for i in range(10)
574 ]
575
576 # Also add genres to these tracks
577 for i, track in enumerate(tracks_with_images):
578 if i < 5:
579 track.metadata.genres = {"Rock"}
580 else:
581 track.metadata.genres = {"Pop"}
582
583 async def mock_tracks_iter(
584 _item_id: str,
585 _provider: str,
586 _force_refresh: bool = False,
587 _allow_dynamic_tracks: bool = False,
588 ) -> AsyncGenerator[Track]:
589 for track in tracks_with_images:
590 yield track
591
592 with (
593 patch.object(provider.mass.music.playlists, "tracks", side_effect=mock_tracks_iter),
594 patch.object(provider, "_render", new_callable=AsyncMock) as mock_render,
595 ):
596 mock_render.return_value = b"fake_image_data"
597
598 result = await provider.get_playlist_metadata(playlist)
599
600 assert result is not None
601 assert result.images is not None
602 assert len(result.images) == 2 # THUMB and FANART
603
604 # Check genres are included
605 assert result.genres is not None
606 assert result.genres == {"Rock", "Pop"}
607
608
609@pytest.mark.asyncio
610async def test_get_playlist_metadata_excludes_genres_when_disabled(
611 tmp_path: Any,
612) -> None:
613 """Provider should not include genres when genre detection is disabled (default)."""
614 provider = _make_provider(tmp_path)
615 await provider.handle_async_init()
616
617 # Default config has genre detection disabled
618 provider.config.get_value = MagicMock( # type: ignore[method-assign]
619 side_effect=lambda key: {
620 CONF_LOG_LEVEL: "GLOBAL",
621 "template": "album_grid",
622 "skip_provider_playlists": False,
623 "enable_genre_detection": False, # Disabled
624 "genre_min_threshold": 10,
625 "genre_max_count": 3,
626 }.get(key)
627 )
628
629 playlist = _make_playlist()
630
631 tracks = [
632 _make_track_with_image(f"track{i}", f"http://example.com/img{i}.jpg") for i in range(10)
633 ]
634
635 # Add genres to tracks (should be ignored)
636 for track in tracks:
637 track.metadata.genres = {"Rock", "Pop"}
638
639 async def mock_tracks_iter(
640 _item_id: str,
641 _provider: str,
642 _force_refresh: bool = False,
643 _allow_dynamic_tracks: bool = False,
644 ) -> AsyncGenerator[Track]:
645 for track in tracks:
646 yield track
647
648 with (
649 patch.object(provider.mass.music.playlists, "tracks", side_effect=mock_tracks_iter),
650 patch.object(provider, "_render", new_callable=AsyncMock) as mock_render,
651 ):
652 mock_render.return_value = b"fake_image_data"
653
654 result = await provider.get_playlist_metadata(playlist)
655
656 assert result is not None
657 assert result.images is not None
658 # Genres should not be included
659 assert result.genres is None
660