music-assistant-server

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