/
/
/
1"""Tests for HLS master playlist selection on the streams controller."""
2
3from __future__ import annotations
4
5from types import TracebackType
6from typing import Self
7from unittest.mock import MagicMock
8
9import pytest
10
11from music_assistant.controllers.streams.audio import StreamsAudio
12
13MASTER_PLAYLIST = (
14 "#EXTM3U\n"
15 '#EXT-X-STREAM-INF:BANDWIDTH=64000,CODECS="mp4a.40.2"\n'
16 "https://radio.example.com/low.m3u8\n"
17 '#EXT-X-STREAM-INF:BANDWIDTH=320000,CODECS="mp4a.40.2"\n'
18 "https://radio.example.com/high.m3u8\n"
19)
20
21
22class _FakeResponse:
23 """Stand-in for the aiohttp response of a master playlist fetch."""
24
25 def __init__(self, raw_data: bytes, charset: str | None) -> None:
26 self._raw_data = raw_data
27 self.charset = charset
28
29 async def __aenter__(self) -> Self:
30 return self
31
32 async def __aexit__(
33 self,
34 exc_type: type[BaseException] | None,
35 exc_val: BaseException | None,
36 exc_tb: TracebackType | None,
37 ) -> None:
38 return None
39
40 def raise_for_status(self) -> None:
41 """Accept the response as a successful one."""
42
43 async def read(self) -> bytes:
44 return self._raw_data
45
46
47def _streams_audio(raw_data: bytes, charset: str | None) -> StreamsAudio:
48 """Build a streams audio controller whose HTTP session serves the given playlist."""
49 mass = MagicMock()
50 mass.http_session_no_ssl.get = MagicMock(return_value=_FakeResponse(raw_data, charset))
51 return StreamsAudio(mass)
52
53
54class TestGetHlsSubstream:
55 """get_hls_substream picks the best child playlist of an HLS master playlist."""
56
57 @pytest.mark.asyncio
58 async def test_unknown_charset_falls_back_to_detection(self) -> None:
59 """
60 A charset the remote server made up must not break substream selection.
61
62 Stations do send names Python has no codec for, which decode() answers with a
63 LookupError that no caller on this path catches.
64 """
65 controller = _streams_audio(MASTER_PLAYLIST.encode(), charset="utf8mb4")
66 substream = await controller.get_hls_substream("https://radio.example.com/master.m3u8")
67 assert substream.path == "https://radio.example.com/high.m3u8"
68
69 @pytest.mark.asyncio
70 async def test_undecodable_byte_degrades_instead_of_raising(self) -> None:
71 """One bad byte costs a character, not the whole stream."""
72 raw_data = MASTER_PLAYLIST.encode().replace(b"#EXTM3U", b"#EXTM3U\n#\xff")
73 controller = _streams_audio(raw_data, charset="utf-8")
74 substream = await controller.get_hls_substream("https://radio.example.com/master.m3u8")
75 assert substream.path == "https://radio.example.com/high.m3u8"
76