/
/
/
1"""Tests for the flow mode sample-rate selection and restart logic."""
2
3from __future__ import annotations
4
5from unittest.mock import MagicMock, patch
6
7import pytest
8from music_assistant_models.enums import ContentType, MediaType, VolumeNormalizationMode
9from music_assistant_models.errors import AudioError
10from music_assistant_models.media_items import AudioFormat
11
12from music_assistant.constants import (
13 CONF_FLOW_MODE_SAMPLE_RATE,
14 FLOW_MODE_SAMPLE_RATE_48000,
15 FLOW_MODE_SAMPLE_RATE_96000,
16 FLOW_MODE_SAMPLE_RATE_BIT_PERFECT,
17 FLOW_MODE_SAMPLE_RATE_HIGHEST,
18 FLOW_MODE_SAMPLE_RATE_SMART,
19)
20from music_assistant.controllers.streams.audio import (
21 StreamsAudio,
22 _snap_supported_rate_down,
23 _snap_supported_rate_up,
24)
25
26# --- snap helpers ---
27
28
29@pytest.mark.parametrize(
30 ("target", "supported", "expected"),
31 [
32 (48000, [44100, 48000, 96000], 48000), # exact match
33 (50000, [44100, 48000, 96000], 96000), # snap up to next higher
34 (200000, [44100, 48000, 96000], 96000), # no higher; fall back to max
35 (88200, [44100, 48000], 44100), # preserve the source sample-rate family
36 (40000, [44100, 48000], 44100), # snap up to lowest higher
37 ],
38)
39def test_snap_supported_rate_up(target: int, supported: list[int], expected: int) -> None:
40 """_snap_supported_rate_up prefers a higher rate, then a source-family divisor."""
41 assert _snap_supported_rate_up(target, supported) == expected
42
43
44@pytest.mark.parametrize(
45 ("target", "supported", "expected"),
46 [
47 (48000, [44100, 48000, 96000], 48000), # exact match
48 (60000, [44100, 48000, 96000], 48000), # snap down to highest lower
49 (40000, [48000, 96000], 48000), # no lower; fall back to min
50 (200000, [44100, 48000], 48000), # snap down (well above all)
51 ],
52)
53def test_snap_supported_rate_down(target: int, supported: list[int], expected: int) -> None:
54 """_snap_supported_rate_down picks the highest supported <= target, else min."""
55 assert _snap_supported_rate_down(target, supported) == expected
56
57
58# --- select_flow_pcm_format ---
59
60
61@pytest.mark.asyncio
62async def test_select_flow_pcm_format_highest() -> None:
63 """'highest' mode picks the player's max supported sample rate."""
64 audio = _make_streams_audio()
65 player = _make_player(
66 supported=[(44100, 16), (48000, 24), (96000, 24)],
67 flow_mode=FLOW_MODE_SAMPLE_RATE_HIGHEST,
68 )
69 fmt = await audio.select_flow_pcm_format(player)
70 assert fmt.sample_rate == 96000
71
72
73@pytest.mark.asyncio
74async def test_select_flow_pcm_format_48000_exact_match() -> None:
75 """'48000' mode picks 48 kHz when supported."""
76 audio = _make_streams_audio()
77 player = _make_player(
78 supported=[(44100, 16), (48000, 24), (96000, 24)],
79 flow_mode=FLOW_MODE_SAMPLE_RATE_48000,
80 )
81 fmt = await audio.select_flow_pcm_format(player)
82 assert fmt.sample_rate == 48000
83
84
85@pytest.mark.asyncio
86async def test_select_flow_pcm_format_48000_snaps_down_when_unsupported() -> None:
87 """'48000' mode falls back to 44.1 kHz when the player can't do 48 kHz."""
88 audio = _make_streams_audio()
89 player = _make_player(
90 supported=[(44100, 16)],
91 flow_mode=FLOW_MODE_SAMPLE_RATE_48000,
92 )
93 fmt = await audio.select_flow_pcm_format(player)
94 assert fmt.sample_rate == 44100
95
96
97@pytest.mark.asyncio
98async def test_select_flow_pcm_format_96000_falls_back_to_highest_below() -> None:
99 """'96000' mode snaps down to the highest supported rate <= 96000."""
100 audio = _make_streams_audio()
101 player = _make_player(
102 supported=[(44100, 16), (48000, 24)],
103 flow_mode=FLOW_MODE_SAMPLE_RATE_96000,
104 )
105 fmt = await audio.select_flow_pcm_format(player)
106 assert fmt.sample_rate == 48000
107
108
109@pytest.mark.asyncio
110async def test_select_flow_pcm_format_96000_snaps_up_when_only_higher_supported() -> None:
111 """'96000' mode falls back to the lowest supported rate above 96 kHz when no lower exists."""
112 audio = _make_streams_audio()
113 player = _make_player(
114 supported=[(192000, 24)],
115 flow_mode=FLOW_MODE_SAMPLE_RATE_96000,
116 )
117 fmt = await audio.select_flow_pcm_format(player)
118 assert fmt.sample_rate == 192000
119
120
121@pytest.mark.asyncio
122async def test_select_flow_pcm_format_smart_anchors_on_start_track() -> None:
123 """'smart' mode anchors on the first track's sample rate when supported."""
124 audio = _make_streams_audio()
125 player = _make_player(
126 supported=[(44100, 16), (48000, 24), (96000, 24)],
127 flow_mode=FLOW_MODE_SAMPLE_RATE_SMART,
128 )
129 streamdetails = _make_streamdetails(sample_rate=44100, bit_depth=16)
130 fmt = await audio.select_flow_pcm_format(player, start_streamdetails=streamdetails)
131 assert fmt.sample_rate == 44100
132
133
134@pytest.mark.asyncio
135async def test_select_flow_pcm_format_smart_snaps_up_unsupported_rate() -> None:
136 """When the anchor rate isn't supported, smart snaps up to the next higher rate."""
137 audio = _make_streams_audio()
138 player = _make_player(
139 supported=[(48000, 24), (96000, 24)],
140 flow_mode=FLOW_MODE_SAMPLE_RATE_SMART,
141 )
142 streamdetails = _make_streamdetails(sample_rate=44100, bit_depth=16)
143 fmt = await audio.select_flow_pcm_format(player, start_streamdetails=streamdetails)
144 assert fmt.sample_rate == 48000
145
146
147@pytest.mark.asyncio
148async def test_select_flow_pcm_format_smart_without_start_format_uses_max() -> None:
149 """Smart mode with no start_streamdetails falls back to the player's highest rate."""
150 audio = _make_streams_audio()
151 player = _make_player(
152 supported=[(44100, 16), (96000, 24)],
153 flow_mode=FLOW_MODE_SAMPLE_RATE_SMART,
154 )
155 fmt = await audio.select_flow_pcm_format(player)
156 assert fmt.sample_rate == 96000
157
158
159@pytest.mark.asyncio
160async def test_select_flow_pcm_format_uses_fallback_rate_without_start_format() -> None:
161 """A caller-provided fallback rate is used when stream details are unavailable."""
162 audio = _make_streams_audio()
163 player = _make_player(
164 supported=[(44100, 16), (48000, 24)],
165 flow_mode=FLOW_MODE_SAMPLE_RATE_SMART,
166 )
167 fmt = await audio.select_flow_pcm_format(player, fallback_sample_rate=44100)
168 assert fmt.sample_rate == 44100
169
170
171@pytest.mark.asyncio
172async def test_select_flow_pcm_format_uses_common_output_rates() -> None:
173 """A shared flow only selects sample rates supported by every output player."""
174 audio = _make_streams_audio()
175 leader = _make_player(
176 supported=[(44100, 24), (48000, 24)],
177 flow_mode=FLOW_MODE_SAMPLE_RATE_SMART,
178 )
179 member = _make_player(
180 supported=[(44100, 16)],
181 flow_mode=FLOW_MODE_SAMPLE_RATE_SMART,
182 )
183 streamdetails = _make_streamdetails(sample_rate=48000, bit_depth=24)
184
185 fmt = await audio.select_flow_pcm_format(
186 leader,
187 start_streamdetails=streamdetails,
188 output_players=(leader, member),
189 )
190
191 assert fmt.sample_rate == 44100
192
193
194@pytest.mark.asyncio
195async def test_select_flow_pcm_format_rejects_outputs_without_common_rate() -> None:
196 """A shared flow requires at least one sample rate supported by every output."""
197 audio = _make_streams_audio()
198 leader = _make_player(
199 supported=[(48000, 24)],
200 flow_mode=FLOW_MODE_SAMPLE_RATE_SMART,
201 )
202 member = _make_player(
203 supported=[(44100, 16)],
204 flow_mode=FLOW_MODE_SAMPLE_RATE_SMART,
205 )
206
207 with pytest.raises(AudioError, match="do not share"):
208 await audio.select_flow_pcm_format(
209 leader,
210 output_players=(leader, member),
211 )
212
213
214@pytest.mark.asyncio
215async def test_select_flow_pcm_format_uses_headroom_for_any_output_dsp() -> None:
216 """DSP on any shared output requires float headroom for the complete flow."""
217 audio = _make_streams_audio()
218 leader = _make_player(
219 supported=[(44100, 24), (48000, 24)],
220 flow_mode=FLOW_MODE_SAMPLE_RATE_SMART,
221 )
222 member = _make_player(
223 supported=[(44100, 24), (48000, 24)],
224 flow_mode=FLOW_MODE_SAMPLE_RATE_SMART,
225 )
226 streamdetails = _make_streamdetails(sample_rate=48000, bit_depth=24)
227
228 with patch.object(
229 audio,
230 "_resolve_player_dsp_config",
231 side_effect=(MagicMock(enabled=False), MagicMock(enabled=True)),
232 ):
233 fmt = await audio.select_flow_pcm_format(
234 leader,
235 start_streamdetails=streamdetails,
236 output_players=(leader, member),
237 )
238
239 assert fmt.content_type == ContentType.PCM_F32LE
240 assert fmt.bit_depth == 32
241
242
243@pytest.mark.asyncio
244async def test_select_flow_pcm_format_bit_perfect_matches_track() -> None:
245 """'bit_perfect' mode also anchors on the first track's sample rate."""
246 audio = _make_streams_audio()
247 player = _make_player(
248 supported=[(44100, 16), (48000, 24), (96000, 24)],
249 flow_mode=FLOW_MODE_SAMPLE_RATE_BIT_PERFECT,
250 )
251 streamdetails = _make_streamdetails(sample_rate=96000, bit_depth=24)
252 fmt = await audio.select_flow_pcm_format(player, start_streamdetails=streamdetails)
253 assert fmt.sample_rate == 96000
254
255
256# --- bit-depth optimization ---
257
258
259@pytest.mark.asyncio
260async def test_select_flow_pcm_format_uses_source_bit_depth_without_processing() -> None:
261 """When no processing is active, the source's bit depth is reused (no F32 upcast)."""
262 audio = _make_streams_audio()
263 player = _make_player(
264 supported=[(44100, 16), (48000, 24)],
265 flow_mode=FLOW_MODE_SAMPLE_RATE_SMART,
266 )
267 streamdetails = _make_streamdetails(sample_rate=44100, bit_depth=16)
268 fmt = await audio.select_flow_pcm_format(player, start_streamdetails=streamdetails)
269 assert fmt.bit_depth == 16
270
271
272@pytest.mark.asyncio
273async def test_select_flow_pcm_format_uses_f32_when_crossfade_enabled() -> None:
274 """Smartfades requires F32 headroom; bit depth must be 32 regardless of source."""
275 audio = _make_streams_audio()
276 player = _make_player(
277 supported=[(44100, 16), (48000, 24)],
278 flow_mode=FLOW_MODE_SAMPLE_RATE_SMART,
279 )
280 streamdetails = _make_streamdetails(sample_rate=44100, bit_depth=16)
281 fmt = await audio.select_flow_pcm_format(
282 player, start_streamdetails=streamdetails, crossfade_enabled=True
283 )
284 assert fmt.bit_depth == 32
285
286
287@pytest.mark.asyncio
288async def test_select_flow_pcm_format_uses_f32_when_overlay_active() -> None:
289 """An active audio overlay requires F32 headroom for clipping-free mixing."""
290 audio = _make_streams_audio()
291 player = _make_player(
292 supported=[(44100, 16), (48000, 24)],
293 flow_mode=FLOW_MODE_SAMPLE_RATE_SMART,
294 )
295 streamdetails = _make_streamdetails(sample_rate=44100, bit_depth=16)
296 fmt = await audio.select_flow_pcm_format(
297 player, start_streamdetails=streamdetails, overlay_active=True
298 )
299 assert fmt.bit_depth == 32
300
301
302@pytest.mark.asyncio
303async def test_select_pcm_format_uses_stereo_f32_for_radio_overlay() -> None:
304 """A mixed radio overlay gets float headroom and a stereo pivot."""
305 audio = _make_streams_audio()
306 player = _make_player(
307 supported=[(44100, 16), (48000, 24)],
308 flow_mode=FLOW_MODE_SAMPLE_RATE_SMART,
309 )
310 streamdetails = _make_streamdetails(sample_rate=44100, bit_depth=16)
311 streamdetails.media_type = MediaType.RADIO
312 streamdetails.audio_format.channels = 1
313
314 fmt = await audio.select_pcm_format(
315 player,
316 streamdetails,
317 crossfade_enabled=False,
318 overlay_active=True,
319 )
320
321 assert fmt.content_type == ContentType.PCM_F32LE
322 assert fmt.bit_depth == 32
323 assert fmt.channels == 2
324
325
326@pytest.mark.asyncio
327async def test_select_pcm_format_folds_surround_source_to_stereo() -> None:
328 """A surround source is narrowed to stereo, since no output format carries more."""
329 audio = _make_streams_audio()
330 player = _make_player(
331 supported=[(44100, 16), (48000, 24)],
332 flow_mode=FLOW_MODE_SAMPLE_RATE_SMART,
333 )
334 streamdetails = _make_streamdetails(sample_rate=48000, bit_depth=24, channels=6)
335
336 fmt = await audio.select_pcm_format(player, streamdetails, crossfade_enabled=False)
337
338 assert fmt.channels == 2
339
340
341@pytest.mark.asyncio
342async def test_select_pcm_format_keeps_mono_source_mono() -> None:
343 """A mono source is not widened when no processing stage needs a stereo pivot."""
344 audio = _make_streams_audio()
345 player = _make_player(
346 supported=[(44100, 16), (48000, 24)],
347 flow_mode=FLOW_MODE_SAMPLE_RATE_SMART,
348 )
349 streamdetails = _make_streamdetails(sample_rate=48000, bit_depth=24, channels=1)
350
351 fmt = await audio.select_pcm_format(player, streamdetails, crossfade_enabled=False)
352
353 assert fmt.channels == 1
354
355
356@pytest.mark.asyncio
357async def test_select_flow_pcm_format_uses_f32_when_volume_normalization_active() -> None:
358 """Active volume normalization on the start track triggers F32 headroom."""
359 audio = _make_streams_audio()
360 player = _make_player(
361 supported=[(44100, 16), (48000, 24)],
362 flow_mode=FLOW_MODE_SAMPLE_RATE_SMART,
363 )
364 streamdetails = _make_streamdetails(
365 sample_rate=44100,
366 bit_depth=16,
367 volume_normalization_mode=VolumeNormalizationMode.MEASUREMENT_ONLY,
368 )
369 fmt = await audio.select_flow_pcm_format(player, start_streamdetails=streamdetails)
370 assert fmt.bit_depth == 32
371
372
373@pytest.mark.asyncio
374async def test_select_flow_pcm_format_uses_f32_when_no_start_streamdetails() -> None:
375 """Without streamdetails we conservatively fall back to F32."""
376 audio = _make_streams_audio()
377 player = _make_player(
378 supported=[(44100, 16), (96000, 24)],
379 flow_mode=FLOW_MODE_SAMPLE_RATE_SMART,
380 )
381 fmt = await audio.select_flow_pcm_format(player)
382 assert fmt.bit_depth == 32
383
384
385# --- select_pcm_format (AUDIO_SOURCE passthrough) ---
386
387
388@pytest.mark.asyncio
389async def test_select_pcm_format_audio_source_passthrough_when_supported() -> None:
390 """AUDIO_SOURCE keeps source rate/bit depth/channels even with smartfades requested."""
391 audio = _make_streams_audio()
392 player = _make_player(supported=[(44100, 16), (48000, 24), (96000, 24)])
393 streamdetails = _make_streamdetails(
394 sample_rate=48000, bit_depth=24, channels=1, media_type=MediaType.AUDIO_SOURCE
395 )
396 fmt = await audio.select_pcm_format(player, streamdetails, crossfade_enabled=True)
397 assert fmt.sample_rate == 48000
398 assert fmt.bit_depth == 24
399 assert fmt.channels == 1
400
401
402@pytest.mark.asyncio
403async def test_select_pcm_format_audio_source_snaps_down_unsupported_rate() -> None:
404 """AUDIO_SOURCE snaps the sample rate down when the player can't accept the source rate."""
405 audio = _make_streams_audio()
406 player = _make_player(supported=[(44100, 16), (48000, 24)])
407 streamdetails = _make_streamdetails(
408 sample_rate=88200, bit_depth=24, media_type=MediaType.AUDIO_SOURCE
409 )
410 fmt = await audio.select_pcm_format(player, streamdetails, crossfade_enabled=False)
411 assert fmt.sample_rate == 48000
412 assert fmt.bit_depth == 24
413
414
415@pytest.mark.asyncio
416async def test_select_pcm_format_audio_source_falls_back_to_min_supported() -> None:
417 """When the source rate is below every supported rate, fall back to the player's lowest."""
418 audio = _make_streams_audio()
419 player = _make_player(supported=[(44100, 16), (48000, 24)])
420 streamdetails = _make_streamdetails(
421 sample_rate=22050, bit_depth=16, media_type=MediaType.AUDIO_SOURCE
422 )
423 fmt = await audio.select_pcm_format(player, streamdetails, crossfade_enabled=False)
424 assert fmt.sample_rate == 44100
425
426
427# --- select_flow_pcm_format (AUDIO_SOURCE passthrough) ---
428
429
430@pytest.mark.asyncio
431async def test_select_flow_pcm_format_audio_source_bypasses_flow_mode() -> None:
432 """AUDIO_SOURCE as start item ignores flow mode config and passes the source through."""
433 audio = _make_streams_audio()
434 # 'highest' would normally pick 96000 â passthrough must override that.
435 player = _make_player(
436 supported=[(44100, 16), (48000, 24), (96000, 24)],
437 flow_mode=FLOW_MODE_SAMPLE_RATE_HIGHEST,
438 )
439 streamdetails = _make_streamdetails(
440 sample_rate=44100, bit_depth=16, media_type=MediaType.AUDIO_SOURCE
441 )
442 fmt = await audio.select_flow_pcm_format(
443 player, start_streamdetails=streamdetails, crossfade_enabled=True
444 )
445 assert fmt.sample_rate == 44100
446 assert fmt.bit_depth == 16
447 assert fmt.channels == 2
448
449
450@pytest.mark.asyncio
451async def test_select_flow_pcm_format_audio_source_preserves_mono() -> None:
452 """AUDIO_SOURCE keeps source channels â no forced stereo widening."""
453 audio = _make_streams_audio()
454 player = _make_player(
455 supported=[(44100, 16), (48000, 24)],
456 flow_mode=FLOW_MODE_SAMPLE_RATE_SMART,
457 )
458 streamdetails = _make_streamdetails(
459 sample_rate=44100, bit_depth=16, channels=1, media_type=MediaType.AUDIO_SOURCE
460 )
461 fmt = await audio.select_flow_pcm_format(player, start_streamdetails=streamdetails)
462 assert fmt.channels == 1
463
464
465# --- _flow_stream_needs_restart ---
466
467
468def test_needs_restart_first_track_never_breaks() -> None:
469 """The first (anchor) track is always allowed to continue."""
470 audio = _make_streams_audio()
471 track = _make_queue_track(rate=192000)
472 pcm = AudioFormat(sample_rate=48000, bit_depth=32, channels=2)
473 assert (
474 audio._flow_stream_needs_restart(
475 track,
476 pcm,
477 supported_sample_rates=[44100, 48000, 96000, 192000],
478 flow_mode_sample_rate_conf=FLOW_MODE_SAMPLE_RATE_BIT_PERFECT,
479 is_first_track=True,
480 )
481 is False
482 )
483
484
485def test_needs_restart_radio_always_breaks() -> None:
486 """Radio items always break out of flow, even on the first track."""
487 audio = _make_streams_audio()
488 track = _make_queue_track(rate=44100, media_type=MediaType.RADIO)
489 pcm = AudioFormat(sample_rate=44100, bit_depth=32, channels=2)
490 assert (
491 audio._flow_stream_needs_restart(
492 track,
493 pcm,
494 supported_sample_rates=[44100, 48000],
495 flow_mode_sample_rate_conf=FLOW_MODE_SAMPLE_RATE_SMART,
496 is_first_track=True,
497 )
498 is True
499 )
500
501
502def test_needs_restart_audio_source_breaks() -> None:
503 """Live AUDIO_SOURCE items break out so the controller can use single-item streaming."""
504 audio = _make_streams_audio()
505 track = _make_queue_track(rate=44100, media_type=MediaType.AUDIO_SOURCE)
506 pcm = AudioFormat(sample_rate=44100, bit_depth=32, channels=2)
507 assert (
508 audio._flow_stream_needs_restart(
509 track,
510 pcm,
511 supported_sample_rates=[44100, 48000],
512 flow_mode_sample_rate_conf=FLOW_MODE_SAMPLE_RATE_SMART,
513 is_first_track=False,
514 )
515 is True
516 )
517
518
519def test_needs_restart_smart_breaks_when_rate_increases() -> None:
520 """Smart mode breaks when the next track's effective rate is higher than the flow's."""
521 audio = _make_streams_audio()
522 track = _make_queue_track(rate=96000)
523 pcm = AudioFormat(sample_rate=48000, bit_depth=32, channels=2)
524 assert (
525 audio._flow_stream_needs_restart(
526 track,
527 pcm,
528 supported_sample_rates=[44100, 48000, 96000],
529 flow_mode_sample_rate_conf=FLOW_MODE_SAMPLE_RATE_SMART,
530 is_first_track=False,
531 )
532 is True
533 )
534
535
536def test_needs_restart_smart_no_break_on_lower_rate() -> None:
537 """Smart mode does not break when the next track's rate is equal or lower."""
538 audio = _make_streams_audio()
539 track = _make_queue_track(rate=44100)
540 pcm = AudioFormat(sample_rate=48000, bit_depth=32, channels=2)
541 assert (
542 audio._flow_stream_needs_restart(
543 track,
544 pcm,
545 supported_sample_rates=[44100, 48000, 96000],
546 flow_mode_sample_rate_conf=FLOW_MODE_SAMPLE_RATE_SMART,
547 is_first_track=False,
548 )
549 is False
550 )
551
552
553def test_needs_restart_bit_perfect_breaks_on_any_rate_change() -> None:
554 """Bit-perfect mode breaks on any rate change (including downsampling)."""
555 audio = _make_streams_audio()
556 track = _make_queue_track(rate=44100)
557 pcm = AudioFormat(sample_rate=48000, bit_depth=32, channels=2)
558 assert (
559 audio._flow_stream_needs_restart(
560 track,
561 pcm,
562 supported_sample_rates=[44100, 48000, 96000],
563 flow_mode_sample_rate_conf=FLOW_MODE_SAMPLE_RATE_BIT_PERFECT,
564 is_first_track=False,
565 )
566 is True
567 )
568
569
570def test_needs_restart_snaps_up_unsupported_next_rate_before_compare() -> None:
571 """An unsupported raw rate is snapped up before being compared to the flow rate."""
572 # raw 44100 isn't supported; the player snaps it up to 48000 â which equals the
573 # current flow rate, so no restart is needed even under bit-perfect.
574 audio = _make_streams_audio()
575 track = _make_queue_track(rate=44100)
576 pcm = AudioFormat(sample_rate=48000, bit_depth=32, channels=2)
577 assert (
578 audio._flow_stream_needs_restart(
579 track,
580 pcm,
581 supported_sample_rates=[48000, 96000],
582 flow_mode_sample_rate_conf=FLOW_MODE_SAMPLE_RATE_BIT_PERFECT,
583 is_first_track=False,
584 )
585 is False
586 )
587
588
589def test_needs_restart_fixed_rate_modes_never_break_on_rate() -> None:
590 """The fixed-rate modes always resample to the flow rate, so no restart."""
591 audio = _make_streams_audio()
592 track = _make_queue_track(rate=96000)
593 pcm = AudioFormat(sample_rate=48000, bit_depth=32, channels=2)
594 for mode in (
595 FLOW_MODE_SAMPLE_RATE_48000,
596 FLOW_MODE_SAMPLE_RATE_96000,
597 FLOW_MODE_SAMPLE_RATE_HIGHEST,
598 ):
599 assert (
600 audio._flow_stream_needs_restart(
601 track,
602 pcm,
603 supported_sample_rates=[44100, 48000, 96000, 192000],
604 flow_mode_sample_rate_conf=mode,
605 is_first_track=False,
606 )
607 is False
608 )
609
610
611def test_needs_restart_returns_false_when_streamdetails_missing() -> None:
612 """Without streamdetails the sample rate can't be evaluated; do not break the flow."""
613 audio = _make_streams_audio()
614 track = _make_queue_track(rate=44100)
615 track.streamdetails = None
616 pcm = AudioFormat(sample_rate=48000, bit_depth=32, channels=2)
617 assert (
618 audio._flow_stream_needs_restart(
619 track,
620 pcm,
621 supported_sample_rates=[44100, 48000],
622 flow_mode_sample_rate_conf=FLOW_MODE_SAMPLE_RATE_BIT_PERFECT,
623 is_first_track=False,
624 )
625 is False
626 )
627
628
629# --- _flow_restart_context ---
630
631
632def test_flow_restart_context_prefers_protocol_player() -> None:
633 """The given (protocol) player's config and rates win over the queue player's."""
634 audio = _make_streams_audio()
635 protocol_player = _make_player(
636 supported=[(44100, 16), (96000, 24)],
637 flow_mode=FLOW_MODE_SAMPLE_RATE_BIT_PERFECT,
638 )
639 # queue (wrapper) player without audio config entries: 44100-only fallback
640 wrapper_player = _make_player(supported=[(44100, 16)])
641 audio.mass.players.get_player = MagicMock( # type: ignore[method-assign]
642 return_value=wrapper_player
643 )
644
645 conf, rates = audio._flow_restart_context("queue-1", protocol_player)
646
647 assert conf == FLOW_MODE_SAMPLE_RATE_BIT_PERFECT
648 assert rates == [44100, 96000]
649 audio.mass.players.get_player.assert_not_called()
650
651
652def test_flow_restart_context_falls_back_to_queue_player() -> None:
653 """Without a protocol player, the queue's own player is used."""
654 audio = _make_streams_audio()
655 queue_player = _make_player(
656 supported=[(48000, 16)], flow_mode=FLOW_MODE_SAMPLE_RATE_BIT_PERFECT
657 )
658 audio.mass.players.get_player = MagicMock( # type: ignore[method-assign]
659 return_value=queue_player
660 )
661
662 conf, rates = audio._flow_restart_context("queue-1", None)
663
664 assert conf == FLOW_MODE_SAMPLE_RATE_BIT_PERFECT
665 assert rates == [48000]
666 audio.mass.players.get_player.assert_called_once_with("queue-1")
667
668
669def test_flow_restart_context_without_any_player() -> None:
670 """When no player can be resolved, the raw queue config and empty rates are used."""
671 audio = _make_streams_audio()
672 audio.mass.players.get_player = MagicMock(return_value=None) # type: ignore[method-assign]
673 audio.mass.config.get_raw_player_config_value = MagicMock( # type: ignore[method-assign]
674 return_value=FLOW_MODE_SAMPLE_RATE_SMART
675 )
676
677 conf, rates = audio._flow_restart_context("queue-1", None)
678
679 assert conf == FLOW_MODE_SAMPLE_RATE_SMART
680 assert rates == []
681
682
683# --- helpers ---
684
685
686def _make_streams_audio() -> StreamsAudio:
687 """Return a StreamsAudio with everything mocked out except the logger."""
688 audio = StreamsAudio(MagicMock())
689 # DSP must be disabled in tests so the bit-depth optimization can engage;
690 # individual tests opt back in via smartfades or volume normalization.
691 audio.mass.config.get_player_dsp_config = MagicMock( # type: ignore[method-assign]
692 return_value=MagicMock(enabled=False)
693 )
694 return audio
695
696
697def _make_player(
698 *,
699 supported: list[tuple[int, int]],
700 flow_mode: str = FLOW_MODE_SAMPLE_RATE_SMART,
701) -> MagicMock:
702 """Build a player double exposing only what select_flow_pcm_format needs."""
703 player = MagicMock()
704 player.get_supported_sample_rates = MagicMock(return_value=supported)
705 player.config.get_value = MagicMock(
706 side_effect=lambda key, default=None: (
707 flow_mode if key == CONF_FLOW_MODE_SAMPLE_RATE else default
708 )
709 )
710 return player
711
712
713def _make_streamdetails(
714 *,
715 sample_rate: int,
716 bit_depth: int,
717 channels: int = 2,
718 volume_normalization_mode: VolumeNormalizationMode = VolumeNormalizationMode.DISABLED,
719 media_type: MediaType = MediaType.TRACK,
720) -> MagicMock:
721 """Build a StreamDetails double carrying just the fields the format selector reads."""
722 streamdetails = MagicMock()
723 streamdetails.audio_format = AudioFormat(
724 sample_rate=sample_rate, bit_depth=bit_depth, channels=channels
725 )
726 # a real StreamDetails leaves this unset unless the provider handed over
727 # already-decoded audio, and a MagicMock attribute would read as one
728 streamdetails.decoded_audio_format = None
729 streamdetails.volume_normalization_mode = volume_normalization_mode
730 streamdetails.media_type = media_type
731 return streamdetails
732
733
734def _make_queue_track(*, rate: int, media_type: MediaType = MediaType.TRACK) -> MagicMock:
735 """Build a queue_item double with the minimum fields the helper inspects."""
736 track = MagicMock()
737 track.queue_item_id = "item-1"
738 track.name = "Test Track"
739 track.media_type = media_type
740 track.streamdetails = MagicMock()
741 track.streamdetails.audio_format = AudioFormat(sample_rate=rate, bit_depth=16, channels=2)
742 return track
743