/
/
/
1"""Tests for the MSX Sendspin bridge (spec 0003)."""
2
3from __future__ import annotations
4
5import json
6from typing import Any
7from unittest.mock import AsyncMock, Mock, patch
8
9import pytest
10
11# The bridge rides on MA's Sendspin provider; skip the whole module when the
12# installed Music Assistant ships no Sendspin provider (as some CI images do).
13pytest.importorskip("music_assistant.providers.sendspin")
14
15from music_assistant_models.enums import ConfigEntryType
16
17from music_assistant.providers.msx_bridge.constants import (
18 CONF_ENABLE_SENDSPIN_BRIDGE,
19)
20from music_assistant.providers.msx_bridge.http_server import MSXHTTPServer
21from music_assistant.providers.msx_bridge.player import MSXPlayer
22from music_assistant.providers.msx_bridge.provider import MSXBridgeProvider
23from music_assistant.providers.msx_bridge.sendspin_bridge import (
24 MSXSendspinBridge,
25 MSXSendspinBridgeManager,
26 bridge_client_id_for,
27)
28
29
30def _wire_sendspin(mass_mock: Mock) -> tuple[Mock, Mock]:
31 """Wire a mock Sendspin provider + server into mass_mock; return (provider, server)."""
32 client = Mock()
33 client.is_connected = False
34 client.roles_by_family = Mock(return_value=[])
35 server = Mock()
36 server.register_external_player = Mock(return_value=client)
37 server.remove_client = AsyncMock()
38 sendspin_prov = Mock()
39 sendspin_prov.server_api = server
40 mass_mock.get_provider = Mock(
41 side_effect=lambda domain: sendspin_prov if domain == "sendspin" else None
42 )
43 return sendspin_prov, server
44
45
46# --- Client id derivation ---
47
48
49def test_bridge_client_id_is_stable() -> None:
50 """Client ids are stable and derived from the MSX player id."""
51 assert bridge_client_id_for("msx_livingroom_tv") == "spb_msx_livingroom_tv"
52 assert bridge_client_id_for("msx_livingroom_tv") == bridge_client_id_for("msx_livingroom_tv")
53
54
55# --- Registration lifecycle (AC1, AC5, AC7) ---
56
57
58async def test_bridge_registered_when_enabled(
59 provider: MSXBridgeProvider, mass_mock: Mock, player: MSXPlayer
60) -> None:
61 """AC1: with the option enabled, an MSX player gets an external Sendspin client."""
62 sendspin_prov, server = _wire_sendspin(mass_mock)
63 mass_mock.players.get_player = Mock(return_value=player)
64 provider.sendspin_bridge_enabled = True
65 manager = MSXSendspinBridgeManager(provider)
66
67 await manager.evaluate_bridge(player)
68
69 server.register_external_player.assert_called_once()
70 hello = server.register_external_player.call_args[0][0]
71 assert hello.client_id == "spb_msx_test"
72 sendspin_prov.register_bridge_underlying_player.assert_called_once_with(
73 "spb_msx_test", "msx_test"
74 )
75 assert manager.get_bridge("msx_test") is not None
76
77
78async def test_no_bridge_when_disabled(
79 provider: MSXBridgeProvider, mass_mock: Mock, player: MSXPlayer
80) -> None:
81 """AC5: with the option disabled (default), no Sendspin client is registered."""
82 _, server = _wire_sendspin(mass_mock)
83 mass_mock.players.get_player = Mock(return_value=player)
84 assert provider.sendspin_bridge_enabled is False
85 manager = MSXSendspinBridgeManager(provider)
86
87 await manager.evaluate_bridge(player)
88
89 server.register_external_player.assert_not_called()
90 assert manager.get_bridge("msx_test") is None
91
92
93async def test_two_tvs_get_distinct_bridges(provider: MSXBridgeProvider, mass_mock: Mock) -> None:
94 """Two registered TVs yield two bridge registrations with stable, distinct ids."""
95 _, server = _wire_sendspin(mass_mock)
96 players = {
97 pid: MSXPlayer(provider, pid, name=pid, output_format="mp3")
98 for pid in ("msx_tv1", "msx_tv2")
99 }
100 for p in players.values():
101 p.update_state = Mock() # type: ignore[misc, method-assign]
102 mass_mock.players.get_player = Mock(side_effect=lambda pid, **_kw: players.get(pid))
103 provider.sendspin_bridge_enabled = True
104 manager = MSXSendspinBridgeManager(provider)
105
106 for p in players.values():
107 await manager.evaluate_bridge(p)
108
109 assert server.register_external_player.call_count == 2
110 client_ids = {call.args[0].client_id for call in server.register_external_player.call_args_list}
111 assert client_ids == {"spb_msx_tv1", "spb_msx_tv2"}
112
113
114async def test_unregister_removes_bridge_client(
115 provider: MSXBridgeProvider, mass_mock: Mock, player: MSXPlayer
116) -> None:
117 """AC7: unregistering an MSX player removes its bridge client from the server."""
118 _, server = _wire_sendspin(mass_mock)
119 mass_mock.players.get_player = Mock(return_value=player)
120 provider.sendspin_bridge_enabled = True
121 provider.bridge_manager = MSXSendspinBridgeManager(provider)
122 await provider.bridge_manager.evaluate_bridge(player)
123 assert provider.bridge_manager.get_bridge("msx_test") is not None
124
125 await provider._handle_player_unregister("msx_test")
126
127 server.remove_client.assert_awaited_once_with("spb_msx_test")
128 assert provider.bridge_manager.get_bridge("msx_test") is None
129
130
131async def test_unload_closes_bridge_manager(
132 provider: MSXBridgeProvider, mass_mock: Mock, player: MSXPlayer
133) -> None:
134 """Provider unload stops all bridges."""
135 _, server = _wire_sendspin(mass_mock)
136 mass_mock.players.get_player = Mock(return_value=player)
137 provider.sendspin_bridge_enabled = True
138 provider.bridge_manager = MSXSendspinBridgeManager(provider)
139 await provider.bridge_manager.evaluate_bridge(player)
140
141 await provider.unload()
142
143 server.remove_client.assert_awaited_once_with("spb_msx_test")
144
145
146# --- Stream start -> WS push (AC3) ---
147
148
149async def test_stream_start_pushes_kiosk_via_ws(
150 provider: MSXBridgeProvider, mass_mock: Mock, player: MSXPlayer
151) -> None:
152 """AC3: a Sendspin stream start pushes the TV into Sendspin kiosk mode over WS."""
153 _, server = _wire_sendspin(mass_mock)
154 provider.http_server = Mock()
155 bridge = MSXSendspinBridge(provider, player, server, "spb_msx_test")
156 await bridge.start()
157
158 bridge._on_stream_start(Mock(client_id="spb_msx_test"))
159
160 provider.http_server.broadcast_sendspin.assert_called_once()
161 player_id, url = provider.http_server.broadcast_sendspin.call_args[0]
162 assert player_id == "msx_test"
163 assert "kiosk=1" in url
164 assert "sendspin=1" in url
165 assert "sendspin_client_id=spb_msx_test" in url
166
167
168# --- Connect watchdog (AC6) ---
169
170
171async def test_connect_watchdog_falls_back_to_http_player(
172 provider: MSXBridgeProvider, mass_mock: Mock, player: MSXPlayer
173) -> None:
174 """AC6: when the TV never connects its Sendspin client, playback transfers to HTTP."""
175 _, server = _wire_sendspin(mass_mock)
176 server.register_external_player.return_value.is_connected = False
177 mass_mock.player_queues.transfer_queue = AsyncMock()
178 bridge = MSXSendspinBridge(provider, player, server, "spb_msx_test")
179 await bridge.start()
180
181 with patch(
182 "music_assistant.providers.msx_bridge.sendspin_bridge.SENDSPIN_CONNECT_TIMEOUT", 0.01
183 ):
184 await bridge._watch_connect()
185
186 mass_mock.player_queues.transfer_queue.assert_awaited_once_with(
187 "spb_msx_test", "msx_test", auto_play=True
188 )
189
190
191async def test_connect_watchdog_noop_when_connected(
192 provider: MSXBridgeProvider, mass_mock: Mock, player: MSXPlayer
193) -> None:
194 """No fallback when the TV's Sendspin client connected in time."""
195 _, server = _wire_sendspin(mass_mock)
196 server.register_external_player.return_value.is_connected = True
197 mass_mock.player_queues.transfer_queue = AsyncMock()
198 bridge = MSXSendspinBridge(provider, player, server, "spb_msx_test")
199 await bridge.start()
200
201 with patch(
202 "music_assistant.providers.msx_bridge.sendspin_bridge.SENDSPIN_CONNECT_TIMEOUT", 0.01
203 ):
204 await bridge._watch_connect()
205
206 mass_mock.player_queues.transfer_queue.assert_not_awaited()
207
208
209# --- WS broadcast plumbing ---
210
211
212async def test_broadcast_sendspin_sends_ws_message(
213 provider: MSXBridgeProvider, mass_mock: Mock
214) -> None:
215 """broadcast_sendspin delivers a typed message to the player's WS clients."""
216 server = MSXHTTPServer(provider, 0)
217 ws = AsyncMock()
218 ws.closed = False
219 server._ws_clients["msx_test"] = {ws}
220 coros: list[Any] = []
221
222 def _capture_task(coro: Any) -> Mock:
223 coros.append(coro)
224 return Mock()
225
226 mass_mock.create_task = Mock(side_effect=_capture_task)
227
228 server.broadcast_sendspin("msx_test", "/web?kiosk=1&sendspin=1")
229
230 assert len(coros) == 1
231 await coros[0]
232 ws.send_str.assert_awaited_once()
233 payload = json.loads(ws.send_str.call_args[0][0])
234 assert payload == {
235 "type": "sendspin",
236 "url": "/web?kiosk=1&sendspin=1",
237 "player_id": "msx_test",
238 }
239
240
241# --- Provider wiring ---
242
243
244async def test_config_entry_default_on(provider: MSXBridgeProvider) -> None:
245 """
246 The enable_sendspin_bridge config entry exists and defaults to True.
247
248 Supersedes the original spec's AC5 (default off while experimental):
249 the default was deliberately flipped on once the bridge stabilized.
250 """
251 entries = await provider.get_config_entries()
252 entry = next(e for e in entries if e.key == CONF_ENABLE_SENDSPIN_BRIDGE)
253 assert entry.type == ConfigEntryType.BOOLEAN
254 assert entry.default_value is True
255
256
257async def test_register_player_evaluates_bridge(
258 provider: MSXBridgeProvider, mass_mock: Mock
259) -> None:
260 """Registering a new MSX player triggers bridge evaluation."""
261 mass_mock.players.get_player = Mock(return_value=None)
262 provider.bridge_manager = Mock()
263 provider.bridge_manager.evaluate_bridge = AsyncMock()
264
265 registered = await provider.get_or_register_player("msx_new")
266
267 assert registered is not None
268 provider.bridge_manager.evaluate_bridge.assert_awaited_once_with(registered)
269