/
/
/
1"""Tests for radio stream resolution on the streams controller."""
2
3from __future__ import annotations
4
5import asyncio
6from typing import Any, cast
7from unittest.mock import AsyncMock, MagicMock
8
9import aiohttp
10import pytest
11from music_assistant_models.enums import StreamType
12from music_assistant_models.errors import InvalidDataError
13
14from music_assistant.controllers.streams.audio import StreamsAudio
15
16PLS_BODY = b"[playlist]\nNumberOfEntries=1\nFile1=http://radio.example.com/stream\n"
17HLS_BODY = b"#EXTM3U\n#EXT-X-VERSION:3\n#EXTINF:10,\nsegment0.aac\n"
18
19
20class _FakeContent:
21 """Minimal stand-in for aiohttp StreamReader content."""
22
23 def __init__(
24 self, raw_data: bytes | Exception, chunk_size: int | None = None, delay: float = 0
25 ) -> None:
26 self._raw_data = raw_data
27 self._chunk_size = chunk_size
28 self._delay = delay
29 self._pos = 0
30
31 async def read(self, n: int) -> bytes:
32 if self._delay:
33 await asyncio.sleep(self._delay)
34 if isinstance(self._raw_data, Exception):
35 raise self._raw_data
36 # like the real stream, a read hands over what has arrived so far
37 available = len(self._raw_data) - self._pos
38 size = min(n, self._chunk_size, available) if self._chunk_size else min(n, available)
39 chunk = self._raw_data[self._pos : self._pos + size]
40 self._pos += size
41 return chunk
42
43
44class _FakeConnCtx:
45 """Async context manager yielding a fake radio probe response."""
46
47 def __init__(
48 self,
49 headers: dict[str, str],
50 raw_data: bytes | Exception = b"",
51 charset: str | None = None,
52 chunk_size: int | None = None,
53 delay: float = 0,
54 ) -> None:
55 self._resp = MagicMock()
56 self._resp.headers = headers
57 self._resp.charset = charset
58 self._resp.content = _FakeContent(raw_data, chunk_size, delay)
59
60 async def __aenter__(self) -> Any:
61 return self._resp
62
63 async def __aexit__(self, *_exc: object) -> bool:
64 return False
65
66
67def _streams_audio() -> StreamsAudio:
68 """Build a streams audio controller with an empty resolved-radio cache."""
69 mass = MagicMock()
70 mass.cache.get = AsyncMock(return_value=None)
71 mass.cache.set = AsyncMock()
72 return StreamsAudio(mass)
73
74
75@pytest.mark.asyncio
76async def test_playlist_is_unwrapped_from_the_probe_response(
77 monkeypatch: pytest.MonkeyPatch,
78) -> None:
79 """
80 The playlist is read from the probe response instead of being fetched again.
81
82 A fetch of its own would go out over another session and user agent than the rest of
83 the radio paths use, which a host is free to answer differently.
84 """
85 audio = _streams_audio()
86 responses = {
87 "http://radio.example.com/station.pls": _FakeConnCtx(
88 {"content-type": "audio/x-scpls"}, PLS_BODY, chunk_size=8
89 ),
90 "http://radio.example.com/stream": _FakeConnCtx({"icy-metaint": "16000"}),
91 }
92 requested: list[str] = []
93
94 def _fake_connect(url: str, **_kwargs: Any) -> _FakeConnCtx:
95 requested.append(url)
96 return responses[url]
97
98 monkeypatch.setattr(audio, "_connect_radio_stream", _fake_connect)
99
100 result = await audio.resolve_radio_stream("http://radio.example.com/station.pls")
101
102 assert result == ("http://radio.example.com/stream", StreamType.ICY)
103 assert requested == [
104 "http://radio.example.com/station.pls",
105 "http://radio.example.com/stream",
106 ]
107 cast("MagicMock", audio.mass).http_session.get.assert_not_called()
108
109
110@pytest.mark.asyncio
111async def test_hls_playlist_resolves_as_hls(monkeypatch: pytest.MonkeyPatch) -> None:
112 """An HLS station without a telling URL is recognised by its content type."""
113 audio = _streams_audio()
114 response = _FakeConnCtx({"content-type": "application/vnd.apple.mpegurl"}, HLS_BODY)
115 monkeypatch.setattr(audio, "_connect_radio_stream", lambda *_args, **_kwargs: response)
116
117 result = await audio.resolve_radio_stream("http://radio.example.com/live")
118
119 assert result == ("http://radio.example.com/live", StreamType.HLS)
120 cast("MagicMock", audio.mass).http_session.get.assert_not_called()
121
122
123@pytest.mark.asyncio
124@pytest.mark.parametrize(
125 ("url", "content_type"),
126 [
127 ("http://radio.example.com/live", "audio/x-mpegurl"),
128 ("http://radio.example.com/live.m3u8", "application/octet-stream"),
129 ],
130)
131async def test_hls_without_version_tag_still_resolves_as_hls(
132 monkeypatch: pytest.MonkeyPatch,
133 url: str,
134 content_type: str,
135) -> None:
136 """A version-less HLS media playlist is recognised by its required tag."""
137 audio = _streams_audio()
138 body = b"#EXTM3U\n#EXT-X-TARGETDURATION:10\n#EXTINF:10.0,\nsegment0.aac\n"
139 response = _FakeConnCtx({"content-type": content_type}, body)
140 monkeypatch.setattr(audio, "_connect_radio_stream", lambda *_args, **_kwargs: response)
141
142 result = await audio.resolve_radio_stream(url)
143
144 assert result == (url, StreamType.HLS)
145
146
147@pytest.mark.asyncio
148async def test_content_type_is_matched_case_insensitively(
149 monkeypatch: pytest.MonkeyPatch,
150) -> None:
151 """A server shouting its media type in mixed case is understood all the same."""
152 audio = _streams_audio()
153 response = _FakeConnCtx({"content-type": "Application/Vnd.Apple.Mpegurl"}, HLS_BODY)
154 monkeypatch.setattr(audio, "_connect_radio_stream", lambda *_args, **_kwargs: response)
155
156 result = await audio.resolve_radio_stream("http://radio.example.com/live")
157
158 assert result == ("http://radio.example.com/live", StreamType.HLS)
159
160
161@pytest.mark.asyncio
162async def test_trickling_playlist_body_gives_up_instead_of_stalling(
163 monkeypatch: pytest.MonkeyPatch,
164) -> None:
165 """A server feeding the playlist a byte at a time cannot hold up resolving."""
166 audio = _streams_audio()
167 response = _FakeConnCtx({"content-type": "audio/x-scpls"}, PLS_BODY, chunk_size=1, delay=0.02)
168 monkeypatch.setattr(audio, "_connect_radio_stream", lambda *_args, **_kwargs: response)
169 # a byte at a time at that pace runs well past the budget
170 monkeypatch.setattr("music_assistant.controllers.streams.audio.PLAYLIST_READ_TIMEOUT", 0.05)
171
172 with pytest.raises(InvalidDataError, match="Timeout"):
173 await audio.resolve_radio_stream("http://radio.example.com/station.pls")
174
175 cast("MagicMock", audio.mass).cache.set.assert_not_called()
176
177
178@pytest.mark.asyncio
179async def test_truncated_playlist_is_not_cached_as_a_direct_stream(
180 monkeypatch: pytest.MonkeyPatch,
181) -> None:
182 """A playlist body that dies mid-transfer fails instead of being streamed as audio."""
183 audio = _streams_audio()
184 response = _FakeConnCtx(
185 {"content-type": "audio/x-scpls"}, aiohttp.ClientPayloadError("connection closed")
186 )
187 monkeypatch.setattr(audio, "_connect_radio_stream", lambda *_args, **_kwargs: response)
188
189 with pytest.raises(InvalidDataError):
190 await audio.resolve_radio_stream("http://radio.example.com/station.pls")
191
192 cast("MagicMock", audio.mass).cache.set.assert_not_called()
193