/
/
/
1"""Tests for music_assistant.helpers.audio."""
2
3from __future__ import annotations
4
5import asyncio
6from collections.abc import AsyncGenerator
7from types import SimpleNamespace
8from unittest.mock import MagicMock
9
10import pytest
11from music_assistant_models.enums import ContentType
12from music_assistant_models.media_items import AudioFormat
13
14from music_assistant.helpers.audio import (
15 build_concat_filelist,
16 calculate_content_length,
17 parse_loudnorm,
18 realtime_pcm_pacer,
19 resolve_output_player_ids,
20)
21from music_assistant.helpers.ffmpeg import DEFAULT_MP3_BIT_RATE
22
23
24def test_resolve_output_player_ids_resolves_parents_and_duplicates() -> None:
25 """Output destinations use visible protocol parents without duplicates."""
26 mass = MagicMock()
27 players = {
28 "leader": SimpleNamespace(protocol_parent_id=None),
29 "protocol-child": SimpleNamespace(protocol_parent_id="child"),
30 "child": SimpleNamespace(protocol_parent_id=None),
31 }
32 mass.players.get_player.side_effect = players.get
33
34 result = resolve_output_player_ids(
35 mass,
36 ("leader", "leader", "protocol-child", "child", "missing", "missing"),
37 )
38
39 assert result == {"leader", "child", "missing"}
40
41
42def test_mp3_content_length_uses_encoder_bitrate() -> None:
43 """MP3 size estimation uses the bitrate configured for FFmpeg encoding."""
44 seconds = 2
45 assert calculate_content_length(
46 AudioFormat(content_type=ContentType.MP3),
47 seconds,
48 ) == int(((DEFAULT_MP3_BIT_RATE * 1000) / 8) * seconds)
49
50
51def test_build_concat_filelist_plain_paths() -> None:
52 """Paths without special characters are wrapped verbatim, one per line."""
53 result = build_concat_filelist(["/music/a.mp3", "/music/b.mp3"])
54 assert result == "file '/music/a.mp3'\nfile '/music/b.mp3'\n"
55
56
57def test_build_concat_filelist_escapes_apostrophes() -> None:
58 r"""
59 A single quote in the path is escaped as '\'' for the concat demuxer.
60
61 Regression test for multipart playback failing on paths such as
62 "Amelia Bedelia's", where the demuxer truncated the path at the apostrophe.
63 """
64 path = "/audiobooks/Herman Parish - Young Amelia Bedelia's Audio Collection/01.mp3"
65 result = build_concat_filelist([path])
66 assert (
67 result
68 == "file '/audiobooks/Herman Parish - Young Amelia Bedelia'\\''s Audio Collection/01.mp3'\n"
69 )
70 # The original apostrophe must survive once the escaping is unwrapped.
71 assert path in result.replace("'\\''", "'")
72
73
74def test_build_concat_filelist_escapes_multiple_apostrophes() -> None:
75 """Every apostrophe in a path is escaped, not just the first."""
76 result = build_concat_filelist(["/x/it's a, b's & c's.mp3"])
77 assert result == "file '/x/it'\\''s a, b'\\''s & c'\\''s.mp3'\n"
78
79
80# verbatim ffmpeg 7.1 output: the report is a block below the marker line, not inline
81FFMPEG_LOUDNORM_OUTPUT = b"""[out#0/null @ 0x93b] Output stream
82[Parsed_loudnorm_0 @ 0x93b41d440] \n{
83\t"input_i" : "-17.86",
84\t"input_tp" : "-1.89",
85\t"input_lra" : "0.40",
86\t"input_thresh" : "-27.86",
87\t"output_i" : "-23.92",
88\t"normalization_type" : "dynamic",
89\t"target_offset" : "-0.08"
90}
91size=N/A time=00:00:03.20 bitrate=N/A speed= 86x
92"""
93
94
95def test_parse_loudnorm_reads_the_measurement_ffmpeg_actually_prints() -> None:
96 """The integrated loudness is read from loudnorm's own JSON report block."""
97 assert parse_loudnorm(FFMPEG_LOUDNORM_OUTPUT) == -17.86
98
99
100def test_parse_loudnorm_accepts_a_decoded_string() -> None:
101 """Callers that already decoded the output get the same measurement."""
102 assert parse_loudnorm(FFMPEG_LOUDNORM_OUTPUT.decode()) == -17.86
103
104
105def test_parse_loudnorm_without_a_report_returns_none() -> None:
106 """Output from a run that never reached the filter carries no measurement."""
107 assert parse_loudnorm(b"ffmpeg: Invalid data found when processing input") is None
108
109
110def test_parse_loudnorm_with_a_truncated_report_returns_none() -> None:
111 """A report cut off mid-object is not mistaken for a measurement."""
112 assert parse_loudnorm(b'[Parsed_loudnorm_0 @ 0x1] \n{\n\t"input_i" : "-17.8') is None
113
114
115def test_parse_loudnorm_treats_digital_silence_as_no_measurement() -> None:
116 """A silent clip reports -inf, which is the absence of a level, not a level."""
117 silent = FFMPEG_LOUDNORM_OUTPUT.replace(b'"-17.86"', b'"-inf"')
118 assert parse_loudnorm(silent) is None
119
120
121def test_parse_loudnorm_reads_a_report_from_further_down_the_filter_chain() -> None:
122 """The marker carries the filter's position, which is not zero behind another filter."""
123 chained = FFMPEG_LOUDNORM_OUTPUT.replace(b"[Parsed_loudnorm_0 @", b"[Parsed_loudnorm_1 @")
124 assert parse_loudnorm(chained) == -17.86
125
126
127@pytest.mark.asyncio
128async def test_realtime_pcm_pacer_grants_bounded_initial_burst() -> None:
129 """The pacer lets a bounded head start through unpaced, then enforces realtime."""
130 # tiny format keeps the test fast: 16000 B/s, so 1s of audio is 16000 bytes
131 pcm_format = AudioFormat(
132 content_type=ContentType.PCM_S16LE,
133 sample_rate=8000,
134 bit_depth=16,
135 channels=1,
136 )
137
138 async def _instant_producer() -> AsyncGenerator[bytes]:
139 for _ in range(10):
140 yield b"\x00" * 1600 # 0.1s of audio per chunk, produced instantly
141
142 loop = asyncio.get_running_loop()
143 start = loop.time()
144 async for _ in realtime_pcm_pacer(_instant_producer(), pcm_format):
145 pass
146 elapsed = loop.time() - start
147
148 # 1.0s of audio with a 0.5s burst allowance should take ~0.5s: clearly less
149 # than realtime (burst granted) but still paced (not instant). Bounds are
150 # deliberately wide to stay robust on loaded CI runners.
151 assert 0.3 < elapsed < 0.9
152