/
/
/
1"""Unit tests for Apple Music library track streaming and windowed enrichment."""
2
3from typing import Any, cast
4from unittest.mock import AsyncMock, MagicMock
5
6import pytest
7from music_assistant_models.enums import MediaType
8from music_assistant_models.errors import MusicAssistantError
9from music_assistant_models.media_items import (
10 Album,
11 Artist,
12 ItemMapping,
13 ProviderMapping,
14 Track,
15 UniqueList,
16)
17
18from music_assistant.providers.apple_music.library import (
19 _DETAIL_BATCH_SIZE,
20 _MAX_SEARCH_FALLBACK_PER_WINDOW,
21 _TRACK_SYNC_WINDOW,
22 AppleMusicLibraryManager,
23)
24
25
26def _library_song(idx: int, *, catalog_id: str | None) -> dict[str, Any]:
27 """Build a minimal me/library/songs listing item, optionally catalog-backed."""
28 play_params: dict[str, Any] = {"id": f"i.{idx}"}
29 if catalog_id is not None:
30 play_params["catalogId"] = catalog_id
31 return {
32 "id": f"i.{idx}",
33 "type": "library-songs",
34 "attributes": {"name": f"Track {idx}", "playParams": play_params},
35 }
36
37
38def _library_song_with_metadata(
39 idx: int,
40 *,
41 catalog_id: str | None,
42 artist_name: str | None = None,
43 album_name: str | None = None,
44) -> dict[str, Any]:
45 """Build a library song item with artist/album metadata."""
46 play_params: dict[str, Any] = {"id": f"i.{idx}"}
47 if catalog_id is not None:
48 play_params["catalogId"] = catalog_id
49
50 attributes: dict[str, Any] = {"name": f"Track {idx}", "playParams": play_params}
51 if artist_name:
52 attributes["artistName"] = artist_name
53 if album_name:
54 attributes["albumName"] = album_name
55
56 return {
57 "id": f"i.{idx}",
58 "type": "library-songs",
59 "attributes": attributes,
60 }
61
62
63def _catalog_song(catalog_id: str) -> dict[str, Any]:
64 """Build a minimal catalog/songs response item."""
65 return {
66 "id": catalog_id,
67 "type": "songs",
68 "attributes": {"name": f"Catalog {catalog_id}", "playParams": {"id": catalog_id}},
69 }
70
71
72def _make_test_track(
73 track_id: str,
74 track_name: str,
75 artist_id: str,
76 artist_name: str,
77 album_id: str | None = None,
78 album_name: str | None = None,
79 instance_id: str = "apple_music--test",
80) -> Track:
81 """Build a real Track instance with Artist and optional Album for testing."""
82 artists = UniqueList(
83 [
84 Artist(
85 provider="apple_music",
86 item_id=artist_id,
87 name=artist_name,
88 provider_mappings={
89 ProviderMapping(
90 item_id=artist_id,
91 provider_domain="apple_music",
92 provider_instance=instance_id,
93 )
94 },
95 )
96 ]
97 )
98
99 album = None
100 if album_id and album_name:
101 album = Album(
102 provider="apple_music",
103 item_id=album_id,
104 name=album_name,
105 provider_mappings={
106 ProviderMapping(
107 item_id=album_id,
108 provider_domain="apple_music",
109 provider_instance=instance_id,
110 )
111 },
112 )
113
114 return Track(
115 provider="apple_music",
116 item_id=track_id,
117 name=track_name,
118 artists=cast("UniqueList[Artist | ItemMapping]", artists),
119 album=album,
120 provider_mappings={
121 ProviderMapping(
122 item_id=track_id,
123 provider_domain="apple_music",
124 provider_instance=instance_id,
125 )
126 },
127 )
128
129
130def _make_test_provider() -> MagicMock:
131 """Build a mock provider for search replacement tests."""
132 provider = MagicMock()
133 provider.domain = "apple_music"
134 provider.instance_id = "apple_music--test"
135 return provider
136
137
138def _make_manager(
139 stream_items: list[dict[str, Any]],
140) -> tuple[AppleMusicLibraryManager, MagicMock, dict[str, Any]]:
141 """
142 Build a library manager whose api streams ``stream_items`` and echoes catalog enrichment.
143
144 The returned ``state`` dict tracks how many listing items have been streamed and at what
145 point the first enrichment request fired, so tests can assert streaming/windowing behaviour.
146 """
147 provider = MagicMock()
148 provider.domain = "apple_music"
149 provider.instance_id = "apple_music--test"
150 provider._storefront = "us"
151 api = provider.api_client
152 state: dict[str, Any] = {"streamed": 0, "first_enrich_at": None}
153
154 async def _iter(*_args: Any, **_kwargs: Any) -> Any:
155 for item in stream_items:
156 state["streamed"] += 1
157 yield item
158
159 async def _get_data(_endpoint: str, **kwargs: Any) -> dict[str, Any]:
160 if state["first_enrich_at"] is None:
161 state["first_enrich_at"] = state["streamed"]
162 ids = kwargs["ids"].split(",")
163 assert len(ids) <= _TRACK_SYNC_WINDOW # never exceed the documented catalog batch limit
164 return {"data": [_catalog_song(cid) for cid in ids]}
165
166 api.iter_all_items = _iter
167 api.get_data = AsyncMock(side_effect=_get_data)
168 api.get_ratings = AsyncMock(return_value={})
169 return AppleMusicLibraryManager(provider), api, state
170
171
172@pytest.mark.asyncio
173async def test_catalog_enrichment_is_windowed() -> None:
174 """Catalog enrichment runs in batches capped at the window size, never one giant request."""
175 count = _TRACK_SYNC_WINDOW * 2 + 20
176 items = [_library_song(i, catalog_id=f"c{i}") for i in range(count)]
177 manager, api, _ = _make_manager(items)
178 tracks = [track async for track in manager.get_library_tracks()]
179 assert len(tracks) == count
180 # 320 catalog ids -> ceil(320 / 150) = 3 enrichment requests.
181 assert api.get_data.call_count == 3
182
183
184@pytest.mark.asyncio
185async def test_enriches_before_listing_completes() -> None:
186 """A window is enriched and yielded as soon as it fills, not after the whole listing."""
187 count = _TRACK_SYNC_WINDOW * 2
188 items = [_library_song(i, catalog_id=f"c{i}") for i in range(count)]
189 manager, _, state = _make_manager(items)
190 [track async for track in manager.get_library_tracks()]
191 assert state["first_enrich_at"] == _TRACK_SYNC_WINDOW
192
193
194@pytest.mark.asyncio
195async def test_per_window_search_limit_enforced() -> None:
196 """Only first N deprecated tracks per window trigger search fallback; rest marked unavailable."""
197 # Create library items with missing catalog IDs and full metadata
198 missing_count = _MAX_SEARCH_FALLBACK_PER_WINDOW + 5
199 library_items = [
200 _library_song_with_metadata(
201 i, catalog_id=f"c{i}", artist_name=f"Artist {i}", album_name=f"Album {i}"
202 )
203 for i in range(missing_count)
204 ]
205
206 # Setup provider similar to _make_manager
207 provider = MagicMock()
208 provider.domain = "apple_music"
209 provider.instance_id = "apple_music--test"
210 provider._storefront = "us"
211
212 # Setup API client
213 api = provider.api_client
214
215 async def _iter(*_args: Any, **_kwargs: Any) -> Any:
216 for item in library_items:
217 yield item
218
219 api.iter_all_items = _iter
220 # Return empty data (all catalog IDs "missing")
221 api.get_data = AsyncMock(return_value={"data": []})
222 api.get_ratings = AsyncMock(return_value={})
223
224 # Track search attempts
225 search_call_count = 0
226
227 async def mock_search(*_args: Any, **_kwargs: Any) -> Any:
228 nonlocal search_call_count
229 search_call_count += 1
230 results = MagicMock()
231 results.tracks = []
232 return results
233
234 # Mock media_manager with search method
235 provider.media_manager.search = AsyncMock(side_effect=mock_search)
236
237 manager = AppleMusicLibraryManager(provider)
238 tracks = [track async for track in manager.get_library_tracks()]
239
240 # All tracks should be returned
241 assert len(tracks) == missing_count
242
243 # Only first _MAX_SEARCH_FALLBACK_PER_WINDOW should trigger search
244 assert search_call_count == _MAX_SEARCH_FALLBACK_PER_WINDOW
245
246 # All tracks should be marked unavailable
247 for track in tracks:
248 provider_mapping = next(
249 (m for m in track.provider_mappings if m.provider_instance == provider.instance_id),
250 None,
251 )
252 assert provider_mapping is not None
253 assert provider_mapping.available is False
254
255
256@pytest.mark.asyncio
257async def test_search_replacement_finds_exact_match() -> None:
258 """Search replacement finds exact match when deprecated catalog ID no longer exists."""
259 provider = _make_test_provider()
260
261 # Mock library item with track metadata
262 library_item = {
263 "id": "i.123",
264 "attributes": {
265 "name": "Test Track",
266 "artistName": "Test Artist",
267 "albumName": "Test Album",
268 },
269 }
270
271 # Create real Track instance for proper isinstance() check
272 mock_track = _make_test_track(
273 track_id="999",
274 track_name="Test Track",
275 artist_id="456",
276 artist_name="Test Artist",
277 album_id="789",
278 album_name="Test Album",
279 )
280
281 search_results = MagicMock()
282 search_results.tracks = [mock_track]
283
284 provider.media_manager.search = AsyncMock(return_value=search_results)
285
286 manager = AppleMusicLibraryManager(provider)
287
288 # Test search replacement
289 result = await manager._try_search_replacement_for_deprecated_track(library_item, True)
290
291 assert result is not None
292 assert result.name == "Test Track"
293 assert result.favorite is True
294 provider.media_manager.search.assert_called_once_with(
295 "Test Artist Test Track", [MediaType.TRACK], limit=10
296 )
297
298
299@pytest.mark.asyncio
300async def test_search_replacement_no_match_wrong_track_name() -> None:
301 """Search replacement returns None when track name doesn't match."""
302 provider = _make_test_provider()
303
304 library_item = {
305 "id": "i.123",
306 "attributes": {
307 "name": "Test Track",
308 "artistName": "Test Artist",
309 },
310 }
311
312 mock_track = _make_test_track(
313 track_id="999",
314 track_name="Different Song", # Wrong name
315 artist_id="456",
316 artist_name="Test Artist",
317 )
318
319 search_results = MagicMock()
320 search_results.tracks = [mock_track]
321
322 provider.media_manager.search = AsyncMock(return_value=search_results)
323
324 manager = AppleMusicLibraryManager(provider)
325 result = await manager._try_search_replacement_for_deprecated_track(library_item, False)
326
327 assert result is None
328
329
330@pytest.mark.asyncio
331async def test_search_replacement_no_match_wrong_artist() -> None:
332 """Search replacement returns None when artist name doesn't match."""
333 provider = _make_test_provider()
334
335 library_item = {
336 "id": "i.123",
337 "attributes": {
338 "name": "Test Track",
339 "artistName": "Test Artist",
340 },
341 }
342
343 mock_track = _make_test_track(
344 track_id="999",
345 track_name="Test Track",
346 artist_id="456",
347 artist_name="Different Artist", # Wrong artist
348 )
349
350 search_results = MagicMock()
351 search_results.tracks = [mock_track]
352
353 provider.media_manager.search = AsyncMock(return_value=search_results)
354
355 manager = AppleMusicLibraryManager(provider)
356 result = await manager._try_search_replacement_for_deprecated_track(library_item, False)
357
358 assert result is None
359
360
361@pytest.mark.asyncio
362async def test_search_replacement_album_mismatch_skipped() -> None:
363 """Search replacement skips tracks with mismatched album when album info available."""
364 provider = _make_test_provider()
365
366 library_item = {
367 "id": "i.123",
368 "attributes": {
369 "name": "Test Track",
370 "artistName": "Test Artist",
371 "albumName": "Test Album",
372 },
373 }
374
375 mock_track = _make_test_track(
376 track_id="999",
377 track_name="Test Track",
378 artist_id="456",
379 artist_name="Test Artist",
380 album_id="789",
381 album_name="Different Album", # Wrong album
382 )
383
384 search_results = MagicMock()
385 search_results.tracks = [mock_track]
386
387 provider.media_manager.search = AsyncMock(return_value=search_results)
388
389 manager = AppleMusicLibraryManager(provider)
390 result = await manager._try_search_replacement_for_deprecated_track(library_item, False)
391
392 assert result is None
393
394
395@pytest.mark.asyncio
396async def test_search_replacement_no_results() -> None:
397 """Search replacement returns None when search yields no results."""
398 provider = _make_test_provider()
399
400 library_item = {
401 "id": "i.123",
402 "attributes": {
403 "name": "Test Track",
404 "artistName": "Test Artist",
405 },
406 }
407
408 search_results = MagicMock()
409 search_results.tracks = [] # Empty results
410
411 provider.media_manager.search = AsyncMock(return_value=search_results)
412
413 manager = AppleMusicLibraryManager(provider)
414 result = await manager._try_search_replacement_for_deprecated_track(library_item, False)
415
416 assert result is None
417
418
419@pytest.mark.asyncio
420async def test_search_replacement_handles_exceptions() -> None:
421 """Search replacement returns None and logs when search raises exception."""
422 provider = _make_test_provider()
423
424 library_item = {
425 "id": "i.123",
426 "attributes": {
427 "name": "Test Track",
428 "artistName": "Test Artist",
429 },
430 }
431
432 provider.media_manager.search = AsyncMock(side_effect=Exception("Network error"))
433
434 manager = AppleMusicLibraryManager(provider)
435 result = await manager._try_search_replacement_for_deprecated_track(library_item, False)
436
437 assert result is None
438
439
440@pytest.mark.asyncio
441async def test_search_replacement_missing_metadata() -> None:
442 """Search replacement returns None when library item lacks required metadata."""
443 provider = _make_test_provider()
444
445 # Missing track name
446 library_item = {
447 "id": "i.123",
448 "attributes": {
449 "artistName": "Test Artist",
450 },
451 }
452
453 manager = AppleMusicLibraryManager(provider)
454 result = await manager._try_search_replacement_for_deprecated_track(library_item, False)
455
456 assert result is None
457
458 # Missing artist name
459 library_item = {
460 "id": "i.123",
461 "attributes": {
462 "name": "Test Track",
463 },
464 }
465
466 result = await manager._try_search_replacement_for_deprecated_track(library_item, False)
467
468 assert result is None
469
470
471@pytest.mark.asyncio
472async def test_search_replacement_skips_item_mappings() -> None:
473 """Search replacement skips ItemMapping entries and only processes Track objects."""
474 provider = _make_test_provider()
475
476 library_item = {
477 "id": "i.123",
478 "attributes": {
479 "name": "Test Track",
480 "artistName": "Test Artist",
481 "albumName": "Test Album",
482 },
483 }
484
485 # Create an ItemMapping instead of a Track
486 item_mapping = ItemMapping(
487 media_type=MediaType.TRACK,
488 item_id="999",
489 provider="apple_music",
490 name="Test Track",
491 )
492
493 search_results = MagicMock()
494 search_results.tracks = [item_mapping] # Only ItemMapping, no Track objects
495
496 provider.media_manager.search = AsyncMock(return_value=search_results)
497
498 manager = AppleMusicLibraryManager(provider)
499 result = await manager._try_search_replacement_for_deprecated_track(library_item, False)
500
501 # Should return None since ItemMapping should be skipped
502 assert result is None
503
504
505def _make_library_only_manager(
506 items: list[dict[str, Any]],
507) -> tuple[AppleMusicLibraryManager, MagicMock, list[tuple[str, dict[str, Any]]]]:
508 """Build a manager streaming library-only songs, recording every api call it makes."""
509 provider = MagicMock()
510 provider.domain = "apple_music"
511 provider.instance_id = "apple_music--test"
512 provider._storefront = "us"
513 api = provider.api_client
514 calls: list[tuple[str, dict[str, Any]]] = []
515
516 async def _iter(*_args: Any, **_kwargs: Any) -> Any:
517 for item in items:
518 yield item
519
520 async def _get_data(endpoint: str, **kwargs: Any) -> dict[str, Any]:
521 calls.append((endpoint, kwargs))
522 requested = kwargs.get("ids", "").split(",") if kwargs.get("ids") else []
523 return {
524 "data": [
525 _library_song_with_metadata(
526 int(item_id.split(".")[1]), catalog_id=None, album_name=f"Album {item_id}"
527 )
528 for item_id in requested
529 ]
530 }
531
532 api.iter_all_items = _iter
533 api.get_data = AsyncMock(side_effect=_get_data)
534 api.get_ratings = AsyncMock(return_value={})
535 return AppleMusicLibraryManager(provider), provider, calls
536
537
538@pytest.mark.asyncio
539async def test_library_only_detail_fetches_are_batched() -> None:
540 """Weak-mapped library-only tracks are enriched in batches, not one request per track."""
541 count = 250
542 items = [_library_song(idx, catalog_id=None) for idx in range(count)]
543 manager, _provider, calls = _make_library_only_manager(items)
544
545 tracks = [track async for track in manager.get_library_tracks()]
546
547 assert len(tracks) == count
548 # two windows (150 + 100), each batching its weak-mapped tracks at _DETAIL_BATCH_SIZE
549 assert len(calls) == 3
550 assert all(len(kwargs["ids"].split(",")) <= _DETAIL_BATCH_SIZE for _endpoint, kwargs in calls)
551 # the batched detail response still resolves the album that the listing lacked
552 assert all(track.album is not None for track in tracks)
553
554
555@pytest.mark.asyncio
556async def test_library_only_detail_batch_failure_reports_how_many_lost_detail() -> None:
557 """A failed detail batch keeps the listing tracks and warns with the number affected."""
558 count = 20
559 items = [_library_song(idx, catalog_id=None) for idx in range(count)]
560 manager, provider, _calls = _make_library_only_manager(items)
561 provider.api_client.get_data = AsyncMock(side_effect=MusicAssistantError("boom"))
562
563 tracks = [track async for track in manager.get_library_tracks()]
564
565 assert len(tracks) == count
566 assert all(track.album is None for track in tracks)
567 provider.logger.warning.assert_called_once()
568 assert count in provider.logger.warning.call_args.args
569