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