/
/
1"""Tests for the Spotify Connect provider."""
2
3from collections.abc import AsyncGenerator
4from pathlib import Path
5from unittest.mock import AsyncMock, MagicMock, patch
6
7from music_assistant.providers.spotify_connect import SpotifyConnectProvider
8from music_assistant.providers.spotify_connect.backends.go_librespot import (
9 API_PORT_RANGE_END,
10 API_PORT_RANGE_START,
11 GoLibrespotBackend,
12)
13from music_assistant.providers.spotify_connect.clients.go_librespot import GoLibrespotClient
14
15
16async def test_backend_start_probes_api_port_on_ipv4_loopback() -> None:
17 """The daemon API port is selected on the address go-librespot binds."""
18 backend = object.__new__(GoLibrespotBackend)
19 backend.mass = MagicMock()
20 backend.logger = MagicMock()
21 backend.mass.create_task.side_effect = lambda coroutine: coroutine.close()
22
23 with (
24 patch(
25 "music_assistant.providers.spotify_connect.backends.go_librespot"
26 ".get_go_librespot_binary",
27 return_value="/usr/bin/go-librespot",
28 ),
29 patch(
30 "music_assistant.providers.spotify_connect.backends.go_librespot.select_free_port",
31 new=AsyncMock(return_value=38801),
32 ) as select_port,
33 ):
34 await backend.start()
35
36 select_port.assert_awaited_once_with(API_PORT_RANGE_START, API_PORT_RANGE_END, host="127.0.0.1")
37 assert backend._client is not None
38 assert backend._client.base_url == "http://127.0.0.1:38801"
39
40
41async def test_daemon_runner_reselects_api_port_when_taken(tmp_path: Path) -> None:
42 """An API port taken while the daemon was down is replaced before (re)starting."""
43 backend = object.__new__(GoLibrespotBackend)
44 backend.mass = MagicMock()
45 backend.mass.streams.get_source_ip = AsyncMock(return_value="192.168.1.5")
46 backend.logger = MagicMock()
47 backend.name = "Spotify Test"
48 backend.cache_dir = str(tmp_path)
49 backend._binary = "/usr/bin/go-librespot"
50 backend._api_port = 38800
51 backend._client = GoLibrespotClient(backend.mass, "http://127.0.0.1:38800", backend.logger)
52 backend._event_callback = AsyncMock()
53 # exit the supervisor loop after a single iteration
54 backend._stop_called = True
55 backend._restart_error_count = 0
56
57 async def _no_stderr() -> AsyncGenerator[str]:
58 return
59 yield
60
61 proc = MagicMock()
62 proc.start = AsyncMock()
63 proc.close = AsyncMock()
64 proc.iter_stderr = _no_stderr
65
66 with (
67 patch(
68 "music_assistant.providers.spotify_connect.backends.go_librespot.is_port_in_use",
69 new=AsyncMock(return_value=True),
70 ) as port_probe,
71 patch(
72 "music_assistant.providers.spotify_connect.backends.go_librespot.select_free_port",
73 new=AsyncMock(return_value=38801),
74 ),
75 patch(
76 "music_assistant.providers.spotify_connect.backends.go_librespot.AsyncProcess",
77 return_value=proc,
78 ),
79 patch.object(GoLibrespotBackend, "_write_config") as write_config,
80 ):
81 await backend._daemon_runner()
82
83 port_probe.assert_awaited_once_with(38800, host="127.0.0.1")
84 assert backend._api_port == 38801
85 assert backend._client.base_url == "http://127.0.0.1:38801"
86 # the daemon config pins the advertisement to the player-facing interface
87 write_config.assert_called_once_with("192.168.1.5")
88
89
90def _volume_sync_provider(volume_level: int | None) -> tuple[SpotifyConnectProvider, AsyncMock]:
91 """Build a minimal provider whose linked player reports the given volume."""
92 provider = object.__new__(SpotifyConnectProvider)
93 provider.mass = MagicMock()
94 provider.logger = MagicMock()
95 provider._last_volume_sent = None
96 backend = MagicMock()
97 set_volume = AsyncMock()
98 backend.set_volume = set_volume
99 provider._backend = backend
100 player = MagicMock()
101 player.state.volume_level = volume_level
102 provider.mass.players.get_player.return_value = player
103 return provider, set_volume
104
105
106async def test_sync_player_volume_pushes_player_volume_to_backend() -> None:
107 """The player's volume is pushed to the backend and cached for echo dedupe."""
108 provider, set_volume = _volume_sync_provider(50)
109
110 await provider._sync_player_volume_to_spotify("player1")
111
112 set_volume.assert_awaited_once_with(50)
113 assert provider._last_volume_sent == 50
114
115
116async def test_sync_player_volume_pushes_when_cache_matches() -> None:
117 """The push is unconditional: the backend's volume resets between sessions."""
118 provider, set_volume = _volume_sync_provider(50)
119 provider._last_volume_sent = 50
120
121 await provider._sync_player_volume_to_spotify("player1")
122
123 set_volume.assert_awaited_once_with(50)
124
125
126async def test_sync_player_volume_skips_when_volume_unknown() -> None:
127 """No push happens when the player does not expose a volume level."""
128 provider, set_volume = _volume_sync_provider(None)
129
130 await provider._sync_player_volume_to_spotify("player1")
131
132 set_volume.assert_not_awaited()
133 assert provider._last_volume_sent is None
134
135
136async def test_sync_player_volume_restores_cache_on_failure() -> None:
137 """A failed push restores the dedupe cache so a retry is not wrongly deduped."""
138 provider, set_volume = _volume_sync_provider(50)
139 set_volume.side_effect = OSError("daemon gone")
140
141 await provider._sync_player_volume_to_spotify("player1")
142
143 assert provider._last_volume_sent is None
144