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 types import SimpleNamespace
12from typing import Any
13from unittest.mock import AsyncMock, MagicMock
14
15import pytest
16from aiohttp import web
17from music_assistant_models.enums import MediaType
18from music_assistant_models.errors import AudioError
19from music_assistant_models.media_items import AudioFormat, AudioSource
20
21from music_assistant.controllers.players.audio_sources import AudioSourceSession
22from music_assistant.controllers.streams import StreamsController
23from music_assistant.models.player import PlayerMedia
24from music_assistant.models.plugin import PluginProvider
25
26OWNER_ID = "player_1"
27CONSUMER_ID = "spb_bridge_1"
28INSTANCE_ID = "spotify_connect--abc"
29
30
31def _session(player_id: str = OWNER_ID) -> AudioSourceSession:
32    return AudioSourceSession(
33        player_id=player_id,
34        source=AudioSource(
35            item_id="main", provider=INSTANCE_ID, name="Spotify Connect", provider_mappings=set()
36        ),
37        provider_instance_id=INSTANCE_ID,
38    )
39
40
41def _controller(session: AudioSourceSession | None) -> tuple[Any, MagicMock, MagicMock]:
42    """Build a bare streams controller whose player controller holds ``session``."""
43    ctrl = StreamsController.__new__(StreamsController)
44    ctrl.mass = MagicMock()
45    ctrl.logger = MagicMock()
46    # a truthy MagicMock here would send _log_request down the verbose path
47    ctrl.logger.isEnabledFor = MagicMock(return_value=False)
48    provider = MagicMock(spec=PluginProvider)
49    provider.instance_id = INSTANCE_ID
50    provider.on_source_selected = AsyncMock()
51    provider.on_source_unselected = AsyncMock()
52    ctrl.mass.get_provider = MagicMock(return_value=provider)
53    ctrl.mass.players.get_audio_source_session = MagicMock(
54        return_value=session, side_effect=lambda pid: session if pid == OWNER_ID else None
55    )
56    player = MagicMock()
57    player.player_id = CONSUMER_ID
58    ctrl.mass.players.get_player = MagicMock(return_value=player)
59    ctrl.mass.players.deselect_source = AsyncMock()
60    return ctrl, provider, player
61
62
63def _request(*, session_id: str, source_player_id: str = OWNER_ID) -> Any:
64    return SimpleNamespace(
65        method="GET",
66        match_info={
67            "session_id": session_id,
68            "source_player_id": source_player_id,
69            "player_id": CONSUMER_ID,
70        },
71        # the request logger reads these off every request it is handed
72        path=f"/source/{session_id}/{source_player_id}/{CONSUMER_ID}.flac",
73        remote="10.0.0.5",
74        version=SimpleNamespace(major=1, minor=1),
75        headers={},
76    )
77
78
79def test_the_url_resolves_to_the_session_it_names() -> None:
80    """A url carrying the live session's own token resolves to it."""
81    session = _session()
82    ctrl, provider, player = _controller(session)
83
84    resolved, resolved_player, resolved_prov = ctrl._resolve_audio_source_request(
85        _request(session_id=session.playback_session_id)
86    )
87
88    assert resolved is session
89    assert resolved_player is player
90    assert resolved_prov is provider
91
92
93def test_a_url_from_a_superseded_session_is_turned_away() -> None:
94    """
95    A renderer reconnecting with a stale token gets a 404, not the current source.
96
97    The token is what separates the session the url was built for from whatever the
98    player happens to be playing now.
99    """
100    ctrl, _provider, _player = _controller(_session())
101
102    with pytest.raises(web.HTTPNotFound):
103        ctrl._resolve_audio_source_request(_request(session_id="a-token-from-before"))
104
105
106def test_a_url_for_a_player_playing_nothing_is_turned_away() -> None:
107    """Without a session there is nothing to serve."""
108    ctrl, _provider, _player = _controller(None)
109
110    with pytest.raises(web.HTTPNotFound):
111        ctrl._resolve_audio_source_request(_request(session_id="anything"))
112
113
114def test_an_unknown_consuming_player_is_turned_away() -> None:
115    """The url also names who is consuming, which has to exist."""
116    session = _session()
117    ctrl, _provider, _player = _controller(session)
118    ctrl.mass.players.get_player = MagicMock(return_value=None)
119
120    with pytest.raises(web.HTTPNotFound):
121        ctrl._resolve_audio_source_request(_request(session_id=session.playback_session_id))
122
123
124async def test_a_head_probe_does_not_trigger_the_plugin() -> None:
125    """
126    A renderer probing with HEAD must not fire the selection side effects.
127
128    on_source_selected stops the previous player and can redirect a disallowed
129    switch — none of which a probe should cause.
130    """
131    session = _session()
132    ctrl, provider, _player = _controller(session)
133    ctrl._serve_audio_source_head = AsyncMock(return_value="head-response")
134    request = _request(session_id=session.playback_session_id)
135    request.method = "HEAD"
136
137    result = await ctrl.serve_audio_source_stream(request)
138
139    assert result == "head-response"
140    provider.on_source_selected.assert_not_awaited()
141
142
143def test_a_direct_pcm_request_from_a_superseded_session_is_refused() -> None:
144    """
145    The PCM consumers are held to the same token as the url renderers.
146
147    They resolve the session from the player rather than a url, so without this a
148    stale request would silently attach to whichever source is playing now.
149    """
150    session = _session()
151    ctrl, _provider, _player = _controller(session)
152
153    with pytest.raises(AudioError, match="Unknown"):
154        ctrl.get_stream(
155            PlayerMedia(
156                uri="x://audio_source/main",
157                media_type=MediaType.AUDIO_SOURCE,
158                source_id=OWNER_ID,
159                queue_session_id="a-token-from-before",
160            ),
161            AudioFormat(),
162            player_id=CONSUMER_ID,
163        )
164
165
166def test_a_direct_pcm_request_carrying_the_live_token_is_served() -> None:
167    """A consumer naming the session that is playing gets its stream."""
168    session = _session()
169    ctrl, _provider, _player = _controller(session)
170    ctrl._get_audio_source_session_stream = MagicMock(return_value="pcm-stream")
171
172    result = ctrl.get_stream(
173        PlayerMedia(
174            uri="x://audio_source/main",
175            media_type=MediaType.AUDIO_SOURCE,
176            source_id=OWNER_ID,
177            queue_session_id=session.playback_session_id,
178        ),
179        AudioFormat(),
180        player_id=CONSUMER_ID,
181    )
182
183    assert result == "pcm-stream"
184    ctrl._get_audio_source_session_stream.assert_called_once_with(
185        session, AudioFormat(), CONSUMER_ID
186    )
187
188
189async def test_a_plugin_refusing_the_stream_takes_the_source_off_the_player() -> None:
190    """
191    A source that never starts is released, not left published.
192
193    The play command that pointed the renderer here has already returned, so nothing
194    else clears the session — and a plugin refusing the stream is a designed path
195    (Ynison raises from the hook when it redirects to its configured target).
196    """
197    session = _session()
198    ctrl, provider, _player = _controller(session)
199    provider.on_source_selected = AsyncMock(side_effect=RuntimeError("switching disabled"))
200
201    with pytest.raises(web.HTTPNotFound):
202        await ctrl.serve_audio_source_stream(_request(session_id=session.playback_session_id))
203
204    ctrl.mass.players.deselect_source.assert_awaited_once_with(OWNER_ID)
205
206
207async def test_failing_stream_details_also_takes_the_source_off_the_player() -> None:
208    """The same holds when the plugin claims the source but cannot describe its stream."""
209    session = _session()
210    ctrl, provider, _player = _controller(session)
211    provider.get_stream_details = AsyncMock(side_effect=OSError("daemon gone"))
212
213    with pytest.raises(web.HTTPNotFound):
214        await ctrl.serve_audio_source_stream(_request(session_id=session.playback_session_id))
215
216    ctrl.mass.players.deselect_source.assert_awaited_once_with(OWNER_ID)
217
218
219async def test_a_session_already_superseded_is_not_released() -> None:
220    """A newer session on the player is not this request's to take away."""
221    session = _session()
222    ctrl, provider, _player = _controller(session)
223    provider.on_source_selected = AsyncMock(side_effect=RuntimeError("nope"))
224    # the player moved on to a different session while this request was setting up
225    ctrl.mass.players.get_audio_source_session = MagicMock(return_value=_session())
226
227    with pytest.raises(web.HTTPNotFound):
228        await ctrl.serve_audio_source_stream(_request(session_id=session.playback_session_id))
229
230    ctrl.mass.players.deselect_source.assert_not_awaited()
231