/
/
1"""Tests for the Spotify Connect provider."""
2
3import json
4from collections.abc import AsyncGenerator
5from pathlib import Path
6from typing import Any
7from unittest.mock import AsyncMock, MagicMock, patch
8
9from music_assistant_models.config_entries import ConfigEntry, ProviderConfig
10from music_assistant_models.enums import ConfigEntryType, ProviderType
11
12from music_assistant.constants import CONF_CROSSFADE_DURATION
13from music_assistant.providers.spotify_connect import (
14 BACKEND_SOLOIST,
15 CONF_API_KEY,
16 CONF_BACKEND,
17 CONF_SOLOIST_CONSENT,
18 CONF_VOLUME_MODE,
19 SpotifyConnectProvider,
20)
21from music_assistant.providers.spotify_connect.base import (
22 AUDIO_QUALITY_HIGH,
23 AUDIO_QUALITY_LOSSLESS,
24)
25from music_assistant.providers.spotify_connect.go_librespot.backend import (
26 API_PORT_RANGE_END,
27 API_PORT_RANGE_START,
28 GoLibrespotBackend,
29)
30from music_assistant.providers.spotify_connect.go_librespot.client import GoLibrespotClient
31from music_assistant.providers.spotify_connect.models import BackendEvent, BackendEventType
32from music_assistant.providers.spotify_connect.provider import (
33 AUDIO_SOURCE_ID,
34 CONF_AUDIO_QUALITY,
35 CONF_LOUDNESS_NORMALIZATION,
36)
37from music_assistant.providers.spotify_connect.soloist.backend import (
38 VOLUME_MODE_SYNC_SPOTIFY,
39 SoloistBackend,
40)
41
42
43async def test_backend_start_probes_api_port_on_ipv4_loopback() -> None:
44 """The daemon API port is selected on the address go-librespot binds."""
45 backend = object.__new__(GoLibrespotBackend)
46 backend.mass = MagicMock()
47 backend.logger = MagicMock()
48 backend.mass.create_task.side_effect = lambda coroutine: coroutine.close()
49
50 with (
51 patch(
52 "music_assistant.providers.spotify_connect.go_librespot.backend"
53 ".get_go_librespot_binary",
54 return_value="/usr/bin/go-librespot",
55 ),
56 patch(
57 "music_assistant.providers.spotify_connect.go_librespot.backend.select_free_port",
58 new=AsyncMock(return_value=38801),
59 ) as select_port,
60 ):
61 await backend.start()
62
63 select_port.assert_awaited_once_with(API_PORT_RANGE_START, API_PORT_RANGE_END, host="127.0.0.1")
64 assert backend._client is not None
65 assert backend._client.base_url == "http://127.0.0.1:38801"
66
67
68async def test_daemon_runner_reselects_api_port_when_taken(tmp_path: Path) -> None:
69 """An API port taken while the daemon was down is replaced before (re)starting."""
70 backend = object.__new__(GoLibrespotBackend)
71 backend.mass = MagicMock()
72 backend.mass.streams.get_source_ip = AsyncMock(return_value="192.168.1.5")
73 backend.logger = MagicMock()
74 backend.name = "Spotify Test"
75 backend.cache_dir = str(tmp_path)
76 backend._binary = "/usr/bin/go-librespot"
77 backend._api_port = 38800
78 backend._client = GoLibrespotClient(backend.mass, "http://127.0.0.1:38800", backend.logger)
79 backend._event_callback = AsyncMock()
80 # exit the supervisor loop after a single iteration
81 backend._stop_called = True
82 backend._restart_error_count = 0
83
84 async def _no_stderr() -> AsyncGenerator[str]:
85 return
86 yield
87
88 proc = MagicMock()
89 proc.start = AsyncMock()
90 proc.close = AsyncMock()
91 proc.iter_stderr = _no_stderr
92
93 with (
94 patch(
95 "music_assistant.providers.spotify_connect.go_librespot.backend.is_port_in_use",
96 new=AsyncMock(return_value=True),
97 ) as port_probe,
98 patch(
99 "music_assistant.providers.spotify_connect.go_librespot.backend.select_free_port",
100 new=AsyncMock(return_value=38801),
101 ),
102 patch(
103 "music_assistant.providers.spotify_connect.go_librespot.backend.AsyncProcess",
104 return_value=proc,
105 ),
106 patch.object(GoLibrespotBackend, "_write_config") as write_config,
107 ):
108 await backend._daemon_runner()
109
110 port_probe.assert_awaited_once_with(38800, host="127.0.0.1")
111 assert backend._api_port == 38801
112 assert backend._client.base_url == "http://127.0.0.1:38801"
113 # the daemon config pins the advertisement to the player-facing interface
114 write_config.assert_called_once_with("192.168.1.5")
115
116
117def _volume_sync_provider(volume_level: int | None) -> tuple[SpotifyConnectProvider, AsyncMock]:
118 """Build a minimal provider whose linked player reports the given volume."""
119 provider = object.__new__(SpotifyConnectProvider)
120 provider.mass = MagicMock()
121 provider.logger = MagicMock()
122 provider._last_volume_sent = None
123 backend = MagicMock()
124 set_volume = AsyncMock()
125 backend.set_volume = set_volume
126 provider._backend = backend
127 player = MagicMock()
128 player.state.volume_level = volume_level
129 provider.mass.players.get_player.return_value = player
130 return provider, set_volume
131
132
133async def test_sync_player_volume_pushes_player_volume_to_backend() -> None:
134 """The player's volume is pushed to the backend and cached for echo dedupe."""
135 provider, set_volume = _volume_sync_provider(50)
136
137 await provider._sync_player_volume_to_spotify("player1")
138
139 set_volume.assert_awaited_once_with(50)
140 assert provider._last_volume_sent == 50
141
142
143async def test_sync_player_volume_pushes_when_cache_matches() -> None:
144 """The push is unconditional: the backend's volume resets between sessions."""
145 provider, set_volume = _volume_sync_provider(50)
146 provider._last_volume_sent = 50
147
148 await provider._sync_player_volume_to_spotify("player1")
149
150 set_volume.assert_awaited_once_with(50)
151
152
153async def test_sync_player_volume_skips_when_volume_unknown() -> None:
154 """No push happens when the player does not expose a volume level."""
155 provider, set_volume = _volume_sync_provider(None)
156
157 await provider._sync_player_volume_to_spotify("player1")
158
159 set_volume.assert_not_awaited()
160 assert provider._last_volume_sent is None
161
162
163async def test_sync_player_volume_restores_cache_on_failure() -> None:
164 """A failed push restores the dedupe cache so a retry is not wrongly deduped."""
165 provider, set_volume = _volume_sync_provider(50)
166 set_volume.side_effect = OSError("daemon gone")
167
168 await provider._sync_player_volume_to_spotify("player1")
169
170 assert provider._last_volume_sent is None
171
172
173def _tethered_provider() -> tuple[SpotifyConnectProvider, AsyncMock]:
174 """Build a provider tethered to queue 'player1' with an active (paused) Spotify session."""
175 provider = object.__new__(SpotifyConnectProvider)
176 provider.mass = MagicMock()
177 provider.logger = MagicMock()
178 backend = MagicMock()
179 deactivate = AsyncMock()
180 backend.deactivate = deactivate
181 provider._backend = backend
182 provider._active_player_id = "player1"
183 provider._in_use_by_queue = None
184 provider._active_session_id = None
185 provider._spotify_session_active = True
186 provider._playing = False
187 provider._pending_pause_stop_task = None
188 provider._pending_play_media_task = None
189 return provider, deactivate
190
191
192async def test_queue_clear_releases_a_paused_spotify_session() -> None:
193 """Clearing the queue releases the session the paused stream's teardown left behind."""
194 provider, deactivate = _tethered_provider()
195
196 await provider.on_source_removed(AUDIO_SOURCE_ID, "player1")
197
198 deactivate.assert_awaited_once()
199
200
201async def test_queue_clear_releases_while_the_stream_is_winding_down() -> None:
202 """
203 A clear landing before the paused stream finished tearing down still releases.
204
205 The teardown itself releases nothing for a paused source, so waiting for it to hand the
206 claim back would leave the Spotify app tethered for good.
207 """
208 provider, deactivate = _tethered_provider()
209 provider._in_use_by_queue = "player1"
210
211 await provider.on_source_removed(AUDIO_SOURCE_ID, "player1")
212
213 deactivate.assert_awaited_once()
214
215
216async def test_clearing_another_queue_leaves_the_session_alone() -> None:
217 """Only the queue the source is tethered to may release it."""
218 provider, deactivate = _tethered_provider()
219
220 await provider.on_source_removed(AUDIO_SOURCE_ID, "player2")
221
222 deactivate.assert_not_awaited()
223
224
225async def test_queue_clear_without_an_active_session_does_nothing() -> None:
226 """There is nothing to release when MA is not the active Spotify device."""
227 provider, deactivate = _tethered_provider()
228 provider._spotify_session_active = False
229
230 await provider.on_source_removed(AUDIO_SOURCE_ID, "player1")
231
232 deactivate.assert_not_awaited()
233
234
235async def _session_inactive(provider: SpotifyConnectProvider) -> list[str]:
236 """Run the backend's 'session inactive' answer and return the players it wanted stopped."""
237 stopped: list[str] = []
238 provider._schedule_pause_stop = lambda player_id: stopped.append(player_id) # type: ignore[method-assign]
239 with patch.object(SpotifyConnectProvider, "name", "Spotify Test"):
240 await provider._handle_backend_event(BackendEvent(type=BackendEventType.SESSION_INACTIVE))
241 return stopped
242
243
244async def test_releasing_the_session_leaves_the_new_playback_alone() -> None:
245 """
246 Releasing must not stop the player that took the source's place.
247
248 The backend answers a release with the same "session inactive" it sends when the user picks
249 another device in the Spotify app - and that one does stop the player. By then this player is
250 playing whatever replaced the source, so stopping it would cut the music the user just started.
251 """
252 provider, _ = _tethered_provider()
253
254 with patch.object(SpotifyConnectProvider, "name", "Spotify Test"):
255 await provider.on_source_removed(AUDIO_SOURCE_ID, "player1")
256
257 assert await _session_inactive(provider) == []
258
259
260async def test_a_spotify_side_deselect_still_stops_the_player() -> None:
261 """Picking another device in the Spotify app does stop what MA was playing from it."""
262 provider, _ = _tethered_provider()
263
264 assert await _session_inactive(provider) == ["player1"]
265
266
267async def test_queue_clear_survives_a_failing_release() -> None:
268 """A backend that cannot be reached must not break clearing the queue."""
269 provider, deactivate = _tethered_provider()
270 deactivate.side_effect = OSError("daemon gone")
271
272 await provider.on_source_removed(AUDIO_SOURCE_ID, "player1")
273
274 deactivate.assert_awaited_once()
275
276
277async def test_a_transferred_source_follows_its_new_queue() -> None:
278 """A paused source moved to another player is tracked on the queue that took it over."""
279 provider, _ = _tethered_provider()
280
281 await provider.on_source_transferred(AUDIO_SOURCE_ID, "player1", "player2")
282
283 assert provider._active_player_id == "player2"
284
285
286async def test_a_transfer_of_another_queue_is_ignored() -> None:
287 """A transfer that does not involve the tethered queue leaves the tracking alone."""
288 provider, _ = _tethered_provider()
289
290 await provider.on_source_transferred(AUDIO_SOURCE_ID, "player3", "player2")
291
292 assert provider._active_player_id == "player1"
293
294
295async def test_clearing_a_transferred_queue_releases_the_session() -> None:
296 """
297 The queue a paused source was transferred to can release it.
298
299 Transferring a paused source never re-selects it on the target, so without the handover
300 the plugin would still be pointed at the queue it left and the release would not fire.
301 """
302 provider, deactivate = _tethered_provider()
303
304 await provider.on_source_transferred(AUDIO_SOURCE_ID, "player1", "player2")
305 await provider.on_source_removed(AUDIO_SOURCE_ID, "player2")
306
307 deactivate.assert_awaited_once()
308
309
310def _provider_with_stored_config(
311 setup_data: dict[str, Any], tmp_path: Path
312) -> SpotifyConnectProvider:
313 """Build a provider whose stored setup_data resolves through the real accessors."""
314 provider = object.__new__(SpotifyConnectProvider)
315 provider.mass = MagicMock()
316 provider.mass.storage_path = str(tmp_path / "storage")
317 provider.mass.cache_path = str(tmp_path / "cache")
318 provider.mass.config.get.return_value = setup_data
319 provider.mass.config.decrypt_string.side_effect = lambda value: value
320 provider.logger = MagicMock()
321 provider._publish_name = "Test Speaker"
322 provider.config = ProviderConfig(
323 values={},
324 type=ProviderType.PLUGIN,
325 domain="spotify_connect",
326 instance_id="spotify_connect--test",
327 name="Spotify Connect",
328 )
329 return provider
330
331
332def test_config_without_backend_choice_loads_go_librespot(tmp_path: Path) -> None:
333 """A config from before the backend choice existed loads go-librespot unchanged."""
334 provider = _provider_with_stored_config({}, tmp_path)
335
336 assert isinstance(provider._create_backend(), GoLibrespotBackend)
337
338
339def test_soloist_setup_data_loads_soloist_backend(tmp_path: Path) -> None:
340 """A flow-configured soloist instance loads the soloist backend with its stored values."""
341 provider = _provider_with_stored_config(
342 {
343 CONF_BACKEND: BACKEND_SOLOIST,
344 CONF_API_KEY: "soloist-api-key-0123456789abcdef",
345 CONF_SOLOIST_CONSENT: True,
346 },
347 tmp_path,
348 )
349 # the volume mode lives in the provider options, not in the setup data
350 provider.config.values[CONF_VOLUME_MODE] = ConfigEntry(
351 key=CONF_VOLUME_MODE,
352 type=ConfigEntryType.STRING,
353 value=VOLUME_MODE_SYNC_SPOTIFY,
354 )
355
356 provider.config.values[CONF_CROSSFADE_DURATION] = ConfigEntry(
357 key=CONF_CROSSFADE_DURATION,
358 type=ConfigEntryType.INTEGER,
359 value=8,
360 )
361 provider.config.values[CONF_LOUDNESS_NORMALIZATION] = ConfigEntry(
362 key=CONF_LOUDNESS_NORMALIZATION,
363 type=ConfigEntryType.BOOLEAN,
364 value=False,
365 )
366 provider.config.values[CONF_AUDIO_QUALITY] = ConfigEntry(
367 key=CONF_AUDIO_QUALITY,
368 type=ConfigEntryType.STRING,
369 value=AUDIO_QUALITY_HIGH,
370 )
371
372 backend = provider._create_backend()
373
374 assert isinstance(backend, SoloistBackend)
375 assert backend._api_key == "soloist-api-key-0123456789abcdef"
376 assert backend._consent is True
377 assert backend._volume_mode == VOLUME_MODE_SYNC_SPOTIFY
378 assert backend._crossfade_ms == 8000
379 assert backend._loudness_normalization is False
380 assert backend._audio_quality == AUDIO_QUALITY_HIGH
381
382
383def test_audio_behavior_defaults_reach_the_backend(tmp_path: Path) -> None:
384 """Without stored values, crossfade is off and normalization enabled."""
385 provider = _provider_with_stored_config({}, tmp_path)
386
387 backend = provider._create_backend()
388
389 assert isinstance(backend, GoLibrespotBackend)
390 assert backend._crossfade_ms == 0
391 assert backend._loudness_normalization is True
392 assert backend._audio_quality == AUDIO_QUALITY_LOSSLESS
393
394
395def test_audio_behavior_values_reach_the_backend(tmp_path: Path) -> None:
396 """The configured crossfade seconds (as ms) and normalization reach the backend."""
397 provider = _provider_with_stored_config({}, tmp_path)
398 provider.config.values[CONF_CROSSFADE_DURATION] = ConfigEntry(
399 key=CONF_CROSSFADE_DURATION,
400 type=ConfigEntryType.INTEGER,
401 value=8,
402 )
403 provider.config.values[CONF_LOUDNESS_NORMALIZATION] = ConfigEntry(
404 key=CONF_LOUDNESS_NORMALIZATION,
405 type=ConfigEntryType.BOOLEAN,
406 value=False,
407 )
408 provider.config.values[CONF_AUDIO_QUALITY] = ConfigEntry(
409 key=CONF_AUDIO_QUALITY,
410 type=ConfigEntryType.STRING,
411 value=AUDIO_QUALITY_HIGH,
412 )
413
414 backend = provider._create_backend()
415
416 assert isinstance(backend, GoLibrespotBackend)
417 assert backend._crossfade_ms == 8000
418 assert backend._loudness_normalization is False
419 assert backend._audio_quality == AUDIO_QUALITY_HIGH
420
421
422def test_write_config_carries_the_audio_behavior_keys(tmp_path: Path) -> None:
423 """The generated config.yml carries crossfade_duration (ms) and normalisation_disabled."""
424 backend = object.__new__(GoLibrespotBackend)
425 backend.mass = MagicMock()
426 backend.logger = MagicMock()
427 backend._publish_name = "Test Speaker"
428 backend._instance_id = "spotify_connect--test"
429 backend._api_port = 38800
430 backend.cache_dir = str(tmp_path)
431 backend._crossfade_ms = 8000
432 backend._loudness_normalization = False
433 backend._audio_quality = AUDIO_QUALITY_HIGH
434
435 backend._write_config(None)
436
437 config = json.loads((tmp_path / "config.yml").read_text(encoding="utf-8"))
438 assert config["crossfade_duration"] == 8000
439 assert config["normalisation_disabled"] is True
440 assert config["bitrate"] == 160
441
442
443def test_write_config_caps_lossless_at_the_engine_maximum(tmp_path: Path) -> None:
444 """go-librespot cannot do lossless, so that tier lands on its 320 kbps ceiling."""
445 backend = object.__new__(GoLibrespotBackend)
446 backend.mass = MagicMock()
447 backend.logger = MagicMock()
448 backend._publish_name = "Test Speaker"
449 backend._instance_id = "spotify_connect--test"
450 backend._api_port = 38800
451 backend.cache_dir = str(tmp_path)
452 backend._crossfade_ms = 0
453 backend._loudness_normalization = True
454 backend._audio_quality = AUDIO_QUALITY_LOSSLESS
455
456 backend._write_config(None)
457
458 config = json.loads((tmp_path / "config.yml").read_text(encoding="utf-8"))
459 assert config["bitrate"] == 320
460