music-assistant-server

1"""
2Tests for the stream route a live source playing on a player is served from.
3
4The url carries the session it was built for, so a renderer that reconnects after the
5player moved on is turned away rather than being handed whatever is playing now. The
6same holds for the direct-PCM consumers, which never reach the route at all.
7"""
8
9from __future__ import annotations
10
11from collections.abc import AsyncGenerator
12from types import SimpleNamespace
13from typing import Any
14from unittest.mock import AsyncMock, MagicMock
15
16import pytest
17from aiohttp import web
18from music_assistant_models.enums import MediaType
19from music_assistant_models.errors import AudioError
20from music_assistant_models.media_items import AudioFormat, AudioSource
21
22from music_assistant.controllers.players.audio_sources import AudioSourceSession
23from music_assistant.controllers.streams import StreamsController
24from music_assistant.models.player import PlayerMedia
25from music_assistant.models.plugin import PluginProvider
26
27OWNER_ID = "player_1"
28CONSUMER_ID = "spb_bridge_1"
29INSTANCE_ID = "spotify_connect--abc"
30
31
32def _session(player_id: str = OWNER_ID) -> AudioSourceSession:
33    return AudioSourceSession(
34        player_id=player_id,
35        source=AudioSource(
36            item_id="main", provider=INSTANCE_ID, name="Spotify Connect", provider_mappings=set()
37        ),
38        provider_instance_id=INSTANCE_ID,
39    )
40
41
42def _controller(session: AudioSourceSession | None) -> tuple[Any, MagicMock, MagicMock]:
43    """Build a bare streams controller whose player controller holds ``session``."""
44    ctrl = StreamsController.__new__(StreamsController)
45    ctrl.mass = MagicMock()
46    ctrl.logger = MagicMock()
47    # a truthy MagicMock here would send _log_request down the verbose path
48    ctrl.logger.isEnabledFor = MagicMock(return_value=False)
49    provider = MagicMock(spec=PluginProvider)
50    provider.instance_id = INSTANCE_ID
51    provider.on_source_selected = AsyncMock()
52    provider.on_source_unselected = AsyncMock()
53    ctrl.mass.get_provider = MagicMock(return_value=provider)
54    ctrl.mass.players.get_audio_source_session = MagicMock(
55        return_value=session, side_effect=lambda pid: session if pid == OWNER_ID else None
56    )
57    player = MagicMock()
58    player.player_id = CONSUMER_ID
59    ctrl.mass.players.get_player = MagicMock(return_value=player)
60    ctrl.mass.players.deselect_source = AsyncMock()
61    return ctrl, provider, player
62
63
64def _request(*, session_id: str, source_player_id: str = OWNER_ID) -> Any:
65    return SimpleNamespace(
66        method="GET",
67        match_info={
68            "session_id": session_id,
69            "source_player_id": source_player_id,
70            "player_id": CONSUMER_ID,
71        },
72        # the request logger reads these off every request it is handed
73        path=f"/source/{session_id}/{source_player_id}/{CONSUMER_ID}.flac",
74        remote="10.0.0.5",
75        version=SimpleNamespace(major=1, minor=1),
76        headers={},
77    )
78
79
80def test_the_url_resolves_to_the_session_it_names() -> None:
81    """A url carrying the live session's own token resolves to it."""
82    session = _session()
83    ctrl, provider, player = _controller(session)
84
85    resolved, resolved_player, resolved_prov = ctrl._resolve_audio_source_request(
86        _request(session_id=session.playback_session_id)
87    )
88
89    assert resolved is session
90    assert resolved_player is player
91    assert resolved_prov is provider
92
93
94def test_a_url_from_a_superseded_session_is_turned_away() -> None:
95    """
96    A renderer reconnecting with a stale token gets a 404, not the current source.
97
98    The token is what separates the session the url was built for from whatever the
99    player happens to be playing now.
100    """
101    ctrl, _provider, _player = _controller(_session())
102
103    with pytest.raises(web.HTTPNotFound):
104        ctrl._resolve_audio_source_request(_request(session_id="a-token-from-before"))
105
106
107def test_a_url_for_a_player_playing_nothing_is_turned_away() -> None:
108    """Without a session there is nothing to serve."""
109    ctrl, _provider, _player = _controller(None)
110
111    with pytest.raises(web.HTTPNotFound):
112        ctrl._resolve_audio_source_request(_request(session_id="anything"))
113
114
115def test_an_unknown_consuming_player_is_turned_away() -> None:
116    """The url also names who is consuming, which has to exist."""
117    session = _session()
118    ctrl, _provider, _player = _controller(session)
119    ctrl.mass.players.get_player = MagicMock(return_value=None)
120
121    with pytest.raises(web.HTTPNotFound):
122        ctrl._resolve_audio_source_request(_request(session_id=session.playback_session_id))
123
124
125async def test_a_head_probe_does_not_trigger_the_plugin() -> None:
126    """
127    A renderer probing with HEAD must not fire the selection side effects.
128
129    on_source_selected stops the previous player and can redirect a disallowed
130    switch — none of which a probe should cause.
131    """
132    session = _session()
133    ctrl, provider, _player = _controller(session)
134    ctrl._serve_audio_source_head = AsyncMock(return_value="head-response")
135    request = _request(session_id=session.playback_session_id)
136    request.method = "HEAD"
137
138    result = await ctrl.serve_audio_source_stream(request)
139
140    assert result == "head-response"
141    provider.on_source_selected.assert_not_awaited()
142
143
144def test_a_direct_pcm_request_from_a_superseded_session_is_refused() -> None:
145    """
146    The PCM consumers are held to the same token as the url renderers.
147
148    They resolve the session from the player rather than a url, so without this a
149    stale request would silently attach to whichever source is playing now.
150    """
151    session = _session()
152    ctrl, _provider, _player = _controller(session)
153
154    with pytest.raises(AudioError, match="Unknown"):
155        ctrl.get_stream(
156            PlayerMedia(
157                uri="x://audio_source/main",
158                media_type=MediaType.AUDIO_SOURCE,
159                source_id=OWNER_ID,
160                queue_session_id="a-token-from-before",
161            ),
162            AudioFormat(),
163            player_id=CONSUMER_ID,
164        )
165
166
167def test_a_direct_pcm_request_carrying_the_live_token_is_served() -> None:
168    """A consumer naming the session that is playing gets its stream."""
169    session = _session()
170    ctrl, _provider, _player = _controller(session)
171    ctrl._get_audio_source_session_stream = MagicMock(return_value="pcm-stream")
172
173    result = ctrl.get_stream(
174        PlayerMedia(
175            uri="x://audio_source/main",
176            media_type=MediaType.AUDIO_SOURCE,
177            source_id=OWNER_ID,
178            queue_session_id=session.playback_session_id,
179        ),
180        AudioFormat(),
181        player_id=CONSUMER_ID,
182    )
183
184    assert result == "pcm-stream"
185    ctrl._get_audio_source_session_stream.assert_called_once_with(
186        session, AudioFormat(), CONSUMER_ID
187    )
188
189
190async def test_a_plugin_refusing_the_stream_takes_the_source_off_the_player() -> None:
191    """
192    A source that never starts is released, not left published.
193
194    The play command that pointed the renderer here has already returned, so nothing
195    else clears the session — and a plugin refusing the stream is a designed path
196    (Ynison raises from the hook when it redirects to its configured target).
197    """
198    session = _session()
199    ctrl, provider, _player = _controller(session)
200    provider.on_source_selected = AsyncMock(side_effect=RuntimeError("switching disabled"))
201
202    with pytest.raises(web.HTTPNotFound):
203        await ctrl.serve_audio_source_stream(_request(session_id=session.playback_session_id))
204
205    ctrl.mass.players.deselect_source.assert_awaited_once_with(
206        OWNER_ID,
207        provider_instance_id=session.provider_instance_id,
208        source_id=session.source_id,
209        playback_session_id=session.playback_session_id,
210    )
211
212
213async def test_failing_stream_details_also_takes_the_source_off_the_player() -> None:
214    """The same holds when the plugin claims the source but cannot describe its stream."""
215    session = _session()
216    ctrl, provider, _player = _controller(session)
217    provider.get_stream_details = AsyncMock(side_effect=OSError("daemon gone"))
218
219    with pytest.raises(web.HTTPNotFound):
220        await ctrl.serve_audio_source_stream(_request(session_id=session.playback_session_id))
221
222    ctrl.mass.players.deselect_source.assert_awaited_once_with(
223        OWNER_ID,
224        provider_instance_id=session.provider_instance_id,
225        source_id=session.source_id,
226        playback_session_id=session.playback_session_id,
227    )
228
229
230async def test_a_session_already_superseded_is_not_released() -> None:
231    """A newer session on the player is not this request's to take away."""
232    session = _session()
233    ctrl, provider, _player = _controller(session)
234    provider.on_source_selected = AsyncMock(side_effect=RuntimeError("nope"))
235    # the player moved on to a different session while this request was setting up
236    ctrl.mass.players.get_audio_source_session = MagicMock(return_value=_session())
237
238    with pytest.raises(web.HTTPNotFound):
239        await ctrl.serve_audio_source_stream(_request(session_id=session.playback_session_id))
240
241    ctrl.mass.players.deselect_source.assert_not_awaited()
242
243
244async def test_a_reselected_session_is_not_released_after_setup_failure() -> None:
245    """A failed request cannot release a newer selection using the same session object."""
246    session = _session()
247    ctrl, provider, _player = _controller(session)
248
249    async def supersede_session(*_args: Any) -> None:
250        session.playback_session_id = "replacement-session"
251        raise RuntimeError("nope")
252
253    provider.on_source_selected = AsyncMock(side_effect=supersede_session)
254
255    with pytest.raises(web.HTTPNotFound):
256        await ctrl.serve_audio_source_stream(_request(session_id=session.playback_session_id))
257
258    ctrl.mass.players.deselect_source.assert_not_awaited()
259
260
261async def test_http_setup_stops_when_the_session_is_reselected() -> None:
262    """HTTP setup cannot stamp its stream token onto a newer source selection."""
263    session = _session()
264    ctrl, provider, _player = _controller(session)
265    request_session_id = session.playback_session_id
266
267    async def supersede_session(*_args: Any) -> None:
268        session.playback_session_id = "replacement-session"
269
270    provider.on_source_selected = AsyncMock(side_effect=supersede_session)
271    ctrl._prepare_audio_source_stream = AsyncMock()
272
273    with pytest.raises(web.HTTPNotFound, match="superseded"):
274        await ctrl.serve_audio_source_stream(_request(session_id=request_session_id))
275
276    ctrl._prepare_audio_source_stream.assert_not_awaited()
277    ctrl.mass.players.deselect_source.assert_not_awaited()
278
279
280async def test_direct_pcm_setup_stops_when_the_session_is_reselected() -> None:
281    """PCM setup cannot stamp its stream token onto a newer source selection."""
282    session = _session()
283    ctrl, provider, _player = _controller(session)
284
285    async def supersede_session(*_args: Any) -> None:
286        session.playback_session_id = "replacement-session"
287
288    provider.on_source_selected = AsyncMock(side_effect=supersede_session)
289    stream = ctrl._get_audio_source_session_stream(session, AudioFormat(), CONSUMER_ID)
290
291    with pytest.raises(AudioError, match="superseded"):
292        await anext(stream)
293
294    ctrl.mass.players.deselect_source.assert_not_awaited()
295
296
297async def test_direct_pcm_stream_stops_when_the_session_is_reselected() -> None:
298    """A running PCM stream ends before yielding audio for a newer selection."""
299    session = _session()
300    session.streamdetails = MagicMock()
301    ctrl, _provider, _player = _controller(session)
302    ctrl.audio = MagicMock()
303
304    async def chunks() -> AsyncGenerator[bytes]:
305        yield b"first"
306        yield b"stale"
307
308    ctrl.audio.get_audio_source_stream = MagicMock(return_value=chunks())
309    stream = ctrl._get_audio_source_session_stream(session, AudioFormat(), CONSUMER_ID)
310
311    assert await anext(stream) == b"first"
312    session.playback_session_id = "replacement-session"
313
314    with pytest.raises(StopAsyncIteration):
315        await anext(stream)
316