/
/
/
1"""Tests for chained Ogg metadata handling."""
2
3from __future__ import annotations
4
5import struct
6from types import SimpleNamespace
7from typing import TYPE_CHECKING, Any, cast
8from unittest.mock import MagicMock
9
10from music_assistant_models.enums import ContentType, MediaType, StreamType
11from music_assistant_models.media_items import AudioFormat
12from music_assistant_models.streamdetails import StreamDetails
13
14from music_assistant.controllers.streams.audio import StreamsAudio
15from music_assistant.controllers.streams.constants import (
16 STREAMDETAILS_INBAND_TITLE_HANDOFF_KEY,
17 STREAMDETAILS_INBAND_TITLE_KEY,
18)
19
20if TYPE_CHECKING:
21 from collections.abc import AsyncGenerator
22
23 from music_assistant.mass import MusicAssistant
24
25from music_assistant.controllers.streams.ogg_handler import (
26 OGG_FLAG_BOS,
27 OGG_FLAG_EOS,
28 get_chained_ogg_stream,
29)
30
31DEFAULT_VENDOR = "Music Assistant"
32DEFAULT_SERIAL = 1
33OGG_FLAC_MAPPING_PACKET = b"\x7fFLAC\x01\x00\x00\x01fLaC"
34FLAC_METADATA_BLOCK_STREAMINFO = 0
35FLAC_METADATA_BLOCK_VORBIS_COMMENT = 4
36FLAC_METADATA_BLOCK_IS_LAST_FLAG = 0x80
37
38
39class _FakeAudioStreamController:
40 """Minimal audio controller stub for get_chained_ogg_stream tests."""
41
42 def __init__(self, chunks: list[bytes]) -> None:
43 self._chunks = chunks
44
45 async def get_reconnecting_radio_stream(self, url: str) -> AsyncGenerator[bytes]:
46 """Yield the provided Ogg chunks."""
47 del url
48 for chunk in self._chunks:
49 yield chunk
50
51
52def _pack_vorbis_comments(comments: dict[str, str], vendor: str = DEFAULT_VENDOR) -> bytes:
53 """Build a Vorbis comments payload."""
54 result = len(vendor).to_bytes(4, "little")
55 result += vendor.encode("utf-8")
56 result += len(comments).to_bytes(4, "little")
57 for key, value in comments.items():
58 comment = f"{key}={value}".encode()
59 result += len(comment).to_bytes(4, "little")
60 result += comment
61 return result
62
63
64def _pack_flac_metadata_block(block_type: int, payload: bytes, last: bool = False) -> bytes:
65 """Build a FLAC metadata block."""
66 block_header = (FLAC_METADATA_BLOCK_IS_LAST_FLAG if last else 0) | block_type
67 block_length = len(payload).to_bytes(3, "big")
68 return bytes([block_header]) + block_length + payload
69
70
71def _build_ogg_page(
72 segment_data: bytes,
73 *,
74 header_type: int = 0,
75 granule_position: int = 0,
76 serial_number: int = DEFAULT_SERIAL,
77 page_sequence: int = 0,
78) -> bytes:
79 """Build a minimal raw Ogg page."""
80 if len(segment_data) > 255:
81 msg = "Test helper only supports single-segment pages"
82 raise ValueError(msg)
83
84 num_segments = 1 if segment_data else 0
85 page = bytearray(27 + num_segments + len(segment_data))
86 page[0:4] = b"OggS"
87 page[4] = 0
88 page[5] = header_type
89 struct.pack_into("<Q", page, 6, granule_position)
90 struct.pack_into("<I", page, 14, serial_number)
91 struct.pack_into("<I", page, 18, page_sequence)
92 page[22:26] = b"\x00\x00\x00\x00"
93 page[26] = num_segments
94 if segment_data:
95 page[27] = len(segment_data)
96 page[28:] = segment_data
97 return bytes(page)
98
99
100def _build_bos_page(segment_data: bytes = b"", page_sequence: int = 0) -> bytes:
101 """Build a beginning-of-stream page."""
102 return _build_ogg_page(segment_data, header_type=OGG_FLAG_BOS, page_sequence=page_sequence)
103
104
105def _build_eos_page(
106 page_sequence: int,
107 *,
108 granule_position: int = 0,
109 serial_number: int = DEFAULT_SERIAL,
110) -> bytes:
111 """Build an end-of-stream page."""
112 return _build_ogg_page(
113 b"",
114 header_type=OGG_FLAG_EOS,
115 granule_position=granule_position,
116 serial_number=serial_number,
117 page_sequence=page_sequence,
118 )
119
120
121def _build_vorbis_comment_page(comments: dict[str, str], page_sequence: int = 1) -> bytes:
122 """Build a Vorbis comment header page."""
123 return _build_ogg_page(
124 b"\x03vorbis" + _pack_vorbis_comments(comments),
125 page_sequence=page_sequence,
126 )
127
128
129def _build_flac_streaminfo_page(page_sequence: int = 2) -> bytes:
130 """Build a FLAC STREAMINFO metadata page."""
131 streaminfo_payload = b"\x00" * 34
132 streaminfo_block = _pack_flac_metadata_block(
133 FLAC_METADATA_BLOCK_STREAMINFO, streaminfo_payload, last=True
134 )
135 return _build_ogg_page(streaminfo_block, page_sequence=page_sequence)
136
137
138def _build_flac_vorbis_comment_page(comments: dict[str, str], page_sequence: int = 2) -> bytes:
139 """Build a FLAC Vorbis comment metadata page."""
140 payload = _pack_vorbis_comments(comments)
141 block = _pack_flac_metadata_block(FLAC_METADATA_BLOCK_VORBIS_COMMENT, payload, last=True)
142 return _build_ogg_page(block, page_sequence=page_sequence)
143
144
145def _build_incomplete_flac_vorbis_comment_page(page_sequence: int = 2) -> bytes:
146 """Build a truncated FLAC Vorbis comment metadata page."""
147 payload = _pack_vorbis_comments({"TITLE": "Lossless Song"})
148 block = _pack_flac_metadata_block(FLAC_METADATA_BLOCK_VORBIS_COMMENT, payload, last=True)
149 truncated_length = len(block) - 1
150 truncated_block = block[:truncated_length]
151 return _build_ogg_page(truncated_block, page_sequence=page_sequence)
152
153
154async def _collect_metadata(chunks: list[bytes]) -> list[dict[str, str]]:
155 """Run a raw Ogg stream through the handler and collect metadata updates."""
156 metadata_updates: list[dict[str, str]] = []
157 mass = cast(
158 "MusicAssistant",
159 SimpleNamespace(
160 streams=SimpleNamespace(audio=_FakeAudioStreamController(chunks)),
161 ),
162 )
163
164 async for _ in get_chained_ogg_stream(
165 mass,
166 "https://example.invalid/test.ogg",
167 metadata_updates.append,
168 ):
169 pass
170
171 return metadata_updates
172
173
174def _stream_details(*, opt_in: bool) -> StreamDetails:
175 """Return stream details for testing the Ogg metadata handoff."""
176 return StreamDetails(
177 item_id="radio1",
178 provider="test",
179 audio_format=AudioFormat(content_type=ContentType.FLAC, channels=2),
180 media_type=MediaType.RADIO,
181 stream_type=StreamType.IN_BAND,
182 path="https://example.invalid/test.ogg",
183 duration=0,
184 data={STREAMDETAILS_INBAND_TITLE_HANDOFF_KEY: True} if opt_in else None,
185 stream_metadata_update_callback=MagicMock(),
186 )
187
188
189def test_ogg_metadata_handoff_requires_explicit_opt_in() -> None:
190 """Only opted-in providers receive chained-Ogg titles through StreamDetails.data."""
191 audio = StreamsAudio(MagicMock())
192 metadata = {"title": "Some Song", "artist": "Some Artist"}
193
194 opted_in = _stream_details(opt_in=True)
195 audio._handle_inband_metadata(opted_in, metadata)
196 assert opted_in.data is not None
197 assert opted_in.data[STREAMDETAILS_INBAND_TITLE_KEY] == "Some Artist - Some Song"
198 assert opted_in.stream_metadata is None
199
200 legacy = _stream_details(opt_in=False)
201 legacy_mass = cast("Any", audio.mass)
202 legacy_mass.metadata.normalize_radio_artist_name.side_effect = lambda artist: artist
203 legacy_mass.metadata.get_radio_stream_station_image.return_value = None
204 audio._handle_inband_metadata(legacy, metadata)
205 assert legacy.stream_metadata is not None
206 assert legacy.stream_metadata.title == "Some Song"
207 assert legacy.stream_metadata.artist == "Some Artist"
208
209
210async def test_vorbis_comment_page_emits_metadata() -> None:
211 """Vorbis comment header pages should emit metadata."""
212 # Setup
213 comments = {"TITLE": "Song Title", "ARTIST": "Test Artist", "ALBUM": "Live Set"}
214 chunks = [
215 _build_bos_page(),
216 _build_vorbis_comment_page(comments),
217 ]
218
219 # Act
220 metadata_updates = await _collect_metadata(chunks)
221
222 # Assert
223 assert metadata_updates == [
224 {
225 "title": "Song Title",
226 "artist": "Test Artist",
227 "album": "Live Set",
228 }
229 ]
230
231
232async def test_flac_vorbis_comment_page_emits_metadata_after_mapping_bos_page() -> None:
233 """FLAC Vorbis comment pages should emit metadata after the mapping BOS page."""
234 # Setup
235 comments = {"TITLE": "Lossless Song", "ARTIST": "Studio Band", "ALBUM": "Broadcast"}
236 chunks = [
237 _build_bos_page(OGG_FLAC_MAPPING_PACKET),
238 _build_flac_vorbis_comment_page(comments),
239 ]
240
241 # Act
242 metadata_updates = await _collect_metadata(chunks)
243
244 # Assert
245 assert metadata_updates == [
246 {
247 "title": "Lossless Song",
248 "artist": "Studio Band",
249 "album": "Broadcast",
250 }
251 ]
252
253
254async def test_flac_vorbis_comment_page_without_mapping_header_does_not_emit_metadata() -> None:
255 """FLAC metadata pages without Ogg FLAC context should not emit metadata."""
256 # Setup
257 comments = {"TITLE": "Lossless Song", "ARTIST": "Studio Band", "ALBUM": "Broadcast"}
258 chunks = [
259 _build_bos_page(),
260 _build_flac_vorbis_comment_page(comments),
261 ]
262
263 # Act
264 metadata_updates = await _collect_metadata(chunks)
265
266 # Assert
267 assert metadata_updates == []
268
269
270async def test_flac_vorbis_comment_page_emits_metadata_after_streaminfo_page() -> None:
271 """FLAC Vorbis comment pages should still emit metadata after other header pages."""
272 # Setup
273 comments = {"TITLE": "Lossless Song", "ARTIST": "Studio Band", "ALBUM": "Broadcast"}
274 chunks = [
275 _build_bos_page(OGG_FLAC_MAPPING_PACKET),
276 _build_flac_streaminfo_page(page_sequence=1),
277 _build_flac_vorbis_comment_page(comments, page_sequence=2),
278 ]
279
280 # Act
281 metadata_updates = await _collect_metadata(chunks)
282
283 # Assert
284 assert metadata_updates == [
285 {
286 "title": "Lossless Song",
287 "artist": "Studio Band",
288 "album": "Broadcast",
289 }
290 ]
291
292
293async def test_flac_identification_pages_without_comments_do_not_emit_metadata() -> None:
294 """Ogg FLAC mapping BOS pages and STREAMINFO pages should not emit metadata."""
295 # Setup
296 chunks = [
297 _build_bos_page(OGG_FLAC_MAPPING_PACKET),
298 _build_flac_streaminfo_page(),
299 ]
300
301 # Act
302 metadata_updates = await _collect_metadata(chunks)
303
304 # Assert
305 assert metadata_updates == []
306
307
308async def test_incomplete_flac_vorbis_comment_page_does_not_emit_metadata() -> None:
309 """Incomplete FLAC Vorbis comment blocks should be skipped."""
310 # Setup
311 chunks = [
312 _build_bos_page(OGG_FLAC_MAPPING_PACKET),
313 _build_incomplete_flac_vorbis_comment_page(),
314 ]
315
316 # Act
317 metadata_updates = await _collect_metadata(chunks)
318
319 # Assert
320 assert metadata_updates == []
321
322
323async def test_flac_metadata_is_emitted_across_chained_streams() -> None:
324 """FLAC metadata should be emitted again after an EOS/BOS chain boundary."""
325 # Setup
326 first_chain_comments = {"TITLE": "First Song", "ARTIST": "First Artist"}
327 second_chain_comments = {"TITLE": "Second Song", "ARTIST": "Second Artist"}
328 chunks = [
329 _build_bos_page(OGG_FLAC_MAPPING_PACKET, page_sequence=0),
330 _build_flac_vorbis_comment_page(first_chain_comments, page_sequence=1),
331 _build_eos_page(page_sequence=2, granule_position=100),
332 _build_bos_page(OGG_FLAC_MAPPING_PACKET, page_sequence=0),
333 _build_flac_vorbis_comment_page(second_chain_comments, page_sequence=1),
334 ]
335
336 # Act
337 metadata_updates = await _collect_metadata(chunks)
338
339 # Assert
340 assert metadata_updates == [
341 {
342 "title": "First Song",
343 "artist": "First Artist",
344 },
345 {
346 "title": "Second Song",
347 "artist": "Second Artist",
348 },
349 ]
350