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