/
/
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
9
10from music_assistant.constants import DB_TABLE_EXTERNAL_ID_LOOKUP
11from music_assistant.controllers.music import MusicController
12from music_assistant.mass import MusicAssistant
13
14from .helpers import ISRC, create_track
15
16MBID = "b1a9c0e9-d987-4042-ae91-78d6a3267d69"
17
18
19@pytest.fixture
20async def music(mass_minimal: MusicAssistant) -> AsyncGenerator[MusicController]:
21 """Return a music controller with a real library database."""
22 controller = MusicController(mass_minimal)
23 mass_minimal.music = controller
24 await controller._setup_database()
25 yield controller
26 if controller._database:
27 await controller._database.close()
28
29
30async def _get_lookup_rows(music: MusicController, item_id: int | str) -> set[tuple[str, str]]:
31 """Return the (external_id_type, external_id) lookup rows stored for a track."""
32 return {
33 (row["external_id_type"], row["external_id"])
34 for row in await music.database.get_rows(
35 DB_TABLE_EXTERNAL_ID_LOOKUP, {"media_type": "track", "item_id": int(item_id)}
36 )
37 }
38
39
40async def test_same_isrc_from_two_providers_dedupes(music: MusicController) -> None:
41 """Two providers exposing the same track with an identical ISRC merge into one item."""
42 library_track_1 = await music.tracks.add_item_to_library(create_track("spotify_1", "track_abc"))
43 library_track_2 = await music.tracks.add_item_to_library(create_track("tidal_1", "track_xyz"))
44
45 assert library_track_1.item_id == library_track_2.item_id
46 assert len(library_track_2.provider_mappings) == 2
47 assert await music.tracks.library_count() == 1
48
49
50async def test_get_library_item_by_external_id(music: MusicController) -> None:
51 """Library items resolve by external id, both typed and untyped."""
52 track = create_track("spotify_1", "track_abc")
53 track.external_ids.add((ExternalID.MB_RECORDING, MBID))
54 library_track = await music.tracks.add_item_to_library(track)
55
56 # typed lookup
57 match = await music.tracks.get_library_item_by_external_id(ISRC, ExternalID.ISRC)
58 assert match is not None
59 assert match.item_id == library_track.item_id
60 match = await music.tracks.get_library_item_by_external_id(MBID, ExternalID.MB_RECORDING)
61 assert match is not None
62 assert match.item_id == library_track.item_id
63 # untyped lookup
64 match = await music.tracks.get_library_item_by_external_id(ISRC)
65 assert match is not None
66 assert match.item_id == library_track.item_id
67 # matching is case-insensitive (as the previous LIKE based scan was)
68 match = await music.tracks.get_library_item_by_external_id(ISRC.lower(), ExternalID.ISRC)
69 assert match is not None
70 assert match.item_id == library_track.item_id
71 # no (partial) match on wrong type or unknown id
72 assert await music.tracks.get_library_item_by_external_id(ISRC, ExternalID.BARCODE) is None
73 assert await music.tracks.get_library_item_by_external_id("something-else") is None
74 assert await music.tracks.get_library_item_by_external_id(ISRC[:-1]) is None
75
76
77async def test_external_id_lookup_rows_follow_item_updates(music: MusicController) -> None:
78 """The lookup rows are kept in sync when an item is updated or removed."""
79 library_track = await music.tracks.add_item_to_library(create_track("spotify_1", "track_abc"))
80 assert await _get_lookup_rows(music, library_track.item_id) == {(str(ExternalID.ISRC), ISRC)}
81
82 # an update merges in newly discovered external ids
83 update = create_track("spotify_1", "track_abc")
84 update.external_ids.add((ExternalID.MB_RECORDING, MBID))
85 updated = await music.tracks.update_item_in_library(library_track.item_id, update)
86 # the item's external_ids attribute is reconstructed from the lookup table on read
87 assert updated.external_ids == {(ExternalID.ISRC, ISRC), (ExternalID.MB_RECORDING, MBID)}
88 assert await _get_lookup_rows(music, library_track.item_id) == {
89 (str(ExternalID.ISRC), ISRC),
90 (str(ExternalID.MB_RECORDING), MBID),
91 }
92
93 # an overwrite update replaces the lookup rows
94 await music.tracks.update_item_in_library(
95 library_track.item_id,
96 create_track("spotify_1", "track_abc", isrc="GBUM71029604"),
97 overwrite=True,
98 )
99 assert await _get_lookup_rows(music, library_track.item_id) == {
100 (str(ExternalID.ISRC), "GBUM71029604")
101 }
102 assert await music.tracks.get_library_item_by_external_id(ISRC) is None
103
104 # removal cleans up the lookup rows
105 await music.tracks.remove_item_from_library(library_track.item_id)
106 assert await _get_lookup_rows(music, library_track.item_id) == set()
107