/
/
/
1"""Tests for ICY radio stream reconnection on disconnect."""
2
3from __future__ import annotations
4
5import asyncio
6from typing import Any
7from unittest.mock import AsyncMock, MagicMock
8
9import aiohttp
10import pytest
11from music_assistant_models.enums import ContentType, MediaType, StreamType
12from music_assistant_models.errors import (
13 InvalidDataError,
14 MediaNotFoundError,
15 ProviderPermissionDenied,
16 RetriesExhausted,
17)
18from music_assistant_models.media_items import AudioFormat
19from music_assistant_models.streamdetails import MultiPartPath, StreamDetails
20
21from music_assistant.controllers.streams.audio import StreamsAudio
22
23_META_INT = 4
24
25
26class _FakeContent:
27 """Minimal stand-in for aiohttp StreamReader content, driven by a script of frames."""
28
29 def __init__(self, frames: list[bytes | Exception]) -> None:
30 self._frames = list(frames)
31
32 async def readexactly(self, _n: int) -> bytes:
33 if not self._frames:
34 raise asyncio.IncompleteReadError(b"", _n)
35 item = self._frames.pop(0)
36 if isinstance(item, Exception):
37 raise item
38 return item
39
40
41class _FakeConnCtx:
42 """Async context manager yielding a fake ICY response."""
43
44 def __init__(
45 self, frames: list[bytes | Exception], status: int = 200, meta_int: str = str(_META_INT)
46 ) -> None:
47 self._resp = MagicMock()
48 self._resp.headers = {"icy-metaint": meta_int}
49 self._resp.content = _FakeContent(frames)
50 if status >= 400:
51 self._resp.raise_for_status.side_effect = aiohttp.ClientResponseError(
52 MagicMock(), (), status=status
53 )
54
55 async def __aenter__(self) -> Any:
56 return self._resp
57
58 async def __aexit__(self, *_exc: object) -> bool:
59 return False
60
61
62def _radio_streamdetails() -> StreamDetails:
63 return StreamDetails(
64 provider="test_provider",
65 item_id="radio1",
66 audio_format=AudioFormat(content_type=ContentType.MP3),
67 media_type=MediaType.RADIO,
68 stream_type=StreamType.ICY,
69 path="http://example.test/radio.mp3",
70 )
71
72
73@pytest.mark.asyncio
74async def test_icy_stream_reconnects_after_disconnect(monkeypatch: pytest.MonkeyPatch) -> None:
75 """A mid-stream disconnect transparently reconnects and keeps yielding audio."""
76 audio = StreamsAudio(MagicMock())
77
78 # connection #1 yields one audio frame then drops; connection #2 yields another.
79 connections = [
80 _FakeConnCtx([b"AAAA", b"\x00", asyncio.IncompleteReadError(b"", _META_INT)]),
81 _FakeConnCtx([b"BBBB", b"\x00", asyncio.IncompleteReadError(b"", _META_INT)]),
82 ]
83 connect_calls = 0
84
85 def _fake_connect(*_args: Any, **_kwargs: Any) -> _FakeConnCtx:
86 nonlocal connect_calls
87 ctx = connections[connect_calls]
88 connect_calls += 1
89 return ctx
90
91 monkeypatch.setattr(audio, "_connect_radio_stream", _fake_connect)
92
93 chunks: list[bytes] = []
94 async for chunk in audio.get_icy_radio_stream(
95 "http://example.test/radio.mp3", _radio_streamdetails()
96 ):
97 chunks.append(chunk)
98 if len(chunks) == 2:
99 # got one frame from each connection - reconnection proven
100 break
101
102 assert chunks == [b"AAAA", b"BBBB"]
103 assert connect_calls == 2
104
105
106@pytest.mark.asyncio
107async def test_icy_stream_raises_on_http_404(monkeypatch: pytest.MonkeyPatch) -> None:
108 """A 404 response is terminal and surfaces as MediaNotFoundError, not a reconnect."""
109 audio = StreamsAudio(MagicMock())
110
111 monkeypatch.setattr(
112 audio, "_connect_radio_stream", lambda *_a, **_k: _FakeConnCtx([], status=404)
113 )
114
115 with pytest.raises(MediaNotFoundError):
116 async for _chunk in audio.get_icy_radio_stream(
117 "http://example.test/radio.mp3", _radio_streamdetails()
118 ):
119 pass
120
121
122@pytest.mark.asyncio
123async def test_icy_stream_gives_up_when_no_data(monkeypatch: pytest.MonkeyPatch) -> None:
124 """A connection that never delivers audio bails out instead of reconnecting forever."""
125 audio = StreamsAudio(MagicMock())
126 connect_calls = 0
127
128 def _fake_connect(*_args: Any, **_kwargs: Any) -> _FakeConnCtx:
129 nonlocal connect_calls
130 connect_calls += 1
131 # empty frames -> readexactly immediately raises IncompleteReadError (no data)
132 return _FakeConnCtx([])
133
134 monkeypatch.setattr(audio, "_connect_radio_stream", _fake_connect)
135 # don't actually wait out the backoff between reconnect attempts
136 monkeypatch.setattr(asyncio, "sleep", AsyncMock())
137
138 with pytest.raises(RetriesExhausted):
139 async for _chunk in audio.get_icy_radio_stream(
140 "http://example.test/radio.mp3", _radio_streamdetails()
141 ):
142 pass
143
144 # bails after the budget is exhausted rather than spinning indefinitely
145 assert connect_calls < 50
146
147
148@pytest.mark.asyncio
149async def test_icy_stream_invalid_metaint_is_terminal(monkeypatch: pytest.MonkeyPatch) -> None:
150 """A malformed icy-metaint header raises a controlled InvalidDataError, not a raw ValueError."""
151 audio = StreamsAudio(MagicMock())
152 monkeypatch.setattr(
153 audio, "_connect_radio_stream", lambda *_a, **_k: _FakeConnCtx([], meta_int="not-a-number")
154 )
155
156 with pytest.raises(InvalidDataError):
157 async for _chunk in audio.get_icy_radio_stream(
158 "http://example.test/radio.mp3", _radio_streamdetails()
159 ):
160 pass
161
162
163@pytest.mark.asyncio
164async def test_icy_failover_on_http_error(monkeypatch: pytest.MonkeyPatch) -> None:
165 """A mirror returning an HTTP error fails over to the next mirror URL."""
166 audio = StreamsAudio(MagicMock())
167 seen: list[str] = []
168
169 def _connect(url: str, *_a: Any, **_k: Any) -> _FakeConnCtx:
170 seen.append(url)
171 if url == "http://primary.test/stream":
172 return _FakeConnCtx([], status=403)
173 return _FakeConnCtx([b"BBBB"])
174
175 monkeypatch.setattr(audio, "_connect_radio_stream", _connect)
176 monkeypatch.setattr(asyncio, "sleep", AsyncMock())
177
178 mirrors = [
179 MultiPartPath(path="http://primary.test/stream"),
180 MultiPartPath(path="http://backup.test/stream"),
181 ]
182 chunks: list[bytes] = []
183 async for chunk in audio.get_reconnecting_icy_radio_stream(mirrors, _radio_streamdetails()):
184 chunks.append(chunk)
185 break
186
187 assert chunks == [b"BBBB"]
188 assert seen == ["http://primary.test/stream", "http://backup.test/stream"]
189
190
191@pytest.mark.asyncio
192async def test_icy_failover_raises_when_all_mirrors_fail(monkeypatch: pytest.MonkeyPatch) -> None:
193 """When every mirror is unreachable the last terminal error is raised, not spun on forever."""
194 audio = StreamsAudio(MagicMock())
195 connect_calls = 0
196
197 def _connect(_url: str, *_a: Any, **_k: Any) -> _FakeConnCtx:
198 nonlocal connect_calls
199 connect_calls += 1
200 return _FakeConnCtx([], status=403)
201
202 monkeypatch.setattr(audio, "_connect_radio_stream", _connect)
203 monkeypatch.setattr(asyncio, "sleep", AsyncMock())
204
205 mirrors = [
206 MultiPartPath(path="http://a.test/stream"),
207 MultiPartPath(path="http://b.test/stream"),
208 ]
209 with pytest.raises(ProviderPermissionDenied):
210 async for _chunk in audio.get_reconnecting_icy_radio_stream(
211 mirrors, _radio_streamdetails()
212 ):
213 pass
214
215 # bounded failover - does not retry indefinitely
216 assert connect_calls <= len(mirrors) * 2 + 1
217
218
219@pytest.mark.asyncio
220async def test_icy_single_mirror_propagates_terminal_error(monkeypatch: pytest.MonkeyPatch) -> None:
221 """A single mirror URL delegates to get_icy_radio_stream and surfaces its terminal error."""
222 audio = StreamsAudio(MagicMock())
223 monkeypatch.setattr(
224 audio, "_connect_radio_stream", lambda *_a, **_k: _FakeConnCtx([], status=404)
225 )
226
227 with pytest.raises(MediaNotFoundError):
228 async for _chunk in audio.get_reconnecting_icy_radio_stream(
229 [MultiPartPath(path="http://only.test/stream")], _radio_streamdetails()
230 ):
231 pass
232
233
234@pytest.mark.asyncio
235async def test_icy_failover_requires_at_least_one_url() -> None:
236 """An empty mirror list is rejected with InvalidDataError."""
237 audio = StreamsAudio(MagicMock())
238 with pytest.raises(InvalidDataError):
239 async for _chunk in audio.get_reconnecting_icy_radio_stream([], _radio_streamdetails()):
240 pass
241