/
/
1"""Tests for the Spotify Connect provider."""
2
3from collections.abc import AsyncGenerator
4from pathlib import Path
5from typing import Any
6from unittest.mock import AsyncMock, MagicMock, patch
7
8from music_assistant_models.config_entries import ConfigEntry, ProviderConfig
9from music_assistant_models.enums import ConfigEntryType, ProviderType
10
11from music_assistant.providers.spotify_connect import (
12 BACKEND_SOLOIST,
13 CONF_API_KEY,
14 CONF_BACKEND,
15 CONF_SOLOIST_CONSENT,
16 CONF_VOLUME_MODE,
17 SpotifyConnectProvider,
18)
19from music_assistant.providers.spotify_connect.go_librespot.backend import (
20 API_PORT_RANGE_END,
21 API_PORT_RANGE_START,
22 GoLibrespotBackend,
23)
24from music_assistant.providers.spotify_connect.go_librespot.client import GoLibrespotClient
25from music_assistant.providers.spotify_connect.soloist.backend import (
26 VOLUME_MODE_SYNC_SPOTIFY,
27 SoloistBackend,
28)
29
30
31async def test_backend_start_probes_api_port_on_ipv4_loopback() -> None:
32 """The daemon API port is selected on the address go-librespot binds."""
33 backend = object.__new__(GoLibrespotBackend)
34 backend.mass = MagicMock()
35 backend.logger = MagicMock()
36 backend.mass.create_task.side_effect = lambda coroutine: coroutine.close()
37
38 with (
39 patch(
40 "music_assistant.providers.spotify_connect.go_librespot.backend"
41 ".get_go_librespot_binary",
42 return_value="/usr/bin/go-librespot",
43 ),
44 patch(
45 "music_assistant.providers.spotify_connect.go_librespot.backend.select_free_port",
46 new=AsyncMock(return_value=38801),
47 ) as select_port,
48 ):
49 await backend.start()
50
51 select_port.assert_awaited_once_with(API_PORT_RANGE_START, API_PORT_RANGE_END, host="127.0.0.1")
52 assert backend._client is not None
53 assert backend._client.base_url == "http://127.0.0.1:38801"
54
55
56async def test_daemon_runner_reselects_api_port_when_taken(tmp_path: Path) -> None:
57 """An API port taken while the daemon was down is replaced before (re)starting."""
58 backend = object.__new__(GoLibrespotBackend)
59 backend.mass = MagicMock()
60 backend.mass.streams.get_source_ip = AsyncMock(return_value="192.168.1.5")
61 backend.logger = MagicMock()
62 backend.name = "Spotify Test"
63 backend.cache_dir = str(tmp_path)
64 backend._binary = "/usr/bin/go-librespot"
65 backend._api_port = 38800
66 backend._client = GoLibrespotClient(backend.mass, "http://127.0.0.1:38800", backend.logger)
67 backend._event_callback = AsyncMock()
68 # exit the supervisor loop after a single iteration
69 backend._stop_called = True
70 backend._restart_error_count = 0
71
72 async def _no_stderr() -> AsyncGenerator[str]:
73 return
74 yield
75
76 proc = MagicMock()
77 proc.start = AsyncMock()
78 proc.close = AsyncMock()
79 proc.iter_stderr = _no_stderr
80
81 with (
82 patch(
83 "music_assistant.providers.spotify_connect.go_librespot.backend.is_port_in_use",
84 new=AsyncMock(return_value=True),
85 ) as port_probe,
86 patch(
87 "music_assistant.providers.spotify_connect.go_librespot.backend.select_free_port",
88 new=AsyncMock(return_value=38801),
89 ),
90 patch(
91 "music_assistant.providers.spotify_connect.go_librespot.backend.AsyncProcess",
92 return_value=proc,
93 ),
94 patch.object(GoLibrespotBackend, "_write_config") as write_config,
95 ):
96 await backend._daemon_runner()
97
98 port_probe.assert_awaited_once_with(38800, host="127.0.0.1")
99 assert backend._api_port == 38801
100 assert backend._client.base_url == "http://127.0.0.1:38801"
101 # the daemon config pins the advertisement to the player-facing interface
102 write_config.assert_called_once_with("192.168.1.5")
103
104
105def _volume_sync_provider(volume_level: int | None) -> tuple[SpotifyConnectProvider, AsyncMock]:
106 """Build a minimal provider whose linked player reports the given volume."""
107 provider = object.__new__(SpotifyConnectProvider)
108 provider.mass = MagicMock()
109 provider.logger = MagicMock()
110 provider._last_volume_sent = None
111 backend = MagicMock()
112 set_volume = AsyncMock()
113 backend.set_volume = set_volume
114 provider._backend = backend
115 player = MagicMock()
116 player.state.volume_level = volume_level
117 provider.mass.players.get_player.return_value = player
118 return provider, set_volume
119
120
121async def test_sync_player_volume_pushes_player_volume_to_backend() -> None:
122 """The player's volume is pushed to the backend and cached for echo dedupe."""
123 provider, set_volume = _volume_sync_provider(50)
124
125 await provider._sync_player_volume_to_spotify("player1")
126
127 set_volume.assert_awaited_once_with(50)
128 assert provider._last_volume_sent == 50
129
130
131async def test_sync_player_volume_pushes_when_cache_matches() -> None:
132 """The push is unconditional: the backend's volume resets between sessions."""
133 provider, set_volume = _volume_sync_provider(50)
134 provider._last_volume_sent = 50
135
136 await provider._sync_player_volume_to_spotify("player1")
137
138 set_volume.assert_awaited_once_with(50)
139
140
141async def test_sync_player_volume_skips_when_volume_unknown() -> None:
142 """No push happens when the player does not expose a volume level."""
143 provider, set_volume = _volume_sync_provider(None)
144
145 await provider._sync_player_volume_to_spotify("player1")
146
147 set_volume.assert_not_awaited()
148 assert provider._last_volume_sent is None
149
150
151async def test_sync_player_volume_restores_cache_on_failure() -> None:
152 """A failed push restores the dedupe cache so a retry is not wrongly deduped."""
153 provider, set_volume = _volume_sync_provider(50)
154 set_volume.side_effect = OSError("daemon gone")
155
156 await provider._sync_player_volume_to_spotify("player1")
157
158 assert provider._last_volume_sent is None
159
160
161def _provider_with_stored_config(
162 setup_data: dict[str, Any], tmp_path: Path
163) -> SpotifyConnectProvider:
164 """Build a provider whose stored setup_data resolves through the real accessors."""
165 provider = object.__new__(SpotifyConnectProvider)
166 provider.mass = MagicMock()
167 provider.mass.storage_path = str(tmp_path / "storage")
168 provider.mass.cache_path = str(tmp_path / "cache")
169 provider.mass.config.get.return_value = setup_data
170 provider.mass.config.decrypt_string.side_effect = lambda value: value
171 provider.logger = MagicMock()
172 provider._publish_name = "Test Speaker"
173 provider.config = ProviderConfig(
174 values={},
175 type=ProviderType.PLUGIN,
176 domain="spotify_connect",
177 instance_id="spotify_connect--test",
178 name="Spotify Connect",
179 )
180 return provider
181
182
183def test_config_without_backend_choice_loads_go_librespot(tmp_path: Path) -> None:
184 """A config from before the backend choice existed loads go-librespot unchanged."""
185 provider = _provider_with_stored_config({}, tmp_path)
186
187 assert isinstance(provider._create_backend(), GoLibrespotBackend)
188
189
190def test_soloist_setup_data_loads_soloist_backend(tmp_path: Path) -> None:
191 """A flow-configured soloist instance loads the soloist backend with its stored values."""
192 provider = _provider_with_stored_config(
193 {
194 CONF_BACKEND: BACKEND_SOLOIST,
195 CONF_API_KEY: "soloist-api-key-0123456789abcdef",
196 CONF_SOLOIST_CONSENT: True,
197 },
198 tmp_path,
199 )
200 # the volume mode lives in the provider options, not in the setup data
201 provider.config.values[CONF_VOLUME_MODE] = ConfigEntry(
202 key=CONF_VOLUME_MODE,
203 type=ConfigEntryType.STRING,
204 value=VOLUME_MODE_SYNC_SPOTIFY,
205 )
206
207 backend = provider._create_backend()
208
209 assert isinstance(backend, SoloistBackend)
210 assert backend._api_key == "soloist-api-key-0123456789abcdef"
211 assert backend._consent is True
212 assert backend._volume_mode == VOLUME_MODE_SYNC_SPOTIFY
213