/
/
/
1"""Tests for the indexed external id lookup of media items."""
2
3from __future__ import annotations
4
5from collections.abc import AsyncGenerator
6
7import pytest
8from music_assistant_models.enums import ExternalID
9from music_assistant_models.media_items import Album, Artist, ProviderMapping, UniqueList
10
11from music_assistant.constants import DB_TABLE_EXTERNAL_ID_LOOKUP
12from music_assistant.controllers.music import MusicController
13from music_assistant.mass import MusicAssistant
14
15from .helpers import ISRC, create_track
16
17MBID = "b1a9c0e9-d987-4042-ae91-78d6a3267d69"
18BARCODE = "0724354283857"
19
20
21@pytest.fixture
22async def music(mass_minimal: MusicAssistant) -> AsyncGenerator[MusicController]:
23 """Return a music controller with a real library database."""
24 controller = MusicController(mass_minimal)
25 mass_minimal.music = controller
26 await controller._setup_database()
27 yield controller
28 if controller._database:
29 await controller._database.close()
30
31
32async def _get_lookup_rows(music: MusicController, item_id: int | str) -> set[tuple[str, str]]:
33 """Return the (external_id_type, external_id) lookup rows stored for a track."""
34 return {
35 (row["external_id_type"], row["external_id"])
36 for row in await music.database.get_rows(
37 DB_TABLE_EXTERNAL_ID_LOOKUP, {"media_type": "track", "item_id": int(item_id)}
38 )
39 }
40
41
42def _create_album(
43 provider_instance: str,
44 item_id: str,
45 name: str,
46 artist_name: str,
47 year: int,
48 barcode: str = BARCODE,
49) -> Album:
50 """
51 Create an album as received from a music provider.
52
53 :param provider_instance: Provider instance the album originates from.
54 :param item_id: Provider-native album identifier.
55 :param name: Album name.
56 :param artist_name: Album artist name.
57 :param year: Album release year.
58 :param barcode: Album barcode.
59 """
60 provider_domain = provider_instance.rsplit("_", 1)[0]
61 return Album(
62 item_id=item_id,
63 provider=provider_instance,
64 name=name,
65 year=year,
66 external_ids={(ExternalID.BARCODE, barcode)},
67 provider_mappings={
68 ProviderMapping(
69 item_id=item_id,
70 provider_domain=provider_domain,
71 provider_instance=provider_instance,
72 )
73 },
74 artists=UniqueList(
75 [
76 Artist(
77 item_id=f"{item_id}_artist",
78 provider=provider_instance,
79 name=artist_name,
80 provider_mappings={
81 ProviderMapping(
82 item_id=f"{item_id}_artist",
83 provider_domain=provider_domain,
84 provider_instance=provider_instance,
85 )
86 },
87 )
88 ]
89 ),
90 )
91
92
93async def test_same_isrc_from_two_providers_dedupes(music: MusicController) -> None:
94 """Two providers exposing the same track with an identical ISRC merge into one item."""
95 library_track_1 = await music.tracks.add_item_to_library(create_track("spotify_1", "track_abc"))
96 library_track_2 = await music.tracks.add_item_to_library(create_track("tidal_1", "track_xyz"))
97
98 assert library_track_1.item_id == library_track_2.item_id
99 assert len(library_track_2.provider_mappings) == 2
100 assert await music.tracks.library_count() == 1
101
102
103async def test_formatted_isrc_from_two_providers_dedupes(music: MusicController) -> None:
104 """Equivalent formatted ISRC values merge into one library track."""
105 first = create_track("spotify_1", "track_abc", isrc="US-RC1-76-07839")
106 second = create_track("tidal_1", "track_xyz", isrc="usrc17607839")
107
108 library_track_1 = await music.tracks.add_item_to_library(first)
109 library_track_2 = await music.tracks.add_item_to_library(second)
110
111 assert library_track_1.item_id == library_track_2.item_id
112 assert library_track_2.external_ids == {(ExternalID.ISRC, ISRC)}
113 untyped_match = await music.tracks.get_library_item_by_external_id("US-RC1-76-07839")
114 assert untyped_match is not None
115 assert untyped_match.item_id == library_track_1.item_id
116
117
118async def test_matching_checks_more_than_fifty_external_id_candidates(
119 music: MusicController,
120) -> None:
121 """Internal matching paginates every row sharing a non-unique identifier."""
122 for index in range(50):
123 await music.tracks.add_item_to_library(
124 create_track(
125 f"provider{index}_1",
126 f"collision_{index}",
127 name=f"Collision {index}",
128 duration=100 + index * 20,
129 )
130 )
131 expected = await music.tracks.add_item_to_library(
132 create_track("expected_1", "expected", name="Expected", duration=2000)
133 )
134
135 matched = await music.tracks.add_item_to_library(
136 create_track("incoming_1", "incoming", name="Different name", duration=2000)
137 )
138
139 assert matched.item_id == expected.item_id
140 assert await music.tracks.library_count() == 51
141
142
143async def test_non_unique_external_id_candidates_are_all_verified(
144 music: MusicController,
145) -> None:
146 """An unrelated barcode collision does not hide the correct album candidate."""
147 unrelated = await music.albums.add_item_to_library(
148 _create_album("apple_music_1", "unrelated", "Keeping It All Low", "XP", 2019)
149 )
150 expected = await music.albums.add_item_to_library(
151 _create_album(
152 "apple_music_1",
153 "expected",
154 "#1",
155 "Fischerspooner",
156 2001,
157 barcode="000724354283857",
158 )
159 )
160 await music.database.execute_write(
161 f"UPDATE {DB_TABLE_EXTERNAL_ID_LOOKUP} SET external_id = :external_id "
162 "WHERE media_type = 'album' AND item_id = :item_id",
163 {"external_id": "000724354283857", "item_id": int(expected.item_id)},
164 )
165
166 matched = await music.albums.add_item_to_library(
167 _create_album("qobuz_1", "qobuz_release", "#1", "Fischerspooner", 2002)
168 )
169
170 assert matched.item_id == expected.item_id
171 assert matched.item_id != unrelated.item_id
172 assert await music.albums.library_count() == 2
173 assert {mapping.provider_instance for mapping in matched.provider_mappings} == {
174 "apple_music_1",
175 "qobuz_1",
176 }
177
178
179async def test_get_library_item_by_external_id(music: MusicController) -> None:
180 """Library items resolve by external id, both typed and untyped."""
181 track = create_track("spotify_1", "track_abc")
182 track.external_ids.add((ExternalID.MB_RECORDING, MBID))
183 library_track = await music.tracks.add_item_to_library(track)
184
185 # typed lookup
186 match = await music.tracks.get_library_item_by_external_id(ISRC, ExternalID.ISRC)
187 assert match is not None
188 assert match.item_id == library_track.item_id
189 match = await music.tracks.get_library_item_by_external_id(MBID, ExternalID.MB_RECORDING)
190 assert match is not None
191 assert match.item_id == library_track.item_id
192 # untyped lookup
193 match = await music.tracks.get_library_item_by_external_id(ISRC)
194 assert match is not None
195 assert match.item_id == library_track.item_id
196 # matching is case-insensitive (as the previous LIKE based scan was)
197 match = await music.tracks.get_library_item_by_external_id(ISRC.lower(), ExternalID.ISRC)
198 assert match is not None
199 assert match.item_id == library_track.item_id
200 # no (partial) match on wrong type or unknown id
201 assert await music.tracks.get_library_item_by_external_id(ISRC, ExternalID.BARCODE) is None
202 assert await music.tracks.get_library_item_by_external_id("something-else") is None
203 assert await music.tracks.get_library_item_by_external_id(ISRC[:-1]) is None
204
205
206async def test_external_id_lookup_rows_follow_item_updates(music: MusicController) -> None:
207 """The lookup rows are kept in sync when an item is updated or removed."""
208 library_track = await music.tracks.add_item_to_library(create_track("spotify_1", "track_abc"))
209 assert await _get_lookup_rows(music, library_track.item_id) == {(str(ExternalID.ISRC), ISRC)}
210
211 # an update merges in newly discovered external ids
212 update = create_track("spotify_1", "track_abc")
213 update.external_ids.add((ExternalID.MB_RECORDING, MBID))
214 updated = await music.tracks.update_item_in_library(library_track.item_id, update)
215 # the item's external_ids attribute is reconstructed from the lookup table on read
216 assert updated.external_ids == {(ExternalID.ISRC, ISRC), (ExternalID.MB_RECORDING, MBID)}
217 assert await _get_lookup_rows(music, library_track.item_id) == {
218 (str(ExternalID.ISRC), ISRC),
219 (str(ExternalID.MB_RECORDING), MBID),
220 }
221
222 # an overwrite update replaces the lookup rows
223 await music.tracks.update_item_in_library(
224 library_track.item_id,
225 create_track("spotify_1", "track_abc", isrc="GBUM71029604"),
226 overwrite=True,
227 )
228 assert await _get_lookup_rows(music, library_track.item_id) == {
229 (str(ExternalID.ISRC), "GBUM71029604")
230 }
231 assert await music.tracks.get_library_item_by_external_id(ISRC) is None
232
233 # removal cleans up the lookup rows
234 await music.tracks.remove_item_from_library(library_track.item_id)
235 assert await _get_lookup_rows(music, library_track.item_id) == set()
236