/
/
/
1"""Tests that imported album tracks are filed under the album they came from."""
2
3from __future__ import annotations
4
5import datetime
6from typing import TYPE_CHECKING
7from unittest.mock import AsyncMock, MagicMock, patch
8
9from music_assistant_models.enums import ProviderType
10from music_assistant_models.media_items import (
11 Album,
12 AudioFormat,
13 ItemMapping,
14 ProviderMapping,
15 Track,
16)
17
18from music_assistant.constants import CONF_ENTRY_LIBRARY_SYNC_ALBUM_TRACKS, CONF_LOG_LEVEL
19from music_assistant.controllers.music.media.base import TrackSyncDetails
20from music_assistant.models.music_provider import MusicProvider
21
22if TYPE_CHECKING:
23 from collections.abc import Sequence
24
25ALBUM_ID = "album_1"
26
27
28class AlbumTracksProvider(MusicProvider):
29 """Provider that leaves the parent album off the tracks in its album listing."""
30
31 album_tracks: list[Track]
32 album: Album
33 get_album_calls: int = 0
34
35 async def get_album_tracks(self, prov_album_id: str) -> list[Track]:
36 """Return the configured album tracks."""
37 return self.album_tracks
38
39 async def get_album(self, prov_album_id: str) -> Album:
40 """Return the configured album and record the call."""
41 self.get_album_calls += 1
42 return self.album
43
44
45def _provider_mapping(item_id: str) -> ProviderMapping:
46 """Return a provider mapping for the test provider instance."""
47 return ProviderMapping(
48 item_id=item_id,
49 provider_domain="test",
50 provider_instance="test--1",
51 audio_format=AudioFormat(),
52 )
53
54
55def _build_album() -> Album:
56 """Return the provider album the imported tracks belong to."""
57 return Album(
58 item_id=ALBUM_ID,
59 provider="test",
60 name="Album One",
61 provider_mappings={_provider_mapping(ALBUM_ID)},
62 )
63
64
65def _build_tracks(count: int = 3) -> list[Track]:
66 """Return provider tracks without an album."""
67 return [
68 Track(
69 item_id=f"track_{index}",
70 provider="test",
71 name=f"Track {index}",
72 provider_mappings={_provider_mapping(f"track_{index}")},
73 )
74 for index in range(1, count + 1)
75 ]
76
77
78def _build_mass(sync_details: TrackSyncDetails | None = None) -> MagicMock:
79 """Return a mocked mass whose track controller records every imported track."""
80 mass = MagicMock()
81 tracks = mass.music.tracks
82 tracks.get_library_item_sync_details = AsyncMock(return_value=sync_details)
83
84 library_track = MagicMock()
85 library_track.item_id = 1
86 tracks.add_item_to_library = AsyncMock(return_value=library_track)
87 tracks.update_item_in_library = AsyncMock(return_value=library_track)
88 mass.music.genres.sync_media_item_genres = AsyncMock()
89 return mass
90
91
92def _build_provider(mass: MagicMock) -> AlbumTracksProvider:
93 """Return a provider instance wired to the given (mocked) mass."""
94 manifest = MagicMock()
95 manifest.type = ProviderType.MUSIC
96 manifest.domain = "test"
97 config = MagicMock()
98 config.instance_id = "test--1"
99 config.domain = "test"
100 values = {
101 CONF_LOG_LEVEL: "GLOBAL",
102 CONF_ENTRY_LIBRARY_SYNC_ALBUM_TRACKS.key: True,
103 }
104 config.get_value.side_effect = lambda key, default=None: values.get(key, default)
105 provider = AlbumTracksProvider(mass, manifest, config)
106 provider.album = _build_album()
107 provider.album_tracks = _build_tracks()
108 return provider
109
110
111def _added_tracks(mass: MagicMock) -> Sequence[Track]:
112 """Return the tracks that were added to the library."""
113 return [call.args[0] for call in mass.music.tracks.add_item_to_library.await_args_list]
114
115
116async def test_import_attaches_the_parent_album() -> None:
117 """A track whose provider listing omits the album is still filed under that album."""
118 mass = _build_mass()
119 provider = _build_provider(mass)
120
121 await provider.import_album_tracks(ALBUM_ID, provider.album)
122
123 added = _added_tracks(mass)
124 assert len(added) == 3
125 for track in added:
126 assert track.album is not None
127 assert track.album.item_id == ALBUM_ID
128 # the album was supplied by the caller, so it was not fetched again
129 assert provider.get_album_calls == 0
130
131
132async def test_album_is_resolved_once_when_not_supplied() -> None:
133 """Without a supplied album the provider is asked for it once, not per track."""
134 mass = _build_mass()
135 provider = _build_provider(mass)
136
137 await provider.import_album_tracks(ALBUM_ID)
138
139 assert provider.get_album_calls == 1
140 assert all(track.album is not None for track in _added_tracks(mass))
141
142
143async def test_tracks_that_already_have_an_album_keep_it() -> None:
144 """Tracks that come in with their own album are left untouched."""
145 mass = _build_mass()
146 provider = _build_provider(mass)
147 own_album = ItemMapping.from_item(
148 Album(
149 item_id="other_album",
150 provider="test",
151 name="Other Album",
152 provider_mappings={_provider_mapping("other_album")},
153 )
154 )
155 for track in provider.album_tracks:
156 track.album = own_album
157
158 await provider.import_album_tracks(ALBUM_ID)
159
160 assert provider.get_album_calls == 0
161 assert all(track.album is own_album for track in _added_tracks(mass))
162
163
164async def test_missing_album_link_is_backfilled() -> None:
165 """An earlier import that stored a track without its album is repaired."""
166 sync_details = TrackSyncDetails(
167 item_id=1,
168 favorite=False,
169 date_added=datetime.datetime(2026, 1, 1, tzinfo=datetime.UTC),
170 provider_mappings=set(),
171 has_album=False,
172 has_artists=True,
173 )
174 mass = _build_mass(sync_details)
175 provider = _build_provider(mass)
176
177 # the provider mappings already match, so the missing album link is the only trigger
178 with patch.object(provider, "_check_provider_mappings", return_value=True):
179 await provider.import_album_tracks(ALBUM_ID, provider.album)
180
181 assert mass.music.tracks.update_item_in_library.await_count == 3
182 for call in mass.music.tracks.update_item_in_library.await_args_list:
183 assert call.args[1].album.item_id == ALBUM_ID
184