/
/
/
1"""Integration tests for non-destructive library item merges."""
2
3from __future__ import annotations
4
5from unittest.mock import AsyncMock, patch
6
7import pytest
8from music_assistant_models.enums import AlbumType, ArtistType, ExternalID, MediaType
9from music_assistant_models.errors import InvalidDataError, MediaNotFoundError, MusicAssistantError
10from music_assistant_models.media_items import (
11 Album,
12 Artist,
13 Audiobook,
14 Genre,
15 ProviderMapping,
16 Track,
17 UniqueList,
18)
19
20from music_assistant.constants import (
21 DB_TABLE_ALBUM_ARTISTS,
22 DB_TABLE_ALBUM_TRACKS,
23 DB_TABLE_ALBUMS,
24 DB_TABLE_GENRE_MEDIA_ITEM_EXCLUSION,
25 DB_TABLE_GENRE_MEDIA_ITEM_MAPPING,
26 DB_TABLE_PLAYLOG,
27 DB_TABLE_PROVIDER_MAPPINGS,
28 DB_TABLE_TRACK_ARTISTS,
29 DB_TABLE_TRACKS,
30)
31from music_assistant.controllers.music.media.base import (
32 SUPPRESS_MEDIA_ITEM_UPDATES,
33 MediaControllerBase,
34)
35from music_assistant.mass import MusicAssistant
36
37
38def _mapping(provider_instance: str, item_id: str) -> ProviderMapping:
39 """Create a provider mapping for a library fixture item."""
40 return ProviderMapping(
41 item_id=item_id,
42 provider_domain=provider_instance.removesuffix("_instance"),
43 provider_instance=provider_instance,
44 in_library=True,
45 )
46
47
48async def _add_artist(
49 mass: MusicAssistant, name: str, provider_instance: str, provider_item_id: str
50) -> Artist:
51 """Create and store a fixture artist."""
52 return await mass.music.artists.add_item_to_library(
53 Artist(
54 item_id="0",
55 provider="library",
56 name=name,
57 provider_mappings={_mapping(provider_instance, provider_item_id)},
58 )
59 )
60
61
62async def _add_album(
63 mass: MusicAssistant,
64 name: str,
65 provider_instance: str,
66 provider_item_id: str,
67 artist: Artist,
68 external_id: str,
69) -> Album:
70 """Create and store a fixture album."""
71 return await mass.music.albums.add_item_to_library(
72 Album(
73 item_id="0",
74 provider="library",
75 name=name,
76 album_type=AlbumType.ALBUM,
77 provider_mappings={_mapping(provider_instance, provider_item_id)},
78 external_ids={(ExternalID.BARCODE, external_id)},
79 artists=UniqueList([artist]),
80 )
81 )
82
83
84async def _add_track(
85 mass: MusicAssistant,
86 name: str,
87 provider_instance: str,
88 provider_item_id: str,
89 artist: Artist,
90 album: Album,
91) -> Track:
92 """Create and store a fixture track."""
93 return await mass.music.tracks.add_item_to_library(
94 Track(
95 item_id="0",
96 provider="library",
97 name=name,
98 provider_mappings={_mapping(provider_instance, provider_item_id)},
99 artists=UniqueList([artist]),
100 album=album,
101 disc_number=1,
102 track_number=1,
103 )
104 )
105
106
107async def _add_playlog(
108 mass: MusicAssistant,
109 item_id: str | int,
110 provider: str,
111 media_type: MediaType,
112 *,
113 timestamp: int,
114 seconds_played: int,
115 user_initiated: bool,
116) -> None:
117 """Insert a playlog row for merge coverage."""
118 await mass.music.database.insert(
119 DB_TABLE_PLAYLOG,
120 {
121 "item_id": item_id,
122 "provider": provider,
123 "media_type": media_type.value,
124 "name": "Merge test",
125 "timestamp": timestamp,
126 "seconds_played": seconds_played,
127 "userid": "test-user",
128 "user_initiated": user_initiated,
129 },
130 )
131
132
133async def test_mapping_conflict_merges_albums_without_deleting_tracks(
134 mass: MusicAssistant,
135) -> None:
136 """A mapping conflict merges albums without recursively deleting their tracks."""
137 target_artist = await _add_artist(mass, "Target Artist", "target_instance", "target-artist")
138 source_artist = await _add_artist(mass, "Source Artist", "source_instance", "source-artist")
139 target = await _add_album(
140 mass,
141 "Target Album",
142 "target_instance",
143 "target-album",
144 target_artist,
145 "target-barcode",
146 )
147 source = await _add_album(
148 mass,
149 "Source Album",
150 "source_instance",
151 "source-album",
152 source_artist,
153 "source-barcode",
154 )
155 shared_track = await _add_track(
156 mass,
157 "Shared Track",
158 "shared_instance",
159 "shared-track",
160 target_artist,
161 target,
162 )
163 source_track = await _add_track(
164 mass,
165 "Source Track",
166 "source-track_instance",
167 "source-track",
168 source_artist,
169 source,
170 )
171 await mass.music.database.insert(
172 DB_TABLE_ALBUM_TRACKS,
173 {
174 "track_id": int(shared_track.item_id),
175 "album_id": int(source.item_id),
176 "disc_number": 1,
177 "track_number": 1,
178 },
179 )
180 await mass.music.database.update(
181 DB_TABLE_ALBUMS,
182 {"item_id": int(target.item_id)},
183 {
184 "favorite": False,
185 "play_count": 2,
186 "last_played": 20,
187 "timestamp_added": 20,
188 "timestamp_modified": 20,
189 },
190 )
191 await mass.music.database.update(
192 DB_TABLE_ALBUMS,
193 {"item_id": int(source.item_id)},
194 {
195 "favorite": True,
196 "play_count": 3,
197 "last_played": 30,
198 "timestamp_added": 10,
199 "timestamp_modified": 30,
200 },
201 )
202 genre = await mass.music.genres.add_item_to_library(
203 Genre(item_id="0", provider="library", name="Merge Genre", provider_mappings=set())
204 )
205 await mass.music.database.insert(
206 DB_TABLE_GENRE_MEDIA_ITEM_MAPPING,
207 {
208 "genre_id": int(genre.item_id),
209 "media_id": int(target.item_id),
210 "media_type": MediaType.ALBUM.value,
211 "alias": "target",
212 "is_derived": True,
213 "is_manual": False,
214 },
215 )
216 await mass.music.database.insert(
217 DB_TABLE_GENRE_MEDIA_ITEM_MAPPING,
218 {
219 "genre_id": int(genre.item_id),
220 "media_id": int(source.item_id),
221 "media_type": MediaType.ALBUM.value,
222 "alias": "source",
223 "is_derived": False,
224 "is_manual": True,
225 },
226 )
227 await mass.music.database.insert(
228 DB_TABLE_GENRE_MEDIA_ITEM_EXCLUSION,
229 {
230 "genre_id": int(genre.item_id),
231 "media_id": int(source.item_id),
232 "media_type": MediaType.ALBUM.value,
233 },
234 )
235 await _add_playlog(
236 mass,
237 target.item_id,
238 "library",
239 MediaType.ALBUM,
240 timestamp=20,
241 seconds_played=20,
242 user_initiated=False,
243 )
244 await _add_playlog(
245 mass,
246 source.item_id,
247 "library",
248 MediaType.ALBUM,
249 timestamp=30,
250 seconds_played=30,
251 user_initiated=True,
252 )
253 await _add_playlog(
254 mass,
255 "source-album",
256 "source_instance",
257 MediaType.ALBUM,
258 timestamp=30,
259 seconds_played=30,
260 user_initiated=True,
261 )
262
263 original_remove = MediaControllerBase.remove_item_from_library
264
265 async def assert_transferred_before_delete(
266 controller: MediaControllerBase[Album], item_id: str | int, recursive: bool = True
267 ) -> None:
268 if int(item_id) == int(source.item_id):
269 assert await mass.music.database.get_row(
270 DB_TABLE_PROVIDER_MAPPINGS,
271 {
272 "media_type": MediaType.ALBUM.value,
273 "item_id": int(target.item_id),
274 "provider_instance": "source_instance",
275 "provider_item_id": "source-album",
276 },
277 )
278 assert await mass.music.database.get_row(
279 DB_TABLE_ALBUM_TRACKS,
280 {"track_id": int(source_track.item_id), "album_id": int(target.item_id)},
281 )
282 await original_remove(controller, item_id, recursive)
283
284 with (
285 patch.object(
286 MediaControllerBase, "remove_item_from_library", assert_transferred_before_delete
287 ),
288 patch.object(
289 mass.music.albums,
290 "update_item_in_library",
291 wraps=mass.music.albums.update_item_in_library,
292 ) as update_item,
293 ):
294 await mass.music.albums.add_provider_mappings(target.item_id, source.provider_mappings)
295 update_item.assert_not_awaited()
296
297 await _assert_album_merge_result(
298 mass,
299 target,
300 source,
301 shared_track,
302 source_track,
303 target_artist,
304 source_artist,
305 genre,
306 )
307
308
309async def test_track_merge_preserves_album_and_artist_relations(mass: MusicAssistant) -> None:
310 """A track merge transfers all album and artist relations without deleting albums."""
311 target_artist = await _add_artist(mass, "Target Artist", "target_instance", "target-artist")
312 source_artist = await _add_artist(mass, "Source Artist", "source_instance", "source-artist")
313 target_album = await _add_album(
314 mass,
315 "Target Album",
316 "target_instance",
317 "target-album",
318 target_artist,
319 "target-barcode",
320 )
321 source_album = await _add_album(
322 mass,
323 "Source Album",
324 "source_instance",
325 "source-album",
326 source_artist,
327 "source-barcode",
328 )
329 target = await _add_track(
330 mass,
331 "Target Track",
332 "target_instance",
333 "target-track",
334 target_artist,
335 target_album,
336 )
337 source = await _add_track(
338 mass,
339 "Source Track",
340 "source_instance",
341 "source-track",
342 source_artist,
343 source_album,
344 )
345 await mass.music.database.insert(
346 DB_TABLE_ALBUM_TRACKS,
347 {
348 "track_id": int(target.item_id),
349 "album_id": int(source_album.item_id),
350 "disc_number": 7,
351 "track_number": 8,
352 },
353 )
354
355 await mass.music.tracks.merge_library_items(target.item_id, source.item_id)
356
357 with pytest.raises(MediaNotFoundError):
358 await mass.music.tracks.get_library_item(source.item_id)
359 assert await mass.music.albums.get_library_item(target_album.item_id)
360 assert await mass.music.albums.get_library_item(source_album.item_id)
361 assert {
362 int(row["album_id"])
363 for row in await mass.music.database.get_rows(
364 DB_TABLE_ALBUM_TRACKS, {"track_id": int(target.item_id)}
365 )
366 } == {int(target_album.item_id), int(source_album.item_id)}
367 assert {
368 int(row["artist_id"])
369 for row in await mass.music.database.get_rows(
370 DB_TABLE_TRACK_ARTISTS, {"track_id": int(target.item_id)}
371 )
372 } == {int(target_artist.item_id), int(source_artist.item_id)}
373 source_album_track = await mass.music.database.get_row(
374 DB_TABLE_ALBUM_TRACKS,
375 {"track_id": int(target.item_id), "album_id": int(source_album.item_id)},
376 )
377 assert source_album_track is not None
378 assert source_album_track["disc_number"] == 7
379 assert source_album_track["track_number"] == 8
380 with pytest.raises(MediaNotFoundError):
381 await mass.music.tracks.merge_library_items(target.item_id, source.item_id)
382
383
384async def test_audiobook_merge_keeps_per_user_resume_state(mass: MusicAssistant) -> None:
385 """An audiobook merge leaves resume transfer to the library playlog merge."""
386 target = await mass.music.audiobooks.add_item_to_library(
387 Audiobook(
388 item_id="0",
389 provider="library",
390 name="Target Book",
391 duration=3600,
392 provider_mappings={_mapping("target_instance", "target-book")},
393 )
394 )
395 source = await mass.music.audiobooks.add_item_to_library(
396 Audiobook(
397 item_id="0",
398 provider="library",
399 name="Source Book",
400 duration=3600,
401 provider_mappings={_mapping("source_instance", "source-book")},
402 )
403 )
404 await _add_playlog(
405 mass,
406 target.item_id,
407 "library",
408 MediaType.AUDIOBOOK,
409 timestamp=20,
410 seconds_played=20,
411 user_initiated=False,
412 )
413 await _add_playlog(
414 mass,
415 source.item_id,
416 "library",
417 MediaType.AUDIOBOOK,
418 timestamp=30,
419 seconds_played=30,
420 user_initiated=True,
421 )
422
423 with patch.object(
424 mass.music.audiobooks,
425 "_set_playlog",
426 wraps=mass.music.audiobooks._set_playlog,
427 ) as set_playlog:
428 await mass.music.audiobooks.merge_library_items(target.item_id, source.item_id)
429
430 assert isinstance(set_playlog, AsyncMock)
431 set_playlog.assert_not_awaited()
432 playlog = await mass.music.database.get_row(
433 DB_TABLE_PLAYLOG,
434 {
435 "item_id": target.item_id,
436 "provider": "library",
437 "media_type": MediaType.AUDIOBOOK.value,
438 "userid": "test-user",
439 },
440 )
441 assert playlog is not None
442 assert playlog["timestamp"] == 30
443 assert playlog["seconds_played"] == 30
444 assert playlog["user_initiated"] == 1
445
446
447async def test_genre_merge_transfers_genre_references(mass: MusicAssistant) -> None:
448 """A genre merge reassigns every media mapping and exclusion to the target genre."""
449 target = await mass.music.genres.add_item_to_library(
450 Genre(item_id="0", provider="library", name="Target Genre", provider_mappings=set())
451 )
452 source = await mass.music.genres.add_item_to_library(
453 Genre(item_id="0", provider="library", name="Source Genre", provider_mappings=set())
454 )
455 await mass.music.database.insert(
456 DB_TABLE_GENRE_MEDIA_ITEM_MAPPING,
457 {
458 "genre_id": int(target.item_id),
459 "media_id": 1,
460 "media_type": MediaType.TRACK.value,
461 "alias": "target",
462 "is_derived": True,
463 "is_manual": False,
464 },
465 )
466 await mass.music.database.insert(
467 DB_TABLE_GENRE_MEDIA_ITEM_MAPPING,
468 {
469 "genre_id": int(source.item_id),
470 "media_id": 1,
471 "media_type": MediaType.TRACK.value,
472 "alias": "source",
473 "is_derived": False,
474 "is_manual": True,
475 },
476 )
477 await mass.music.database.insert(
478 DB_TABLE_GENRE_MEDIA_ITEM_EXCLUSION,
479 {
480 "genre_id": int(source.item_id),
481 "media_id": 2,
482 "media_type": MediaType.TRACK.value,
483 },
484 )
485
486 await mass.music.genres.merge_library_items(target.item_id, source.item_id)
487
488 genre_mapping = await mass.music.database.get_row(
489 DB_TABLE_GENRE_MEDIA_ITEM_MAPPING,
490 {
491 "genre_id": int(target.item_id),
492 "media_id": 1,
493 "media_type": MediaType.TRACK.value,
494 },
495 )
496 assert genre_mapping is not None
497 assert dict(genre_mapping) == {
498 "genre_id": int(target.item_id),
499 "media_id": 1,
500 "media_type": MediaType.TRACK.value,
501 "alias": "source",
502 "is_derived": 1,
503 "is_manual": 1,
504 }
505 assert await mass.music.database.get_row(
506 DB_TABLE_GENRE_MEDIA_ITEM_EXCLUSION,
507 {
508 "genre_id": int(target.item_id),
509 "media_id": 2,
510 "media_type": MediaType.TRACK.value,
511 },
512 )
513 assert not await mass.music.database.get_rows(
514 DB_TABLE_GENRE_MEDIA_ITEM_MAPPING, {"genre_id": int(source.item_id)}
515 )
516 assert not await mass.music.database.get_rows(
517 DB_TABLE_GENRE_MEDIA_ITEM_EXCLUSION, {"genre_id": int(source.item_id)}
518 )
519
520
521async def test_merge_honors_outer_event_suppression(mass: MusicAssistant) -> None:
522 """A merge inside a suppressed update scope does not emit item events."""
523 target = await mass.music.genres.add_item_to_library(
524 Genre(item_id="0", provider="library", name="Target Genre", provider_mappings=set())
525 )
526 source = await mass.music.genres.add_item_to_library(
527 Genre(item_id="0", provider="library", name="Source Genre", provider_mappings=set())
528 )
529 token = SUPPRESS_MEDIA_ITEM_UPDATES.set(True)
530 try:
531 with patch.object(mass, "signal_event") as signal_event:
532 await mass.music.genres.merge_library_items(target.item_id, source.item_id)
533 finally:
534 SUPPRESS_MEDIA_ITEM_UPDATES.reset(token)
535 signal_event.assert_not_called()
536
537
538async def test_genre_merge_rejects_different_taxonomies(mass: MusicAssistant) -> None:
539 """A genre merge cannot cross music and podcast taxonomies."""
540 target = await mass.music.genres.add_item_to_library(
541 Genre(
542 item_id="0",
543 provider="library",
544 name="Podcast Genre",
545 content_type=MediaType.PODCAST,
546 provider_mappings=set(),
547 )
548 )
549 source = await mass.music.genres.add_item_to_library(
550 Genre(item_id="0", provider="library", name="Music Genre", provider_mappings=set())
551 )
552
553 with pytest.raises(InvalidDataError, match="same taxonomy"):
554 await mass.music.genres.merge_library_items(target.item_id, source.item_id)
555
556 assert await mass.music.genres.get_library_item(target.item_id)
557 assert await mass.music.genres.get_library_item(source.item_id)
558
559
560async def test_artist_merge_rejects_different_roles(mass: MusicAssistant) -> None:
561 """An artist merge cannot cross music and spoken-word roles."""
562 target = await mass.music.artists.add_item_to_library(
563 Artist(
564 item_id="0",
565 provider="library",
566 name="Singer",
567 artist_type=ArtistType.SINGER,
568 provider_mappings={_mapping("target_instance", "target-artist")},
569 )
570 )
571 source = await mass.music.artists.add_item_to_library(
572 Artist(
573 item_id="0",
574 provider="library",
575 name="Author",
576 artist_type=ArtistType.AUTHOR,
577 provider_mappings={_mapping("source_instance", "source-artist")},
578 )
579 )
580
581 with pytest.raises(InvalidDataError, match="same role"):
582 await mass.music.artists.merge_library_items(target.item_id, source.item_id)
583
584 assert await mass.music.artists.get_library_item(target.item_id)
585 assert await mass.music.artists.get_library_item(source.item_id)
586
587
588async def test_merge_retry_does_not_double_play_count(mass: MusicAssistant) -> None:
589 """A retry after a later transfer failure preserves the merged play count."""
590 artist = await _add_artist(mass, "Artist", "target_instance", "target-artist")
591 target_album = await _add_album(
592 mass,
593 "Target Album",
594 "target_instance",
595 "target-album",
596 artist,
597 "target-barcode",
598 )
599 source_album = await _add_album(
600 mass,
601 "Source Album",
602 "source_instance",
603 "source-album",
604 artist,
605 "source-barcode",
606 )
607 target = await _add_track(
608 mass,
609 "Target Track",
610 "target_instance",
611 "target-track",
612 artist,
613 target_album,
614 )
615 source = await _add_track(
616 mass,
617 "Source Track",
618 "source_instance",
619 "source-track",
620 artist,
621 source_album,
622 )
623 await mass.music.database.update(
624 DB_TABLE_TRACKS, {"item_id": int(target.item_id)}, {"play_count": 2}
625 )
626 await mass.music.database.update(
627 DB_TABLE_TRACKS, {"item_id": int(source.item_id)}, {"play_count": 3}
628 )
629
630 with (
631 patch.object(
632 mass.music.tracks,
633 "_merge_genre_mappings",
634 AsyncMock(side_effect=MusicAssistantError("merge failure")),
635 ),
636 pytest.raises(MusicAssistantError, match="merge failure"),
637 ):
638 await mass.music.tracks.merge_library_items(target.item_id, source.item_id)
639
640 target_row = await mass.music.database.get_row(
641 DB_TABLE_TRACKS, {"item_id": int(target.item_id)}
642 )
643 source_row = await mass.music.database.get_row(
644 DB_TABLE_TRACKS, {"item_id": int(source.item_id)}
645 )
646 assert target_row is not None
647 assert source_row is not None
648 assert target_row["play_count"] == 5
649 assert source_row["play_count"] == 0
650
651 await mass.music.tracks.merge_library_items(target.item_id, source.item_id)
652
653 merged_row = await mass.music.database.get_row(
654 DB_TABLE_TRACKS, {"item_id": int(target.item_id)}
655 )
656 assert merged_row is not None
657 assert merged_row["play_count"] == 5
658
659
660async def test_merge_interrupted_after_relation_transfer_leaves_source_reconcilable(
661 mass: MusicAssistant,
662) -> None:
663 """A merge cut short right after the relation transfer keeps the source row repairable."""
664 artist = await _add_artist(mass, "Shared Artist", "target_instance", "shared-artist")
665 album = await _add_album(
666 mass, "Shared Album", "target_instance", "shared-album", artist, "shared-barcode"
667 )
668 target = await _add_track(
669 mass, "Target Track", "target_instance", "target-track", artist, album
670 )
671 source = await _add_track(
672 mass, "Source Track", "source_instance", "source-track", artist, album
673 )
674
675 original = MediaControllerBase._copy_library_item_relations
676
677 async def _fail_after_relation_transfer(
678 self: MediaControllerBase[Track], target_id: int, source_id: int
679 ) -> None:
680 await original(self, target_id, source_id)
681 raise MusicAssistantError("interrupted")
682
683 with (
684 patch.object(
685 MediaControllerBase, "_copy_library_item_relations", _fail_after_relation_transfer
686 ),
687 pytest.raises(MusicAssistantError, match="interrupted"),
688 ):
689 await mass.music.tracks.merge_library_items(target.item_id, source.item_id)
690
691 # the source still owns the album and artist relations that identify it, so the
692 # duplicate reconciliation pass can still see the pair and finish the merge later
693 assert await mass.music.database.get_rows(
694 DB_TABLE_ALBUM_TRACKS, {"track_id": int(source.item_id)}
695 )
696 assert await mass.music.database.get_rows(
697 DB_TABLE_TRACK_ARTISTS, {"track_id": int(source.item_id)}
698 )
699 assert await mass.music.database.get_rows(
700 DB_TABLE_PROVIDER_MAPPINGS,
701 {"media_type": MediaType.TRACK.value, "item_id": int(source.item_id)},
702 )
703
704 await mass.music.tracks.merge_library_items(target.item_id, source.item_id)
705
706 with pytest.raises(MediaNotFoundError):
707 await mass.music.tracks.get_library_item(source.item_id)
708 assert {
709 int(row["album_id"])
710 for row in await mass.music.database.get_rows(
711 DB_TABLE_ALBUM_TRACKS, {"track_id": int(target.item_id)}
712 )
713 } == {int(album.item_id)}
714 assert not await mass.music.database.get_rows(
715 DB_TABLE_ALBUM_TRACKS, {"track_id": int(source.item_id)}
716 )
717 assert not await mass.music.database.get_rows(
718 DB_TABLE_TRACK_ARTISTS, {"track_id": int(source.item_id)}
719 )
720 assert {
721 int(row["item_id"])
722 for row in await mass.music.database.get_rows(
723 DB_TABLE_PROVIDER_MAPPINGS, {"media_type": MediaType.TRACK.value}
724 )
725 } == {int(target.item_id)}
726
727
728async def test_merge_interrupted_before_relation_drop_keeps_target_complete(
729 mass: MusicAssistant,
730) -> None:
731 """A merge cut short before the source relations are dropped leaves nothing to lose."""
732 artist = await _add_artist(mass, "Shared Artist", "target_instance", "shared-artist")
733 album = await _add_album(
734 mass, "Shared Album", "target_instance", "shared-album", artist, "shared-barcode"
735 )
736 source_album = await _add_album(
737 mass, "Source Album", "source_instance", "source-album", artist, "source-barcode"
738 )
739 target = await _add_track(
740 mass, "Target Track", "target_instance", "target-track", artist, album
741 )
742 source = await _add_track(
743 mass, "Source Track", "source_instance", "source-track", artist, source_album
744 )
745
746 with (
747 patch.object(
748 MediaControllerBase,
749 "_drop_library_item_relations",
750 AsyncMock(side_effect=MusicAssistantError("interrupted")),
751 ),
752 pytest.raises(MusicAssistantError, match="interrupted"),
753 ):
754 await mass.music.tracks.merge_library_items(target.item_id, source.item_id)
755
756 # the source has already handed over its provider mappings, so the source-less item
757 # cleanup may delete it at any time: the target must hold every relation by now
758 assert not await mass.music.database.get_rows(
759 DB_TABLE_PROVIDER_MAPPINGS,
760 {"media_type": MediaType.TRACK.value, "item_id": int(source.item_id)},
761 )
762 assert {
763 int(row["album_id"])
764 for row in await mass.music.database.get_rows(
765 DB_TABLE_ALBUM_TRACKS, {"track_id": int(target.item_id)}
766 )
767 } == {int(album.item_id), int(source_album.item_id)}
768 assert {
769 int(row["artist_id"])
770 for row in await mass.music.database.get_rows(
771 DB_TABLE_TRACK_ARTISTS, {"track_id": int(target.item_id)}
772 )
773 } == {int(artist.item_id)}
774
775
776async def _assert_album_merge_result(
777 mass: MusicAssistant,
778 target: Album,
779 source: Album,
780 shared_track: Track,
781 source_track: Track,
782 target_artist: Artist,
783 source_artist: Artist,
784 genre: Genre,
785) -> None:
786 """Assert that an album merge preserved the source state on the target."""
787 with pytest.raises(MediaNotFoundError):
788 await mass.music.albums.get_library_item(source.item_id)
789 assert await mass.music.tracks.get_library_item(shared_track.item_id)
790 assert await mass.music.tracks.get_library_item(source_track.item_id)
791 merged = await mass.music.albums.get_library_item(target.item_id)
792 assert merged.favorite is True
793 merged_row = await mass.music.database.get_row(
794 DB_TABLE_ALBUMS, {"item_id": int(target.item_id)}
795 )
796 assert merged_row is not None
797 assert merged_row["play_count"] == 5
798 assert merged_row["last_played"] == 30
799 assert merged_row["timestamp_added"] == 10
800 assert {(kind, value) for kind, value in merged.external_ids} == {
801 (ExternalID.BARCODE, "target-barcode"),
802 (ExternalID.BARCODE, "source-barcode"),
803 }
804 assert {
805 int(row["track_id"])
806 for row in await mass.music.database.get_rows(
807 DB_TABLE_ALBUM_TRACKS, {"album_id": int(target.item_id)}
808 )
809 } == {int(shared_track.item_id), int(source_track.item_id)}
810 assert not await mass.music.database.get_rows(
811 DB_TABLE_ALBUM_TRACKS, {"album_id": int(source.item_id)}
812 )
813 assert {
814 int(row["artist_id"])
815 for row in await mass.music.database.get_rows(
816 DB_TABLE_ALBUM_ARTISTS, {"album_id": int(target.item_id)}
817 )
818 } == {int(target_artist.item_id), int(source_artist.item_id)}
819 genre_mapping = await mass.music.database.get_row(
820 DB_TABLE_GENRE_MEDIA_ITEM_MAPPING,
821 {
822 "genre_id": int(genre.item_id),
823 "media_id": int(target.item_id),
824 "media_type": MediaType.ALBUM.value,
825 },
826 )
827 assert genre_mapping is not None
828 assert dict(genre_mapping) == {
829 "genre_id": int(genre.item_id),
830 "media_id": int(target.item_id),
831 "media_type": MediaType.ALBUM.value,
832 "alias": "source",
833 "is_derived": 1,
834 "is_manual": 1,
835 }
836 assert await mass.music.database.get_row(
837 DB_TABLE_GENRE_MEDIA_ITEM_EXCLUSION,
838 {
839 "genre_id": int(genre.item_id),
840 "media_id": int(target.item_id),
841 "media_type": MediaType.ALBUM.value,
842 },
843 )
844 playlog = await mass.music.database.get_row(
845 DB_TABLE_PLAYLOG,
846 {
847 "item_id": target.item_id,
848 "provider": "library",
849 "media_type": MediaType.ALBUM.value,
850 "userid": "test-user",
851 },
852 )
853 assert playlog is not None
854 assert playlog["timestamp"] == 30
855 assert playlog["seconds_played"] == 30
856 assert playlog["user_initiated"] == 1
857 assert await mass.music.database.get_row(
858 DB_TABLE_PLAYLOG,
859 {
860 "item_id": "source-album",
861 "provider": "source_instance",
862 "media_type": MediaType.ALBUM.value,
863 "userid": "test-user",
864 },
865 )
866