/
/
/
1"""Tests for the tracks controller."""
2
3from collections.abc import AsyncGenerator
4from typing import Any
5from unittest.mock import AsyncMock, MagicMock, patch
6
7import pytest
8from music_assistant_models.media_items import Artist, ProviderMapping, UniqueList
9
10from music_assistant.controllers.music import MusicController
11from music_assistant.mass import MusicAssistant
12
13from .helpers import create_track
14
15
16@pytest.fixture
17async def music(mass_minimal: MusicAssistant) -> AsyncGenerator[MusicController]:
18 """Return a music controller attached to the minimal mass instance."""
19 controller = MusicController(mass_minimal)
20 mass_minimal.music = controller
21 yield controller
22 if controller._database:
23 await controller._database.close()
24
25
26@pytest.mark.asyncio
27async def test_explicit_filter_true_generates_sql(music: MusicController) -> None:
28 """Test that explicit=True generates correct SQL filter."""
29 captured_parts: list[str] = []
30
31 async def mock_query(*_args: Any, **kwargs: Any) -> list[Any]:
32 captured_parts.extend(kwargs.get("extra_query_parts", []))
33 return []
34
35 with patch.object(music.tracks, "get_library_items_by_query", mock_query):
36 await music.tracks.library_items(explicit=True, limit=10)
37 assert any(
38 "json_extract(tracks.metadata, '$.explicit') = 1" in part for part in captured_parts
39 )
40
41
42@pytest.mark.asyncio
43async def test_explicit_filter_false_generates_sql(music: MusicController) -> None:
44 """Test that explicit=False generates correct SQL filter."""
45 captured_parts: list[str] = []
46
47 async def mock_query(*_args: Any, **kwargs: Any) -> list[Any]:
48 captured_parts.extend(kwargs.get("extra_query_parts", []))
49 return []
50
51 with patch.object(music.tracks, "get_library_items_by_query", mock_query):
52 await music.tracks.library_items(explicit=False, limit=10)
53 assert any("IS NULL" in part and "= 0" in part for part in captured_parts)
54
55
56@pytest.mark.asyncio
57async def test_explicit_filter_none_generates_no_sql(music: MusicController) -> None:
58 """Test that explicit=None generates no explicit filter."""
59 captured_parts: list[str] = []
60
61 async def mock_query(*_args: Any, **kwargs: Any) -> list[Any]:
62 captured_parts.extend(kwargs.get("extra_query_parts", []))
63 return []
64
65 with patch.object(music.tracks, "get_library_items_by_query", mock_query):
66 await music.tracks.library_items(explicit=None, limit=10)
67 assert not any("explicit" in part.lower() for part in captured_parts)
68
69
70@pytest.mark.asyncio
71async def test_explicit_filter_default_is_none(music: MusicController) -> None:
72 """Test that omitting explicit parameter behaves like explicit=None."""
73 captured_parts: list[str] = []
74
75 async def mock_query(*_args: Any, **kwargs: Any) -> list[Any]:
76 captured_parts.extend(kwargs.get("extra_query_parts", []))
77 return []
78
79 with patch.object(music.tracks, "get_library_items_by_query", mock_query):
80 await music.tracks.library_items(limit=10)
81 assert not any("explicit" in part.lower() for part in captured_parts)
82
83
84@pytest.mark.asyncio
85async def test_by_prov_id_batches_item_ids_into_in_clause(music: MusicController) -> None:
86 """provider_item_ids builds a single parameterized IN (...) subquery."""
87 captured_parts: list[str] = []
88 captured_params: dict[str, Any] = {}
89
90 async def mock_query(*_args: Any, **kwargs: Any) -> list[Any]:
91 captured_parts.extend(kwargs.get("extra_query_parts", []))
92 captured_params.update(kwargs.get("extra_query_params", {}))
93 return []
94
95 with patch.object(music.tracks, "get_library_items_by_query", mock_query):
96 await music.tracks.get_library_items_by_prov_id(
97 provider_instance_id_or_domain="spotify",
98 provider_item_ids=["x", "y", "z"],
99 )
100
101 subquery = " ".join(captured_parts)
102 assert "provider_mappings.provider_item_id IN (:item_id_0, :item_id_1, :item_id_2)" in subquery
103 assert captured_params["prov_id"] == "spotify"
104 assert [captured_params[f"item_id_{i}"] for i in range(3)] == ["x", "y", "z"]
105
106
107@pytest.mark.asyncio
108async def test_by_prov_id_empty_item_ids_matches_nothing(music: MusicController) -> None:
109 """An explicit empty provider_item_ids returns [] (not the whole provider library)."""
110 ran = False
111
112 async def mock_query(*_args: Any, **_kwargs: Any) -> list[Any]:
113 nonlocal ran
114 ran = True
115 return []
116
117 with patch.object(music.tracks, "get_library_items_by_query", mock_query):
118 result = await music.tracks.get_library_items_by_prov_id(
119 provider_instance_id_or_domain="spotify", provider_item_ids=[]
120 )
121
122 assert result == []
123 assert ran is False # short-circuits before it can build an unconstrained query
124
125
126async def test_match_provider_uses_full_track_mapping(music: MusicController) -> None:
127 """Provider matching stores mapping details from the fetched track."""
128 base_track = create_track("spotify_1", "base")
129 search_track = create_track("qobuz_1", "candidate")
130 full_track = create_track("qobuz_1", "candidate")
131 full_track.provider_mappings = {
132 ProviderMapping(
133 item_id="candidate",
134 provider_domain="qobuz",
135 provider_instance="qobuz_1",
136 url="https://provider.example/full",
137 )
138 }
139 provider = MagicMock()
140 provider.name = "Qobuz"
141 provider.domain = "qobuz"
142
143 with (
144 patch.object(music.tracks, "search", AsyncMock(return_value=[search_track])),
145 patch.object(
146 music.tracks,
147 "get_provider_item",
148 AsyncMock(return_value=full_track),
149 ),
150 patch(
151 "music_assistant.controllers.music.media.tracks.compare_media_item",
152 return_value=True,
153 ),
154 patch(
155 "music_assistant.controllers.music.media.tracks.compare_track",
156 return_value=True,
157 ),
158 ):
159 mappings = await music.tracks.match_provider(base_track, provider, ref_albums=[])
160
161 assert mappings == list(full_track.provider_mappings)
162
163
164async def test_overwrite_update_keeps_artists_when_none_are_given(
165 mass: MusicAssistant, caplog: pytest.LogCaptureFixture
166) -> None:
167 """An overwrite update carrying no artists must not clear the stored ones."""
168 db_track = await mass.music.tracks.add_item_to_library(create_track("spotify_1", "track1"))
169
170 update = create_track("spotify_1", "track1")
171 update.artists = UniqueList()
172 await mass.music.tracks.update_item_in_library(db_track.item_id, update, overwrite=True)
173
174 refreshed = await mass.music.tracks.get_library_item(db_track.item_id)
175 assert [artist.name for artist in refreshed.artists] == ["Test Artist"]
176 assert "Ignoring request to clear all artists" in caplog.text
177
178
179async def test_overwrite_update_replaces_artists(mass: MusicAssistant) -> None:
180 """An overwrite update carrying artists still replaces the stored ones."""
181 db_track = await mass.music.tracks.add_item_to_library(create_track("spotify_1", "track1"))
182
183 update = create_track("spotify_1", "track1")
184 update.artists = UniqueList(
185 [
186 Artist(
187 item_id="other_artist",
188 provider="spotify_1",
189 name="Other Artist",
190 provider_mappings={
191 ProviderMapping(
192 item_id="other_artist",
193 provider_domain="spotify",
194 provider_instance="spotify_1",
195 )
196 },
197 )
198 ]
199 )
200 await mass.music.tracks.update_item_in_library(db_track.item_id, update, overwrite=True)
201
202 refreshed = await mass.music.tracks.get_library_item(db_track.item_id)
203 assert [artist.name for artist in refreshed.artists] == ["Other Artist"]
204