/
/
/
1"""Tests for the music controller."""
2
3import sqlite3
4from collections.abc import AsyncGenerator
5from typing import Any
6from unittest.mock import AsyncMock, MagicMock, patch
7
8import pytest
9from music_assistant_models.config_entries import ConfigActionResult
10from music_assistant_models.enums import MediaType
11from music_assistant_models.errors import (
12 MediaNotFoundError,
13 MusicAssistantError,
14 UnsupportedFeaturedException,
15)
16from music_assistant_models.media_items import Radio, SoundEffect, Track
17
18from music_assistant.constants import VACUUM_MIN_RECLAIM_RATIO
19from music_assistant.controllers.music import MusicController
20from music_assistant.controllers.music.constants import CONF_RESET_DB
21from music_assistant.controllers.music.database import MusicDatabaseSetupMixin
22from music_assistant.helpers.database import DatabaseConnection
23from music_assistant.mass import MusicAssistant
24
25
26@pytest.fixture
27async def music(mass_minimal: MusicAssistant) -> AsyncGenerator[MusicController]:
28 """Return a music controller attached to the minimal mass instance."""
29 controller = MusicController(mass_minimal)
30 mass_minimal.music = controller
31 yield controller
32 # close the db connection so its worker thread does not outlive the test
33 if controller._database:
34 await controller._database.close()
35
36
37async def test_setup_fails_clearly_without_fts5_support(music: MusicController) -> None:
38 """Missing FTS5 trigram support surfaces as a clear error instead of a raw SQL error."""
39 orig_execute = DatabaseConnection.execute
40
41 async def fake_execute(
42 self: DatabaseConnection, query: str, values: dict[str, Any] | None = None
43 ) -> Any:
44 if "USING fts5" in query:
45 raise sqlite3.OperationalError("no such tokenizer: trigram")
46 return await orig_execute(self, query, values)
47
48 with (
49 patch.object(DatabaseConnection, "execute", fake_execute),
50 pytest.raises(MusicAssistantError, match=r"SQLite 3\.34"),
51 ):
52 await music._setup_database()
53
54
55async def test_setup_skips_vacuum_when_little_reclaimable(music: MusicController) -> None:
56 """Test that the library db startup vacuum is skipped when little can be reclaimed."""
57 with (
58 patch.object(
59 DatabaseConnection,
60 "get_reclaimable_ratio",
61 AsyncMock(return_value=VACUUM_MIN_RECLAIM_RATIO / 2),
62 ),
63 patch.object(DatabaseConnection, "vacuum", AsyncMock()) as mock_vacuum,
64 ):
65 await music._setup_database()
66 mock_vacuum.assert_not_called()
67
68
69async def test_setup_runs_vacuum_when_reclaimable(music: MusicController) -> None:
70 """Test that the library db startup vacuum runs when enough space can be reclaimed."""
71 with (
72 patch.object(
73 DatabaseConnection,
74 "get_reclaimable_ratio",
75 AsyncMock(return_value=VACUUM_MIN_RECLAIM_RATIO + 0.1),
76 ),
77 patch.object(DatabaseConnection, "vacuum", AsyncMock()) as mock_vacuum,
78 ):
79 await music._setup_database()
80 mock_vacuum.assert_awaited_once_with()
81
82
83async def test_reset_db_action_reports_its_outcome(music: MusicController) -> None:
84 """The reset-database action reports its outcome instead of re-rendering the config form."""
85 with (
86 patch.object(music, "_reset_database", AsyncMock()) as mock_reset,
87 patch.object(music.mass.cache, "clear", AsyncMock()) as mock_clear,
88 patch.object(music, "start_sync", AsyncMock()) as mock_sync,
89 ):
90 result = await music.handle_config_action(CONF_RESET_DB)
91
92 mock_reset.assert_awaited_once()
93 mock_clear.assert_awaited_once()
94 mock_sync.assert_awaited_once()
95 assert isinstance(result, ConfigActionResult)
96 assert result.translation_key == f"{CONF_RESET_DB}.result"
97 assert result.open_url is None
98
99
100def test_database_setup_mixin_applied() -> None:
101 """The database setup/migration logic is provided to the controller via the mixin."""
102 assert issubclass(MusicController, MusicDatabaseSetupMixin)
103 # the schema/migration entrypoints resolve on the controller
104 assert hasattr(MusicController, "_setup_database")
105 assert hasattr(MusicController, "_reset_database")
106 assert hasattr(MusicController, "_cleanup_database")
107
108
109async def test_cleanup_database_runs_on_empty_library(music: MusicController) -> None:
110 """Database maintenance/cleanup completes without error on a freshly created library."""
111 await music._setup_database()
112 # should be a no-op on an empty database, but must run end-to-end against a real connection
113 await music._cleanup_database()
114
115
116async def test_missing_library_artist_lookup_on_empty_library(music: MusicController) -> None:
117 """A background lookup for a stale library artist completes without an exception."""
118 await music._setup_database()
119
120 with pytest.raises(MediaNotFoundError):
121 await music.artists.get_library_item("27")
122
123 task = music.mass.create_task(
124 music.get_library_item_by_prov_id(MediaType.ARTIST, "27", "library")
125 )
126
127 assert await task is None
128
129
130def _sound_effect() -> SoundEffect:
131 """Return a minimal SoundEffect item."""
132 return SoundEffect(
133 item_id="rain.mp3",
134 provider="filesystem_local--test",
135 name="rain",
136 provider_mappings=set(),
137 )
138
139
140async def test_add_to_favorites_rejects_sound_effect() -> None:
141 """add_item_to_favorites raises UnsupportedFeaturedException for SOUND_EFFECT."""
142 controller = MusicController.__new__(MusicController)
143 controller.mass = MagicMock()
144 with pytest.raises(UnsupportedFeaturedException, match="can not be favorites"):
145 await controller.add_item_to_favorites(_sound_effect())
146
147
148async def test_add_to_library_rejects_sound_effect() -> None:
149 """add_item_to_library raises UnsupportedFeaturedException for SOUND_EFFECT."""
150 controller = MusicController.__new__(MusicController)
151 controller.mass = MagicMock()
152 controller.get_item = AsyncMock(return_value=_sound_effect()) # type: ignore[method-assign]
153 with pytest.raises(UnsupportedFeaturedException, match="can not be library items"):
154 await controller.add_item_to_library(_sound_effect())
155
156
157async def test_add_to_favorites_rejects_stale_sound_effect_uri() -> None:
158 """A stale sound-effect URI (provider unloaded) raises the honest error."""
159 controller = MusicController.__new__(MusicController)
160 controller.mass = MagicMock()
161 with pytest.raises(UnsupportedFeaturedException, match="can not be favorites"):
162 await controller.add_item_to_favorites("stale_provider://sound_effect/rain.mp3")
163
164
165async def test_add_to_library_rejects_stale_sound_effect_uri() -> None:
166 """A stale sound-effect URI (provider unloaded) raises the honest error."""
167 controller = MusicController.__new__(MusicController)
168 controller.mass = MagicMock()
169 with pytest.raises(UnsupportedFeaturedException, match="can not be library items"):
170 await controller.add_item_to_library("stale_provider://sound_effect/rain.mp3")
171
172
173async def test_get_item_passes_sound_effect_type_to_builtin_provider() -> None:
174 """Builtin instance URIs keep the parsed SOUND_EFFECT media type."""
175 controller = MusicController.__new__(MusicController)
176 sound_effect = _sound_effect()
177 builtin_provider = MagicMock()
178 builtin_provider.domain = "builtin"
179 builtin_provider.parse_item = AsyncMock(return_value=sound_effect)
180 controller.mass = MagicMock()
181 controller.mass.get_provider.side_effect = lambda provider_id: (
182 builtin_provider if provider_id in {"builtin", "builtin_1"} else None
183 )
184
185 result = await controller.get_item(
186 MediaType.SOUND_EFFECT,
187 "http://example.com/intro.mp3",
188 "builtin_1",
189 )
190
191 assert result is sound_effect
192 builtin_provider.parse_item.assert_awaited_once_with(
193 "http://example.com/intro.mp3",
194 requested_media_type=MediaType.SOUND_EFFECT,
195 )
196
197
198async def test_get_item_builtin_radio_keeps_user_supplied_details() -> None:
199 """A builtin radio URI resolves through get_radio so it stays a radio station."""
200 controller = MusicController.__new__(MusicController)
201 radio = Radio(
202 item_id="http://stream.example.com",
203 provider="builtin",
204 name="My Station",
205 provider_mappings=set(),
206 )
207 builtin_provider = MagicMock()
208 builtin_provider.domain = "builtin"
209 builtin_provider.parse_item = AsyncMock()
210 builtin_provider.get_radio = AsyncMock(return_value=radio)
211 controller.mass = MagicMock()
212 controller.mass.get_provider.side_effect = lambda provider_id: (
213 builtin_provider if provider_id in {"builtin", "builtin_1"} else None
214 )
215
216 result = await controller.get_item(
217 MediaType.RADIO,
218 "http://stream.example.com",
219 "builtin_1",
220 )
221
222 assert result is radio
223 builtin_provider.get_radio.assert_awaited_once_with("http://stream.example.com")
224 builtin_provider.parse_item.assert_not_awaited()
225
226
227async def test_get_item_builtin_track_keeps_user_supplied_details() -> None:
228 """A builtin track URI resolves through get_track so it stays a track."""
229 controller = MusicController.__new__(MusicController)
230 track = Track(
231 item_id="http://media.example.com/song.mp3",
232 provider="builtin",
233 name="My Song",
234 provider_mappings=set(),
235 )
236 builtin_provider = MagicMock()
237 builtin_provider.domain = "builtin"
238 builtin_provider.parse_item = AsyncMock()
239 builtin_provider.get_track = AsyncMock(return_value=track)
240 controller.mass = MagicMock()
241 controller.mass.get_provider.side_effect = lambda provider_id: (
242 builtin_provider if provider_id in {"builtin", "builtin_1"} else None
243 )
244
245 result = await controller.get_item(
246 MediaType.TRACK,
247 "http://media.example.com/song.mp3",
248 "builtin_1",
249 )
250
251 assert result is track
252 builtin_provider.get_track.assert_awaited_once_with("http://media.example.com/song.mp3")
253 builtin_provider.parse_item.assert_not_awaited()
254
255
256async def test_get_item_builtin_unknown_type_still_parses_url() -> None:
257 """A plain URL of unknown media type is probed by the builtin provider."""
258 controller = MusicController.__new__(MusicController)
259 radio = Radio(
260 item_id="http://stream.example.com",
261 provider="builtin",
262 name="My Station",
263 provider_mappings=set(),
264 )
265 builtin_provider = MagicMock()
266 builtin_provider.domain = "builtin"
267 builtin_provider.parse_item = AsyncMock(return_value=radio)
268 builtin_provider.get_radio = AsyncMock()
269 controller.mass = MagicMock()
270 controller.mass.get_provider.side_effect = lambda provider_id: (
271 builtin_provider if provider_id in {"builtin", "builtin_1"} else None
272 )
273
274 result = await controller.get_item(
275 MediaType.UNKNOWN,
276 "http://stream.example.com",
277 "builtin_1",
278 )
279
280 assert result is radio
281 builtin_provider.parse_item.assert_awaited_once_with(
282 "http://stream.example.com",
283 requested_media_type=MediaType.UNKNOWN,
284 )
285 builtin_provider.get_radio.assert_not_awaited()
286
287
288async def test_get_item_builtin_playlist_instance_uses_playlist_controller() -> None:
289 """Builtin instance playlists should still resolve through the playlist controller."""
290 controller = MusicController.__new__(MusicController)
291 builtin_provider = MagicMock()
292 builtin_provider.domain = "builtin"
293 builtin_provider.parse_item = AsyncMock()
294 playlist_controller = MagicMock()
295 playlist_item = MagicMock()
296 playlist_controller.get = AsyncMock(return_value=playlist_item)
297 controller.mass = MagicMock()
298 controller.mass.get_provider.side_effect = lambda provider_id: (
299 builtin_provider if provider_id in {"builtin", "builtin_1"} else None
300 )
301 controller.get_controller = MagicMock(return_value=playlist_controller) # type: ignore[method-assign]
302
303 result = await controller.get_item(
304 MediaType.PLAYLIST,
305 "playlist_123",
306 "builtin_1",
307 )
308
309 assert result is playlist_item
310 builtin_provider.parse_item.assert_not_awaited()
311 controller.get_controller.assert_called_once_with(MediaType.PLAYLIST)
312 playlist_controller.get.assert_awaited_once_with(
313 item_id="playlist_123",
314 provider_instance_id_or_domain="builtin_1",
315 allow_update_metadata=True,
316 )
317