/
/
/
1"""
2Tests for the player-queue artist track selection (``get_artist_tracks``).
3
4A library-only artist (no streaming provider mappings) stands in for a file-only
5library: with the bare ``mass`` fixture (no streaming/metadata providers loaded),
6its top-tracks and per-provider discography legs resolve to nothing, so only the
7in-library leg can yield tracks. The integration tests use that fixture (a real
8MusicAssistant with a real SQLite database); the unit tests at the bottom exercise
9the multi-source union/dedup directly with mocked sources.
10"""
11
12from __future__ import annotations
13
14from typing import TYPE_CHECKING, cast
15from unittest.mock import AsyncMock, MagicMock
16from uuid import uuid4
17
18from music_assistant_models.media_items import Artist, ProviderMapping, Track
19from music_assistant_models.unique_list import UniqueList
20
21from music_assistant.controllers.player_queues.constants import (
22 CONF_DEFAULT_ENQUEUE_SELECT_ARTIST,
23)
24from music_assistant.controllers.player_queues.media_resolver import MediaResolver
25
26if TYPE_CHECKING:
27 from music_assistant.mass import MusicAssistant
28
29
30def _library_mapping() -> set[ProviderMapping]:
31 """Create a single library provider mapping with a unique provider item id."""
32 return {
33 ProviderMapping(
34 item_id=uuid4().hex,
35 provider_domain="library",
36 provider_instance="library",
37 in_library=True,
38 )
39 }
40
41
42async def _add_artist(mass: MusicAssistant, name: str) -> Artist:
43 """Add a minimal library artist and return the stored item."""
44 return await mass.music.artists.add_item_to_library(
45 Artist(item_id="0", provider="library", name=name, provider_mappings=_library_mapping())
46 )
47
48
49async def _add_track(mass: MusicAssistant, name: str, artists: list[Artist]) -> Track:
50 """Add a minimal library track credited to the given artists and return the stored item."""
51 added = await mass.music.tracks.add_item_to_library(
52 Track(
53 item_id="0",
54 provider="library",
55 name=name,
56 provider_mappings=_library_mapping(),
57 artists=UniqueList(artists),
58 )
59 )
60 return await mass.music.tracks.get_library_item(added.item_id)
61
62
63def _select(mass: MusicAssistant, value: str) -> None:
64 """Set the artist enqueue-selection option."""
65 mass.config.set_raw_core_config_value(
66 "player_queues", CONF_DEFAULT_ENQUEUE_SELECT_ARTIST, value
67 )
68
69
70async def test_default_resolves_library_only_artist(mass: MusicAssistant) -> None:
71 """The default option resolves a library-only artist to its in-library tracks."""
72 artist = await _add_artist(mass, "ABBA")
73 track_a = await _add_track(mass, "Dancing Queen", [artist])
74 track_b = await _add_track(mass, "Mamma Mia", [artist])
75
76 result = await mass.player_queues._media_resolver.get_artist_tracks(artist)
77
78 assert {t.name for t in result} == {track_a.name, track_b.name}
79
80
81async def test_prefer_library_returns_library_tracks(mass: MusicAssistant) -> None:
82 """prefer_library returns the artist's in-library tracks when present."""
83 artist = await _add_artist(mass, "ABBA")
84 track = await _add_track(mass, "Waterloo", [artist])
85 _select(mass, "prefer_library")
86
87 result = await mass.player_queues._media_resolver.get_artist_tracks(artist)
88
89 assert {t.name for t in result} == {track.name}
90
91
92async def test_legacy_all_album_tracks_maps_to_all_tracks(mass: MusicAssistant) -> None:
93 """A stored legacy 'all_album_tracks' resolves like 'all_tracks' (no empty result)."""
94 artist = await _add_artist(mass, "ABBA")
95 track = await _add_track(mass, "SOS", [artist])
96 _select(mass, "all_album_tracks")
97
98 result = await mass.player_queues._media_resolver.get_artist_tracks(artist)
99
100 assert {t.name for t in result} == {track.name}
101
102
103async def test_legacy_library_album_tracks_maps_to_library_tracks(mass: MusicAssistant) -> None:
104 """A stored legacy 'library_album_tracks' resolves like 'library_tracks'."""
105 artist = await _add_artist(mass, "ABBA")
106 track = await _add_track(mass, "Fernando", [artist])
107 _select(mass, "library_album_tracks")
108
109 result = await mass.player_queues._media_resolver.get_artist_tracks(artist)
110
111 assert {t.name for t in result} == {track.name}
112
113
114async def test_top_tracks_excludes_library_tracks(mass: MusicAssistant) -> None:
115 """top_tracks returns only provider top tracks, never the in-library tracks."""
116 artist = await _add_artist(mass, "ABBA")
117 await _add_track(mass, "Honey Honey", [artist])
118 _select(mass, "top_tracks")
119
120 result = await mass.player_queues._media_resolver.get_artist_tracks(artist)
121
122 assert "Honey Honey" not in {t.name for t in result}
123
124
125# --- unit tests for the all_tracks union and provider-tracks helper (mocked, no DB) ---
126
127
128def _track_obj(name: str, version: str = "") -> Track:
129 """Create a minimal track with the given name and version (for dedup-key tests)."""
130 return Track(item_id=name, provider="test", name=name, version=version, provider_mappings=set())
131
132
133def _prov_mapping(instance: str, item_id: str) -> ProviderMapping:
134 """Create a provider mapping for the given (streaming) provider instance."""
135 return ProviderMapping(
136 item_id=item_id, provider_domain="test", provider_instance=instance, in_library=False
137 )
138
139
140def _artist_obj(mappings: set[ProviderMapping] | None = None) -> Artist:
141 """Create a minimal library artist for the resolution unit tests."""
142 return Artist(item_id="1", provider="library", name="ABBA", provider_mappings=mappings or set())
143
144
145def _fake_queues(selection: str) -> MagicMock:
146 """Create a mock standing in for the media resolver, with the artist option preselected."""
147 fake = MagicMock()
148 fake.mass.config.get_raw_core_config_value = MagicMock(return_value=selection)
149 return fake
150
151
152async def test_all_tracks_unions_and_dedups_sources() -> None:
153 """all_tracks merges the in-library and provider tracks, deduplicated by name+version."""
154 fake = _fake_queues("all_tracks")
155 fake._library_artist_tracks = AsyncMock(
156 return_value=[_track_obj("Shared"), _track_obj("LibOnly")]
157 )
158 fake._provider_artist_tracks = AsyncMock(
159 return_value=[_track_obj("Shared"), _track_obj("ProvOnly")]
160 )
161
162 result = await MediaResolver.get_artist_tracks(cast("MediaResolver", fake), _artist_obj())
163
164 assert sorted(t.name for t in result) == ["LibOnly", "ProvOnly", "Shared"]
165
166
167async def test_all_tracks_keeps_distinct_versions() -> None:
168 """Tracks sharing a name but differing in version are both kept."""
169 fake = _fake_queues("all_tracks")
170 fake._library_artist_tracks = AsyncMock(return_value=[_track_obj("Song", "")])
171 fake._provider_artist_tracks = AsyncMock(return_value=[_track_obj("Song", "Remix")])
172
173 result = await MediaResolver.get_artist_tracks(cast("MediaResolver", fake), _artist_obj())
174
175 assert sorted(t.version for t in result) == ["", "Remix"]
176
177
178async def test_all_tracks_survives_a_failing_source() -> None:
179 """A failing source (e.g. a flaky provider) is dropped; the in-library tracks still resolve."""
180 fake = _fake_queues("all_tracks")
181 fake._library_artist_tracks = AsyncMock(return_value=[_track_obj("Local")])
182 fake._provider_artist_tracks = AsyncMock(side_effect=RuntimeError("provider down"))
183
184 result = await MediaResolver.get_artist_tracks(cast("MediaResolver", fake), _artist_obj())
185
186 assert {t.name for t in result} == {"Local"}
187 fake.logger.warning.assert_called_once()
188
189
190async def test_prefer_library_falls_back_to_top_tracks() -> None:
191 """prefer_library falls back to top tracks when the artist has no in-library tracks."""
192 fake = _fake_queues("prefer_library")
193 fake._library_artist_tracks = AsyncMock(return_value=[])
194 fake.mass.music.artists.top_tracks = AsyncMock(return_value=[_track_obj("Top")])
195
196 result = await MediaResolver.get_artist_tracks(cast("MediaResolver", fake), _artist_obj())
197
198 assert {t.name for t in result} == {"Top"}
199
200
201async def test_provider_artist_tracks_respects_unique_providers() -> None:
202 """_provider_artist_tracks queries only unique providers and aggregates their tracks."""
203 fake = _fake_queues("all_tracks")
204 artist = _artist_obj({_prov_mapping("p1", "a1"), _prov_mapping("p2", "a2")})
205 fake.mass.music.get_unique_providers = MagicMock(return_value=["p1"]) # p2 filtered out
206 fake.mass.music.artists.tracks = AsyncMock(return_value=[_track_obj("Dancing Queen")])
207
208 result = await MediaResolver._provider_artist_tracks(cast("MediaResolver", fake), artist)
209
210 assert [t.name for t in result] == ["Dancing Queen"]
211 # only the unique provider is queried (the duplicate streaming domain is skipped)
212 assert fake.mass.music.artists.tracks.await_count == 1
213