music-assistant-server

1"""Tests for effective audio processing plans and stream details."""
2
3from __future__ import annotations
4
5from collections.abc import AsyncGenerator
6from copy import deepcopy
7from types import SimpleNamespace
8from typing import Any, cast
9from unittest.mock import AsyncMock, MagicMock
10
11import pytest
12from music_assistant_models.audio_processing import (
13    AudioDSPDetails,
14    AudioFidelity,
15    AudioNormalizationDetails,
16    AudioOutputDetails,
17    AudioProcessingChain,
18    AudioQuality,
19    AudioQueueProcessing,
20)
21from music_assistant_models.dsp import (
22    AudioChannel,
23    ConvolutionFilter,
24    DSPConfig,
25    DSPState,
26    ToneControlFilter,
27)
28from music_assistant_models.enums import (
29    ContentType,
30    CrossfadeMode,
31    MediaType,
32    VolumeNormalizationMode,
33)
34from music_assistant_models.errors import QueueEmpty
35from music_assistant_models.media_items import AudioFormat
36from music_assistant_models.streamdetails import StreamDetails
37
38from music_assistant.controllers.streams.audio import StreamsAudio
39from music_assistant.controllers.streams.audio_processing import (
40    AudioOutputPlan,
41    AudioProcessingManager,
42    get_audio_quality,
43    get_normalization_details,
44)
45from music_assistant.controllers.streams.controller import StreamsController
46from music_assistant.helpers.dsp import ComplexFilter
47
48
49def _format(
50    content_type: ContentType,
51    sample_rate: int = 44100,
52    bit_depth: int = 16,
53    *,
54    channels: int = 2,
55    bit_rate: int | None = None,
56) -> AudioFormat:
57    """Return an AudioFormat with matching container and codec."""
58    return AudioFormat(
59        content_type=content_type,
60        codec_type=content_type,
61        sample_rate=sample_rate,
62        bit_depth=bit_depth,
63        channels=channels,
64        bit_rate=bit_rate,
65    )
66
67
68@pytest.mark.parametrize(
69    ("audio_format", "expected"),
70    [
71        (_format(ContentType.FLAC, 44100, 16), AudioQuality.LOSSLESS),
72        (_format(ContentType.FLAC, 96000, 24), AudioQuality.HI_RES),
73        (_format(ContentType.MP3, bit_rate=320), AudioQuality.STANDARD),
74        (_format(ContentType.AAC, bit_rate=128), AudioQuality.LOW),
75        (_format(ContentType.MP3, bit_rate=128000), AudioQuality.LOW),
76        (_format(ContentType.AAC), AudioQuality.UNKNOWN),
77    ],
78)
79def test_get_audio_quality(audio_format: AudioFormat, expected: AudioQuality) -> None:
80    """Quality classification uses codec, resolution and normalized bitrate."""
81    assert get_audio_quality(audio_format) == expected
82
83
84def test_get_normalization_details_uses_album_measurement() -> None:
85    """Album normalization reports the selected measurement and applied gain."""
86    streamdetails = _streamdetails()
87    streamdetails.volume_normalization_mode = VolumeNormalizationMode.MEASUREMENT_ONLY
88    streamdetails.prefer_album_loudness = True
89    streamdetails.loudness = -12.0
90    streamdetails.loudness_album = -14.5
91    streamdetails.target_loudness = -17.0
92
93    details = get_normalization_details(streamdetails, applied_gain_db=-2.5)
94
95    assert details is not None
96    assert details.measurement_source.value == "album"
97    assert details.measured_lufs == -14.5
98    assert details.target_lufs == -17.0
99    assert details.applied_gain_db == -2.5
100
101
102def test_audio_processing_manager_attaches_grouped_chain() -> None:
103    """A complete chain is attached to StreamDetails with grouped outputs."""
104    manager, _mass, _queue_data, streamdetails, lossless_plan, lossy_plan = _manager_context()
105    assert streamdetails.audio_processing is None
106
107    assert manager.update_output(
108        "player-2",
109        lossless_plan,
110        queue_id="queue-1",
111        session_id="session-1",
112        queue_item_id="item-1",
113    )
114    assert manager.update_output(
115        "player-1",
116        lossless_plan,
117        queue_id="queue-1",
118        session_id="session-1",
119        queue_item_id="item-1",
120    )
121    assert manager.update_output(
122        "player-3",
123        lossy_plan,
124        queue_id="queue-1",
125        session_id="session-1",
126        queue_item_id="item-1",
127    )
128
129    chain = cast("AudioProcessingChain", streamdetails.audio_processing)
130    assert chain.input_fidelity.quality == AudioQuality.HI_RES
131    assert chain.queue_processing is not None
132    assert chain.outputs[0].player_ids == ["player-1", "player-2"]
133    assert chain.outputs[0].fidelity == AudioFidelity(
134        quality=AudioQuality.HI_RES,
135        bit_perfect=True,
136    )
137    assert chain.outputs[1].player_ids == ["player-3"]
138    assert chain.outputs[1].fidelity == AudioFidelity(
139        quality=AudioQuality.LOW,
140        bit_perfect=False,
141    )
142
143
144def test_lossy_source_can_have_bit_perfect_lossless_output() -> None:
145    """Lossy source quality does not prevent preserving its decoded PCM samples."""
146    manager, _mass, _queue_data, streamdetails, lossless_plan, lossy_plan = _manager_context()
147    streamdetails.audio_format = AudioFormat(
148        content_type=ContentType.OGG,
149        codec_type=ContentType.VORBIS,
150        sample_rate=44100,
151        bit_depth=16,
152        channels=2,
153        bit_rate=320,
154    )
155    pcm_format = _format(ContentType.PCM_S16LE)
156    manager.update_item_runtime(
157        "queue-1",
158        "session-1",
159        "item-1",
160        input_format=pcm_format,
161        pcm_format=pcm_format,
162        normalization=None,
163        playback_speed=1.0,
164    )
165    lossless_plan.input_format = pcm_format
166    lossless_plan.output_details.output_format = _format(ContentType.FLAC)
167    lossy_plan.input_format = pcm_format
168    lossy_plan.output_details.output_format = _format(ContentType.MP3, bit_rate=320)
169
170    manager.update_output(
171        "lossless-player",
172        lossless_plan,
173        queue_id="queue-1",
174        session_id="session-1",
175        queue_item_id="item-1",
176    )
177    manager.update_output(
178        "lossy-player",
179        lossy_plan,
180        queue_id="queue-1",
181        session_id="session-1",
182        queue_item_id="item-1",
183    )
184
185    chain = cast("AudioProcessingChain", streamdetails.audio_processing)
186    outputs = {output.player_ids[0]: output for output in chain.outputs}
187    assert chain.input_fidelity.quality == AudioQuality.STANDARD
188    assert outputs["lossless-player"].fidelity == AudioFidelity(
189        quality=AudioQuality.STANDARD,
190        bit_perfect=True,
191    )
192    assert outputs["lossy-player"].fidelity == AudioFidelity(
193        quality=AudioQuality.STANDARD,
194        bit_perfect=False,
195    )
196    serialized = streamdetails.to_dict()
197    serialized_outputs = {
198        output["player_ids"][0]: output for output in serialized["audio_processing"]["outputs"]
199    }
200    assert serialized_outputs["lossless-player"]["fidelity"]["bit_perfect"] is True
201
202
203def test_a_wider_provider_handoff_preserves_the_source_samples() -> None:
204    """A provider that decodes upstream into wider PCM does not lose the source samples."""
205    streamdetails = _source_handled_soloist_item(_format(ContentType.FLAC, 44100, 24))
206
207    chain = cast("AudioProcessingChain", streamdetails.audio_processing)
208    assert chain.input_fidelity.quality == AudioQuality.HI_RES
209    assert chain.outputs[0].fidelity.bit_perfect is True
210
211
212def test_an_output_narrower_than_the_source_is_not_bit_perfect() -> None:
213    """Dropping a 24-bit source to a 16-bit output loses bits, wide handoff or not."""
214    streamdetails = _source_handled_soloist_item(_format(ContentType.FLAC, 44100, 16))
215
216    chain = cast("AudioProcessingChain", streamdetails.audio_processing)
217    assert chain.outputs[0].fidelity.bit_perfect is False
218
219
220def test_an_internal_stage_narrower_than_the_source_is_not_bit_perfect() -> None:
221    """A narrowed internal stage loses bits the output cannot bring back."""
222    manager, _mass, _queue_data, streamdetails, lossless_plan, _lossy_plan = _manager_context()
223    streamdetails.audio_format = _format(ContentType.FLAC, 44100, 24)
224    narrowed = _format(ContentType.PCM_S16LE, 44100, 16)
225    manager.update_item_runtime(
226        "queue-1",
227        "session-1",
228        "item-1",
229        input_format=narrowed,
230        pcm_format=narrowed,
231        normalization=None,
232        playback_speed=1.0,
233    )
234    lossless_plan.input_format = narrowed
235    lossless_plan.output_details.output_format = _format(ContentType.FLAC, 44100, 24)
236    manager.update_output(
237        "player-1",
238        lossless_plan,
239        queue_id="queue-1",
240        session_id="session-1",
241        queue_item_id="item-1",
242    )
243
244    chain = cast("AudioProcessingChain", streamdetails.audio_processing)
245    assert chain.outputs[0].fidelity.bit_perfect is False
246
247
248def test_float_headroom_alone_does_not_break_the_bit_perfect_claim() -> None:
249    """DSP enabled with no filters gets F32 headroom but leaves the samples alone."""
250    manager, _mass, _queue_data, streamdetails, lossless_plan, _lossy_plan = _manager_context()
251    streamdetails.audio_format = _format(ContentType.FLAC, 44100, 24)
252    headroom = _format(ContentType.PCM_F32LE, 44100, 32)
253    manager.update_item_runtime(
254        "queue-1",
255        "session-1",
256        "item-1",
257        input_format=headroom,
258        pcm_format=headroom,
259        normalization=None,
260        playback_speed=1.0,
261    )
262    lossless_plan.input_format = headroom
263    lossless_plan.output_details.dsp = AudioDSPDetails(state=DSPState.ENABLED)
264    lossless_plan.output_details.output_format = _format(ContentType.FLAC, 44100, 24)
265    manager.update_output(
266        "player-1",
267        lossless_plan,
268        queue_id="queue-1",
269        session_id="session-1",
270        queue_item_id="item-1",
271    )
272
273    chain = cast("AudioProcessingChain", streamdetails.audio_processing)
274    assert chain.outputs[0].fidelity.bit_perfect is True
275
276
277def test_shared_output_destinations_are_registered_atomically() -> None:
278    """One shared output publishes all destinations in a single queue update."""
279    manager, mass, _queue_data, streamdetails, output_plan, _lossy_plan = _manager_context()
280    mass.player_queues.signal_update.reset_mock()
281
282    assert manager.update_output(
283        "leader",
284        output_plan,
285        shared_player_ids={"leader", "sync-child"},
286        queue_id="queue-1",
287        session_id="session-1",
288        queue_item_id="item-1",
289    )
290
291    assert streamdetails.audio_processing is not None
292    assert len(streamdetails.audio_processing.outputs) == 1
293    assert streamdetails.audio_processing.outputs[0].player_ids == ["leader", "sync-child"]
294    mass.player_queues.signal_update.assert_called_once_with("queue-1")
295
296    mass.player_queues.signal_update.reset_mock()
297    assert not manager.update_output(
298        "leader",
299        output_plan,
300        shared_player_ids={"sync-child"},
301        queue_id="queue-1",
302        session_id="session-1",
303        queue_item_id="item-1",
304    )
305    mass.player_queues.signal_update.assert_not_called()
306
307
308def test_shared_output_adds_member_without_stream_restart() -> None:
309    """A late native-sync member inherits the active shared output path."""
310    manager, mass, _queue_data, streamdetails, output_plan, _lossy_plan = _manager_context()
311    manager.update_output(
312        "leader",
313        output_plan,
314        shared_player_ids=(),
315        queue_id="queue-1",
316        session_id="session-1",
317        queue_item_id="item-1",
318    )
319    mass.player_queues.signal_update.reset_mock()
320
321    assert manager.retain_outputs("queue-1", {"queue-1", "leader", "late-member"})
322
323    assert streamdetails.audio_processing is not None
324    assert streamdetails.audio_processing.outputs[0].player_ids == ["late-member", "leader"]
325    mass.player_queues.signal_update.assert_called_once_with("queue-1")
326
327
328def test_independent_output_does_not_add_group_member() -> None:
329    """Membership changes do not expand an independently processed output."""
330    manager, mass, _queue_data, streamdetails, output_plan, _lossy_plan = _manager_context()
331    manager.update_output(
332        "leader",
333        output_plan,
334        queue_id="queue-1",
335        session_id="session-1",
336        queue_item_id="item-1",
337    )
338    mass.player_queues.signal_update.reset_mock()
339
340    assert not manager.retain_outputs("queue-1", {"leader", "independent-member"})
341
342    assert streamdetails.audio_processing is not None
343    assert streamdetails.audio_processing.outputs[0].player_ids == ["leader"]
344    mass.player_queues.signal_update.assert_not_called()
345
346
347def test_preset_identity_update_republishes_current_chain() -> None:
348    """Preset updates follow a changed config owner even when output details match."""
349    manager, mass, _queue_data, streamdetails, output_plan, _lossy_plan = _manager_context()
350    output_plan.dsp_config_id = "old-config-player"
351    output_plan.output_details.dsp.preset_id = "night"
352    manager.update_output(
353        "player-1",
354        output_plan,
355        queue_id="queue-1",
356        session_id="session-1",
357        queue_item_id="item-1",
358    )
359    replacement = deepcopy(output_plan)
360    replacement.dsp_config_id = "configured-player"
361    assert manager.update_output(
362        "player-1",
363        replacement,
364        queue_id="queue-1",
365        session_id="session-1",
366        queue_item_id="item-1",
367    )
368    mass.player_queues.signal_update.reset_mock()
369
370    manager.update_player_dsp_preset("configured-player", None)
371
372    assert streamdetails.audio_processing is not None
373    assert streamdetails.audio_processing.outputs[0].dsp.preset_id is None
374    mass.player_queues.signal_update.assert_called_once_with("queue-1")
375
376
377def test_retain_outputs_signals_current_chain_change() -> None:
378    """Pruning a current output publishes the reduced chain."""
379    manager, mass, _queue_data, streamdetails, output_plan, _lossy_plan = _manager_context()
380    manager.update_output(
381        "player-1",
382        output_plan,
383        shared_player_ids={"player-2"},
384        queue_id="queue-1",
385        session_id="session-1",
386        queue_item_id="item-1",
387    )
388    mass.player_queues.signal_update.reset_mock()
389
390    assert manager.retain_outputs("queue-1", {"player-1"})
391
392    assert streamdetails.audio_processing is not None
393    assert streamdetails.audio_processing.outputs[0].player_ids == ["player-1"]
394    mass.player_queues.signal_update.assert_called_once_with("queue-1")
395
396
397def test_prefetched_output_does_not_replace_current_chain() -> None:
398    """An output prepared for the next item does not change the current item."""
399    manager, mass, queue_data, streamdetails, lossless_plan, lossy_plan = _manager_context()
400    next_streamdetails = _streamdetails(item_id="item-2")
401    next_item = SimpleNamespace(queue_item_id="item-2", streamdetails=next_streamdetails)
402    queue_data.items.append(next_item)
403    mass.player_queues.get_item.side_effect = lambda _queue_id, item_id: (
404        next_item if item_id == "item-2" else queue_data.items[0]
405    )
406    manager.update_output(
407        "player-1",
408        lossless_plan,
409        queue_id="queue-1",
410        session_id="session-1",
411        queue_item_id="item-1",
412    )
413    current_chain = streamdetails.audio_processing
414
415    manager.update_item_context(
416        "queue-1",
417        "session-1",
418        "item-2",
419        AudioQueueProcessing(pcm_format=lossless_plan.input_format),
420    )
421    manager.update_output(
422        "player-1",
423        lossy_plan,
424        queue_id="queue-1",
425        session_id="session-1",
426        queue_item_id="item-2",
427    )
428
429    assert streamdetails.audio_processing == current_chain
430    assert next_streamdetails.audio_processing is not None
431    assert next_streamdetails.audio_processing.outputs[0].fidelity.quality == AudioQuality.LOW
432
433
434def test_context_refresh_preserves_runtime_normalization() -> None:
435    """A second consumer does not erase normalization resolved at stream time."""
436    manager, _mass, _queue_data, streamdetails, lossless_plan, _lossy_plan = _manager_context()
437    normalization = AudioNormalizationDetails(
438        mode=VolumeNormalizationMode.DYNAMIC,
439        target_lufs=-17.0,
440    )
441    manager.update_item_runtime(
442        "queue-1",
443        "session-1",
444        "item-1",
445        input_format=lossless_plan.input_format,
446        pcm_format=lossless_plan.input_format,
447        normalization=normalization,
448        playback_speed=1.0,
449    )
450    manager.update_output(
451        "player-1",
452        lossless_plan,
453        queue_id="queue-1",
454        session_id="session-1",
455        queue_item_id="item-1",
456    )
457
458    manager.update_item_context(
459        "queue-1",
460        "session-1",
461        "item-1",
462        AudioQueueProcessing(
463            pcm_format=lossless_plan.input_format,
464            crossfade_mode=CrossfadeMode.SMART_CROSSFADE,
465        ),
466    )
467
468    chain = cast("AudioProcessingChain", streamdetails.audio_processing)
469    assert chain.queue_processing is not None
470    assert chain.queue_processing.normalization == normalization
471    assert chain.outputs[0].fidelity.bit_perfect is False
472
473
474def test_manager_rejects_superseded_and_cleared_sessions() -> None:
475    """Late producers cannot update or recreate a replacement queue session."""
476    manager, mass, queue_data, streamdetails, lossless_plan, _lossy_plan = _manager_context()
477    manager.update_output(
478        "player-1",
479        lossless_plan,
480        queue_id="queue-1",
481        session_id="session-1",
482        queue_item_id="item-1",
483    )
484    assert streamdetails.to_dict()["audio_processing"] is not None
485
486    mass.player_queues.signal_update.reset_mock()
487    manager.start_session("queue-1", "session-2")
488    queue_data.session_id = "session-2"
489    assert streamdetails.to_dict()["audio_processing"] is None
490    mass.player_queues.signal_update.assert_called_once_with("queue-1")
491    assert not manager.update_output(
492        "stale-player",
493        lossless_plan,
494        shared_player_ids={"stale-child"},
495        queue_id="queue-1",
496        session_id="session-1",
497    )
498    assert streamdetails.audio_processing is None
499
500    manager.update_item_context(
501        "queue-1",
502        "session-2",
503        "item-1",
504        AudioQueueProcessing(pcm_format=lossless_plan.input_format),
505    )
506    manager.update_output(
507        "current-player",
508        lossless_plan,
509        queue_id="queue-1",
510        session_id="session-2",
511        queue_item_id="item-1",
512    )
513    assert streamdetails.to_dict()["audio_processing"] is not None
514    mass.player_queues.signal_update.reset_mock()
515    manager.clear("queue-1", "session-2")
516    assert streamdetails.to_dict()["audio_processing"] is None
517    mass.player_queues.signal_update.assert_called_once_with("queue-1")
518    assert not manager.update_output(
519        "late-player",
520        lossless_plan,
521        queue_id="queue-1",
522        session_id="session-2",
523    )
524
525
526def test_manager_prunes_played_item_chains() -> None:
527    """Advancing the queue drops processing state from completed items."""
528    manager, mass, queue_data, streamdetails, lossless_plan, _lossy_plan = _manager_context()
529    manager.update_output(
530        "player-1",
531        lossless_plan,
532        queue_id="queue-1",
533        session_id="session-1",
534        queue_item_id="item-1",
535    )
536    assert streamdetails.audio_processing is not None
537    next_streamdetails = _streamdetails(item_id="item-2")
538    next_item = SimpleNamespace(queue_item_id="item-2", streamdetails=next_streamdetails)
539    queue_data.items.append(next_item)
540    queue_data.queue.current_index = 1
541    queue_data.queue.current_item = next_item
542    mass.player_queues.get_item.side_effect = lambda _queue_id, item_id: (
543        next_item if item_id == "item-2" else queue_data.items[0]
544    )
545
546    manager.update_item_context(
547        "queue-1",
548        "session-1",
549        "item-2",
550        AudioQueueProcessing(pcm_format=lossless_plan.input_format),
551    )
552    manager.update_item_runtime(
553        "queue-1",
554        "session-1",
555        "item-1",
556        input_format=lossless_plan.input_format,
557        pcm_format=lossless_plan.input_format,
558        normalization=None,
559        playback_speed=1.0,
560    )
561
562    assert streamdetails.audio_processing is None
563
564
565def test_hidden_and_intermediate_processing_prevents_bit_perfect_claim() -> None:
566    """Hidden fades and lower-resolution handoffs prevent bit-perfect output."""
567    manager, _mass, _queue_data, streamdetails, output_plan, _lossy_plan = _manager_context(
568        alters_audio=True
569    )
570    output_plan.handoff_format = _format(ContentType.PCM_S24LE, 48000, 24)
571
572    manager.update_output(
573        "player-1",
574        output_plan,
575        queue_id="queue-1",
576        session_id="session-1",
577        queue_item_id="item-1",
578    )
579
580    assert streamdetails.audio_processing is not None
581    assert streamdetails.audio_processing.outputs[0].fidelity.bit_perfect is False
582
583
584def test_player_output_plan_matches_ffmpeg_filters() -> None:
585    """Typed output details describe the FFmpeg filters returned to callers."""
586    mass = MagicMock()
587    mass.players.get_player.return_value = None
588    mass.config.get_player_dsp_config.return_value = DSPConfig(
589        enabled=True,
590        input_gain=-1.0,
591        filters=[ToneControlFilter(enabled=True, bass_level=2.0)],
592        output_gain=-0.5,
593        preset_id="night",
594    )
595    mass.config.get_raw_player_config_value.return_value = "left"
596    audio = StreamsAudio(cast("Any", mass))
597    input_format = _format(ContentType.PCM_F32LE, 96000, 32)
598    output_format = _format(ContentType.FLAC, 48000, 16, channels=1)
599
600    plan = audio.get_player_output_plan(
601        "player-1",
602        input_format,
603        output_format,
604        queue_id="queue-1",
605        session_id="session-1",
606        queue_item_id="item-1",
607    )
608
609    assert plan.filter_params[0] == "volume=-1.0dB"
610    assert plan.filter_params[-1] == "pan=mono|c0=FL"
611    assert plan.output_details.dsp == AudioDSPDetails(
612        state=DSPState.ENABLED,
613        input_gain=-1.0,
614        filters=[ToneControlFilter(enabled=True, bass_level=2.0)],
615        output_gain=-0.5,
616        preset_id="night",
617    )
618    assert plan.dsp_config_id == "player-1"
619    assert plan.output_details.source_channel == AudioChannel.FL
620    assert plan.output_details.output_format == output_format
621    mass.streams.audio_processing.update_output.assert_called_once_with(
622        "player-1",
623        plan,
624        shared_player_ids=None,
625        queue_id="queue-1",
626        session_id="session-1",
627        queue_item_id="item-1",
628    )
629
630
631def test_player_output_plan_downmixes_to_mono() -> None:
632    """The mono output mode folds both source channels into a single channel."""
633    mass = MagicMock()
634    mass.players.get_player.return_value = None
635    mass.config.get_player_dsp_config.return_value = DSPConfig(enabled=False)
636    mass.config.get_raw_player_config_value.return_value = "mono"
637    audio = StreamsAudio(cast("Any", mass))
638    input_format = _format(ContentType.PCM_F32LE, 48000, 32)
639    output_format = _format(ContentType.FLAC, 48000, 16, channels=1)
640
641    plan = audio.get_player_output_plan(
642        "player-1",
643        input_format,
644        output_format,
645        queue_id="queue-1",
646        session_id="session-1",
647        queue_item_id="item-1",
648    )
649
650    assert plan.filter_params == ["pan=mono|c0=0.5*FL+0.5*FR"]
651    assert plan.output_details.source_channel == AudioChannel.ALL
652
653
654def test_player_output_plan_feeds_every_output_channel() -> None:
655    """A stereo output carries the downmix on both channels instead of being upmixed."""
656    mass = MagicMock()
657    mass.players.get_player.return_value = None
658    mass.config.get_player_dsp_config.return_value = DSPConfig(enabled=False)
659    mass.config.get_raw_player_config_value.return_value = "mono"
660    audio = StreamsAudio(cast("Any", mass))
661    audio_format = _format(ContentType.PCM_F32LE, 48000, 32)
662
663    plan = audio.get_player_output_plan(
664        "player-1",
665        audio_format,
666        audio_format,
667        queue_id="queue-1",
668        session_id="session-1",
669        queue_item_id="item-1",
670    )
671
672    assert plan.filter_params == ["pan=stereo|c0=0.5*FL+0.5*FR|c1=0.5*FL+0.5*FR"]
673    assert plan.output_details.source_channel == AudioChannel.ALL
674
675
676def test_player_output_plan_skips_channel_selection_for_mono_source() -> None:
677    """A single channel source has no channels to select, so it is left untouched."""
678    mass = MagicMock()
679    mass.players.get_player.return_value = None
680    mass.config.get_player_dsp_config.return_value = DSPConfig(enabled=False)
681    mass.config.get_raw_player_config_value.return_value = "mono"
682    audio = StreamsAudio(cast("Any", mass))
683    audio_format = _format(ContentType.PCM_F32LE, 48000, 32, channels=1)
684
685    plan = audio.get_player_output_plan(
686        "player-1",
687        audio_format,
688        audio_format,
689        queue_id="queue-1",
690        session_id="session-1",
691        queue_item_id="item-1",
692    )
693
694    assert plan.filter_params == []
695    assert plan.output_details.source_channel is None
696
697
698def test_player_output_plan_pans_for_the_handoff_format() -> None:
699    """The pan follows the format FFmpeg emits, not a later provider side encode."""
700    mass = MagicMock()
701    mass.players.get_player.return_value = None
702    mass.config.get_player_dsp_config.return_value = DSPConfig(enabled=False)
703    mass.config.get_raw_player_config_value.return_value = "mono"
704    audio = StreamsAudio(cast("Any", mass))
705    pcm_format = _format(ContentType.PCM_F32LE, 48000, 32)
706
707    plan = audio.get_player_output_plan(
708        "player-1",
709        pcm_format,
710        _format(ContentType.FLAC, 48000, 16, channels=1),
711        handoff_format=pcm_format,
712        queue_id="queue-1",
713        session_id="session-1",
714        queue_item_id="item-1",
715    )
716
717    assert plan.filter_params == ["pan=stereo|c0=0.5*FL+0.5*FR|c1=0.5*FL+0.5*FR"]
718
719
720def test_mono_downmix_prevents_bit_perfect_claim() -> None:
721    """A mono downmix alters the samples, even when every format stays stereo."""
722    manager, _mass, _queue_data, streamdetails, output_plan, _lossy_plan = _manager_context()
723    output_plan.output_details.source_channel = AudioChannel.ALL
724
725    manager.update_output(
726        "player-1",
727        output_plan,
728        queue_id="queue-1",
729        session_id="session-1",
730        queue_item_id="item-1",
731    )
732
733    assert streamdetails.audio_processing is not None
734    assert streamdetails.audio_processing.outputs[0].fidelity.bit_perfect is False
735
736
737def test_player_output_plan_excludes_neutral_filters() -> None:
738    """A filter that emits no FFmpeg params is left out of the reported chain."""
739    mass = MagicMock()
740    mass.players.get_player.return_value = None
741    mass.config.get_player_dsp_config.return_value = DSPConfig(
742        enabled=True,
743        filters=[ToneControlFilter(enabled=True)],
744    )
745    mass.config.get_raw_player_config_value.return_value = "stereo"
746    audio = StreamsAudio(cast("Any", mass))
747    audio_format = _format(ContentType.PCM_F32LE, 48000, 32)
748
749    plan = audio.get_player_output_plan(
750        "player-1",
751        audio_format,
752        audio_format,
753        queue_id="queue-1",
754        session_id="session-1",
755        queue_item_id="item-1",
756    )
757
758    assert plan.output_details.dsp.filters == []
759    assert not any(
760        isinstance(param, str) and param.startswith("equalizer=") for param in plan.filter_params
761    )
762
763
764def _convolution_plan(known_ir_ids: list[str]) -> AudioOutputPlan:
765    """Build an output plan for a player convolving with impulse response "abc123"."""
766    mass = MagicMock()
767    mass.players.get_player.return_value = None
768    mass.storage_path = "/storage"
769    mass.config.get_player_dsp_config.return_value = DSPConfig(
770        enabled=True,
771        filters=[ConvolutionFilter(enabled=True, ir_id="abc123")],
772    )
773    mass.config.get_dsp_irs.return_value = [{"ir_id": ir_id} for ir_id in known_ir_ids]
774    mass.config.get_raw_player_config_value.return_value = "stereo"
775    audio = StreamsAudio(cast("Any", mass))
776    audio_format = _format(ContentType.PCM_F32LE, 48000, 32)
777    return audio.get_player_output_plan(
778        "player-1",
779        audio_format,
780        audio_format,
781        queue_id="queue-1",
782        session_id="session-1",
783        queue_item_id="item-1",
784    )
785
786
787def test_player_output_plan_drops_convolution_with_unknown_ir() -> None:
788    """An impulse response with no stored record is left out rather than failing ffmpeg."""
789    plan = _convolution_plan(known_ir_ids=["other"])
790
791    assert plan.output_details.dsp.filters == []
792    assert not any(isinstance(param, ComplexFilter) for param in plan.filter_params)
793
794
795def test_player_output_plan_keeps_convolution_with_known_ir() -> None:
796    """An impulse response that is still stored convolves as configured."""
797    plan = _convolution_plan(known_ir_ids=["abc123"])
798
799    assert len(plan.output_details.dsp.filters) == 1
800    complex_filters = [param for param in plan.filter_params if isinstance(param, ComplexFilter)]
801    assert [f.inputs[0].path for f in complex_filters] == ["/storage/dsp_irs/abc123.wav"]
802
803
804def test_player_output_plan_prefers_rendering_player_channels() -> None:
805    """Output channels stored on the rendering player win over the parent's value."""
806    mass = MagicMock()
807    player = MagicMock(player_id="child-1", protocol_parent_id="parent-1")
808    player.state.active_group = None
809    player.state.synced_to = None
810    mass.players.get_player.return_value = player
811    mass.config.get_player_dsp_config.return_value = DSPConfig(enabled=False)
812    mass.config.get_raw_player_config_value.side_effect = lambda player_id, _key, default: (
813        "left" if player_id == "child-1" else default
814    )
815    audio = StreamsAudio(cast("Any", mass))
816    audio_format = _format(ContentType.PCM_F32LE, 48000, 32)
817
818    plan = audio.get_player_output_plan(
819        "child-1",
820        audio_format,
821        audio_format,
822        queue_id="queue-1",
823        session_id="session-1",
824        queue_item_id="item-1",
825    )
826
827    assert plan.output_details.source_channel == AudioChannel.FL
828    assert "pan=stereo|c0=FL|c1=FL" in plan.filter_params
829    # processing attribution still points at the visible parent player
830    assert mass.streams.audio_processing.update_output.call_args.args[0] == "parent-1"
831
832
833@pytest.mark.asyncio
834async def test_output_format_prefers_rendering_player_channels() -> None:
835    """The output format channel count follows the rendering player's own stored value."""
836    mass = MagicMock()
837    player = MagicMock(player_id="child-1", protocol_parent_id="parent-1")
838    player.get_supported_sample_rates.return_value = [(48000, 24)]
839    mass.config.get_raw_player_config_value.side_effect = lambda player_id, _key, default: (
840        "left" if player_id == "child-1" else default
841    )
842    audio = StreamsAudio(cast("Any", mass))
843
844    fmt = await audio.get_output_format("flac", player, 48000, 24, MediaType.TRACK)
845
846    assert fmt.channels == 1
847
848
849@pytest.mark.asyncio
850async def test_single_stream_handler_shares_native_group_members(
851    monkeypatch: pytest.MonkeyPatch,
852) -> None:
853    """The regular single-item HTTP stream registers native group members."""
854    controller, request, group_members = _native_stream_handler_context(monkeypatch)
855
856    with pytest.raises(_OutputPlanRequested):
857        await controller.serve_queue_item_stream(request)
858
859    assert controller.audio.get_player_output_plan.call_args.kwargs["shared_player_ids"] is (
860        group_members
861    )
862
863
864@pytest.mark.asyncio
865async def test_flow_stream_handler_shares_native_group_members(
866    monkeypatch: pytest.MonkeyPatch,
867) -> None:
868    """The regular flow HTTP stream registers native group members."""
869    controller, request, group_members = _native_stream_handler_context(monkeypatch)
870
871    with pytest.raises(_OutputPlanRequested):
872        await controller.serve_queue_flow_stream(request)
873
874    assert controller.audio.get_player_output_plan.call_args.kwargs["shared_player_ids"] is (
875        group_members
876    )
877
878
879def test_protocol_output_uses_parent_settings(monkeypatch: pytest.MonkeyPatch) -> None:
880    """Protocol output details use the user-facing parent configuration."""
881    mass = MagicMock()
882    player = MagicMock(player_id="protocol-1", protocol_parent_id="player-1")
883    player.state.active_group = None
884    player.state.synced_to = None
885    shared_player = MagicMock(player_id="protocol-2", protocol_parent_id="player-2")
886    mass.players.get_player.side_effect = lambda player_id: {
887        "protocol-1": player,
888        "protocol-2": shared_player,
889    }.get(player_id)
890    mass.config.get_player_dsp_config.return_value = DSPConfig(enabled=False)
891    mass.config.get_raw_player_config_value.side_effect = lambda _player_id, _key, default: (
892        False if isinstance(default, bool) else "right"
893    )
894    audio = StreamsAudio(cast("Any", mass))
895    monkeypatch.setattr(
896        audio,
897        "_resolve_player_dsp_config",
898        lambda _player: DSPConfig(preset_id="parent-preset"),
899    )
900    pcm_format = _format(ContentType.PCM_S16LE, 44100, 16)
901
902    plan = audio.get_player_output_plan(
903        "protocol-1",
904        pcm_format,
905        pcm_format,
906        shared_player_ids={"protocol-1", "protocol-2"},
907        queue_id="queue-1",
908        session_id="session-1",
909    )
910
911    assert plan.output_details.player_ids == ["player-1", "player-2"]
912    assert plan.output_details.dsp.preset_id == "parent-preset"
913    assert plan.dsp_config_id == "player-1"
914    assert plan.output_details.source_channel == AudioChannel.FR
915    # the output channels are looked up on the rendering player first (no value
916    # stored there in this scenario), then resolved from the user-facing parent
917    assert {call.args[0] for call in mass.config.get_raw_player_config_value.call_args_list} == {
918        "player-1",
919        "protocol-1",
920    }
921    mass.streams.audio_processing.update_output.assert_called_once_with(
922        "player-1",
923        plan,
924        shared_player_ids={"player-2"},
925        queue_id="queue-1",
926        session_id="session-1",
927        queue_item_id=None,
928    )
929
930
931def test_single_member_group_uses_child_dsp_preset() -> None:
932    """A single-member player group reports the child's effective preset."""
933    mass = MagicMock()
934    player = MagicMock(player_id="group-1", protocol_parent_id=None)
935    player.provider.domain = "player_group"
936    player.state.active_group = None
937    player.state.synced_to = None
938    player.state.group_members = ["child-1"]
939    player.state.supported_features = set()
940    child = MagicMock(player_id="child-1")
941    mass.players.get_player.side_effect = lambda player_id: (
942        player if player_id == "group-1" else child
943    )
944    mass.config.get_player_dsp_config.side_effect = lambda player_id: (
945        DSPConfig(enabled=True, preset_id="child-preset")
946        if player_id == "child-1"
947        else DSPConfig(enabled=False)
948    )
949    mass.config.get_raw_player_config_value.return_value = "stereo"
950    audio = StreamsAudio(cast("Any", mass))
951    pcm_format = _format(ContentType.PCM_F32LE, 48000, 32)
952
953    plan = audio.get_player_output_plan("group-1", pcm_format, pcm_format)
954
955    assert plan.dsp_config_id == "child-1"
956    assert plan.output_details.dsp.state == DSPState.ENABLED
957    assert plan.output_details.dsp.preset_id == "child-preset"
958
959
960def test_unsupported_group_preserves_configured_preset() -> None:
961    """Runtime DSP suppression retains the selected preset identity."""
962    mass = MagicMock()
963    player = MagicMock(player_id="leader-1", protocol_parent_id=None)
964    player.provider.domain = "test"
965    player.state.active_group = None
966    player.state.synced_to = None
967    player.state.group_members = ["child-1"]
968    player.state.supported_features = set()
969    mass.players.get_player.return_value = player
970    mass.config.get_player_dsp_config.side_effect = lambda _player_id: DSPConfig(
971        enabled=True,
972        preset_id="group-preset",
973    )
974    mass.config.get_raw_player_config_value.return_value = "stereo"
975    audio = StreamsAudio(cast("Any", mass))
976    pcm_format = _format(ContentType.PCM_F32LE, 48000, 32)
977
978    plan = audio.get_player_output_plan("leader-1", pcm_format, pcm_format)
979
980    assert plan.dsp_config_id == "leader-1"
981    assert plan.output_details.dsp.state == DSPState.DISABLED_BY_UNSUPPORTED_GROUP
982    assert plan.output_details.dsp.preset_id == "group-preset"
983
984
985@pytest.mark.asyncio
986async def test_stale_flow_generator_does_not_mutate_active_session() -> None:
987    """A deferred flow generator exits before clearing newer session state."""
988    mass = MagicMock()
989    queue_data = SimpleNamespace(session_id="session-2", flow_mode_stream_log=["current"])
990    mass.player_queues.queue_data.return_value = queue_data
991    audio = StreamsAudio(cast("Any", mass))
992    queue = SimpleNamespace(queue_id="queue-1", display_name="Queue", flow_mode=False)
993    stream = audio.get_queue_flow_stream(
994        cast("Any", queue),
995        MagicMock(),
996        _format(ContentType.PCM_F32LE, 48000, 32),
997        session_id="session-1",
998    )
999
1000    with pytest.raises(StopAsyncIteration):
1001        await anext(stream)
1002
1003    assert not queue.flow_mode
1004    assert queue_data.flow_mode_stream_log == ["current"]
1005
1006
1007@pytest.mark.asyncio
1008async def test_duplicate_flow_producer_does_not_interleave_the_play_log() -> None:
1009    """A second flow request for one session keeps the play log of the first out of the queue."""
1010
1011    def _flow_item(item_id: str) -> Any:
1012        return SimpleNamespace(
1013            queue_item_id=item_id,
1014            name=item_id,
1015            media_type=MediaType.TRACK,
1016            duration=300,
1017            extra_attributes={},
1018            streamdetails=SimpleNamespace(
1019                fade_in=False,
1020                stream_error=False,
1021                uri=f"test://{item_id}",
1022                seek_position=0,
1023                duration=300,
1024                buffer=None,
1025                seconds_streamed=None,
1026                is_realtime=False,
1027                audio_format=_format(ContentType.PCM_F32LE, 48000, 32),
1028            ),
1029        )
1030
1031    items = {item_id: _flow_item(item_id) for item_id in ("item-1", "item-2")}
1032
1033    async def _load_next(_queue_id: str, current_id: str) -> Any:
1034        if current_id == "item-1":
1035            return items["item-2"]
1036        raise QueueEmpty
1037
1038    mass = MagicMock()
1039    queue_data = SimpleNamespace(session_id="session-1", flow_mode_stream_log=[])
1040    mass.player_queues.queue_data.return_value = queue_data
1041    mass.player_queues.load_next_queue_item = _load_next
1042    mass.streams.get_crossfade_mode.return_value = CrossfadeMode.DISABLED
1043    mass.config.get_raw_core_config_value.return_value = 0
1044    mass.player_queues.get_active_queue.return_value = None
1045    audio = StreamsAudio(cast("Any", mass))
1046
1047    async def _one_chunk(*_args: object, **_kwargs: object) -> AsyncGenerator[bytes]:
1048        yield b"\x00" * 16
1049
1050    audio.get_queue_item_stream = _one_chunk  # type: ignore[method-assign]
1051    queue = cast(
1052        "Any",
1053        SimpleNamespace(
1054            queue_id="queue-1",
1055            display_name="Queue",
1056            flow_mode=False,
1057            overlay_enabled=False,
1058            overlay_source=None,
1059        ),
1060    )
1061    pcm_format = _format(ContentType.PCM_F32LE, 48000, 32)
1062
1063    # the probing connection opens the flow url and logs its first track
1064    first = audio.get_queue_flow_stream(queue, items["item-1"], pcm_format, session_id="session-1")
1065    await anext(first)
1066    assert [entry.queue_item_id for entry in queue_data.flow_mode_stream_log] == ["item-1"]
1067
1068    # the connection that really plays opens the same url and publishes its own play log
1069    second = audio.get_queue_flow_stream(queue, items["item-1"], pcm_format, session_id="session-1")
1070    await anext(second)
1071    live_log = queue_data.flow_mode_stream_log
1072
1073    # the first producer moves on to its next track; that entry must not reach the live log
1074    await anext(first)
1075    assert queue_data.flow_mode_stream_log is live_log
1076    assert [entry.queue_item_id for entry in live_log] == ["item-1"]
1077
1078    await first.aclose()
1079    await second.aclose()
1080
1081
1082@pytest.mark.asyncio
1083async def test_flow_source_error_skips_item_without_completing_it() -> None:
1084    """An item-stream error skips to the next queue item; the flow itself continues."""
1085    mass = MagicMock()
1086    streamdetails = SimpleNamespace(
1087        fade_in=False,
1088        stream_error=False,
1089        uri="audiobookshelf://book",
1090        seek_position=0,
1091        duration=3600,
1092        is_realtime=False,
1093    )
1094    queue_item = SimpleNamespace(
1095        queue_item_id="item-1",
1096        name="book",
1097        media_type=MediaType.AUDIOBOOK,
1098        streamdetails=streamdetails,
1099        extra_attributes={},
1100    )
1101    queue_data = SimpleNamespace(session_id="session-1", flow_mode_stream_log=[])
1102    mass.player_queues.queue_data.return_value = queue_data
1103    mass.player_queues.load_next_queue_item.side_effect = QueueEmpty
1104    mass.streams.get_crossfade_mode.return_value = CrossfadeMode.DISABLED
1105    mass.config.get_raw_core_config_value.return_value = 0
1106    mass.streams.audio_processing.update_item_context = MagicMock()
1107    mass.player_queues.queue_buffer_completed = MagicMock()
1108    mass.player_queues.get_active_queue.return_value = None
1109    audio = StreamsAudio(cast("Any", mass))
1110
1111    async def _failed_stream(*_args: object, **_kwargs: object) -> AsyncGenerator[bytes]:
1112        yield b"buffered audio"
1113        streamdetails.stream_error = True
1114
1115    audio.get_queue_item_stream = _failed_stream  # type: ignore[method-assign]
1116    stream = audio.get_queue_flow_stream(
1117        cast(
1118            "Any",
1119            SimpleNamespace(
1120                queue_id="queue-1",
1121                display_name="Queue",
1122                flow_mode=False,
1123                overlay_enabled=False,
1124                overlay_source=None,
1125            ),
1126        ),
1127        cast("Any", queue_item),
1128        _format(ContentType.PCM_F32LE, 48000, 32),
1129        session_id="session-1",
1130    )
1131
1132    chunks = [chunk async for chunk in stream]
1133
1134    assert chunks == [b"buffered audio"]
1135    # the flow ran to natural completion (next item lookup raised QueueEmpty)
1136    mass.player_queues.queue_buffer_completed.assert_called_once()
1137    # the play log entry is kept, honest about the partial amount actually sent
1138    assert len(queue_data.flow_mode_stream_log) == 1
1139    entry = queue_data.flow_mode_stream_log[0]
1140    assert entry.queue_item_id == "item-1"
1141    assert entry.seconds_streamed is not None
1142    assert entry.seconds_streamed > 0
1143
1144
1145@pytest.mark.asyncio
1146async def test_flow_zero_audio_skip_restores_seek_position(
1147    monkeypatch: pytest.MonkeyPatch,
1148) -> None:
1149    """A zero-audio item keeps its original seek position when its crossfade is skipped."""
1150    mass = MagicMock()
1151    pcm_format = _format(ContentType.PCM_S16LE, 8000, 16)
1152    first_streamdetails = SimpleNamespace(
1153        audio_format=pcm_format,
1154        fade_in=False,
1155        stream_error=False,
1156        uri="test://first",
1157        seek_position=0,
1158        seconds_streamed=0,
1159        duration=120,
1160        buffer=SimpleNamespace(eof=True, cancelled=False, has_error=False, max_size_seconds=300),
1161        is_realtime=False,
1162    )
1163    first_item = SimpleNamespace(
1164        queue_id="queue-1",
1165        queue_item_id="item-1",
1166        name="first",
1167        media_type=MediaType.TRACK,
1168        media_item=None,
1169        streamdetails=first_streamdetails,
1170        extra_attributes={},
1171    )
1172    raw_seek_position = 12
1173    skipped_streamdetails = SimpleNamespace(
1174        audio_format=pcm_format,
1175        buffer=SimpleNamespace(
1176            has_error=False,
1177            is_valid=lambda *_args: True,
1178            duration_available=16,
1179            eof=False,
1180            ready=SimpleNamespace(is_set=lambda: True),
1181        ),
1182        fade_in=False,
1183        stream_error=False,
1184        uri="test://skipped",
1185        seek_position=raw_seek_position,
1186        seconds_streamed=0,
1187        duration=120,
1188        is_realtime=False,
1189    )
1190    skipped_item = SimpleNamespace(
1191        queue_id="queue-1",
1192        queue_item_id="item-2",
1193        name="skipped",
1194        media_type=MediaType.TRACK,
1195        media_item=None,
1196        streamdetails=skipped_streamdetails,
1197        extra_attributes={"playback_speed": 2.0},
1198    )
1199    queue = SimpleNamespace(
1200        queue_id="queue-1",
1201        display_name="Queue",
1202        flow_mode=False,
1203        overlay_enabled=False,
1204        overlay_source=None,
1205    )
1206    queue_data = SimpleNamespace(session_id="session-1", flow_mode_stream_log=[])
1207    mass.player_queues.queue_data.return_value = queue_data
1208    mass.player_queues.load_next_queue_item = AsyncMock(side_effect=[skipped_item, QueueEmpty])
1209    mass.player_queues.get.return_value = queue
1210    mass.player_queues.get_next_item.return_value = skipped_item
1211    mass.streams.get_crossfade_mode.return_value = CrossfadeMode.STANDARD_CROSSFADE
1212    mass.streams.get_source_crossfade_mode.return_value = CrossfadeMode.DISABLED
1213    mass.config.get_raw_core_config_value.return_value = 8
1214    mass.streams.audio_processing.update_item_context = MagicMock()
1215    mass.player_queues.queue_buffer_completed = MagicMock()
1216    player = MagicMock()
1217    player.config.get_value.return_value = "fixed_48000"
1218    player.get_supported_sample_rates.return_value = []
1219    mass.players.get_player.return_value = player
1220    audio = StreamsAudio(cast("Any", mass))
1221    audio.setup()
1222    build = AsyncMock(
1223        return_value=SimpleNamespace(
1224            timing_info=SimpleNamespace(
1225                fadein_trimmed_duration=2,
1226                crossfade_duration=8,
1227            )
1228        )
1229    )
1230    monkeypatch.setattr(audio.smart_fades_mixer, "build", build)
1231    eager_seek_positions: list[float] = []
1232
1233    async def _item_stream(
1234        queue_item: SimpleNamespace,
1235        *_args: object,
1236        **_kwargs: object,
1237    ) -> AsyncGenerator[bytes]:
1238        if queue_item is first_item:
1239            # warmup worth of audio, then a full crossfade tail
1240            yield bytes(pcm_format.pcm_sample_size * 8)
1241            yield bytes(pcm_format.pcm_sample_size * 8)
1242        else:
1243            eager_seek_positions.append(queue_item.streamdetails.seek_position)
1244
1245    monkeypatch.setattr(audio, "get_queue_item_stream", _item_stream)
1246    stream = audio.get_queue_flow_stream(
1247        cast("Any", queue),
1248        cast("Any", first_item),
1249        pcm_format,
1250        session_id="session-1",
1251    )
1252
1253    async for _ in stream:
1254        pass
1255
1256    build.assert_awaited_once()
1257    # the prefetcher's early open sees the raw position; the reopen after the failed
1258    # handover sees the eager (crossfade-adjusted) one
1259    assert len(eager_seek_positions) == 2
1260    assert eager_seek_positions[-1] == 32
1261    # ... and the zero-audio skip restores the raw position afterwards
1262    assert skipped_streamdetails.seek_position == raw_seek_position
1263
1264
1265@pytest.mark.parametrize(
1266    ("source_cancelled", "expected_duration"),
1267    [(True, 300), (False, 3)],
1268    ids=["aborted_source", "clean_source"],
1269)
1270@pytest.mark.asyncio
1271async def test_flow_does_not_write_back_a_duration_for_an_aborted_source(
1272    monkeypatch: pytest.MonkeyPatch, source_cancelled: bool, expected_duration: int
1273) -> None:
1274    """An externally cancelled buffer ends in a clean EOF that must not shorten the item."""
1275    mass = MagicMock()
1276    pcm_format = _format(ContentType.PCM_S16LE, 8000, 16)
1277    streamdetails = SimpleNamespace(
1278        audio_format=pcm_format,
1279        buffer=SimpleNamespace(cancelled=source_cancelled),
1280        fade_in=False,
1281        stream_error=False,
1282        uri="test://track",
1283        seek_position=0,
1284        seconds_streamed=0,
1285        duration=300,
1286        is_realtime=False,
1287    )
1288    queue_track = SimpleNamespace(
1289        queue_id="queue-1",
1290        queue_item_id="item-1",
1291        name="track",
1292        media_type=MediaType.TRACK,
1293        media_item=None,
1294        streamdetails=streamdetails,
1295        duration=300,
1296        extra_attributes={},
1297    )
1298    queue = SimpleNamespace(
1299        queue_id="queue-1",
1300        display_name="Queue",
1301        flow_mode=False,
1302        overlay_enabled=False,
1303        overlay_source=None,
1304    )
1305    queue_data = SimpleNamespace(session_id="session-1", flow_mode_stream_log=[])
1306    mass.player_queues.queue_data.return_value = queue_data
1307    mass.player_queues.load_next_queue_item = AsyncMock(side_effect=QueueEmpty)
1308    mass.player_queues.get.return_value = queue
1309    mass.streams.get_crossfade_mode.return_value = CrossfadeMode.DISABLED
1310    mass.config.get_raw_core_config_value.return_value = 8
1311    mass.streams.audio_processing.update_item_context = MagicMock()
1312    mass.player_queues.queue_buffer_completed = MagicMock()
1313    player = MagicMock()
1314    player.config.get_value.return_value = "fixed_48000"
1315    player.get_supported_sample_rates.return_value = []
1316    mass.players.get_player.return_value = player
1317    audio = StreamsAudio(cast("Any", mass))
1318    audio.setup()
1319
1320    async def _item_stream(*_args: object, **_kwargs: object) -> AsyncGenerator[bytes]:
1321        # a cancelled buffer stops yielding without an error, exactly like a real EOF
1322        for _ in range(3):
1323            yield bytes(pcm_format.pcm_sample_size)
1324
1325    monkeypatch.setattr(audio, "get_queue_item_stream", _item_stream)
1326    stream = audio.get_queue_flow_stream(
1327        cast("Any", queue), cast("Any", queue_track), pcm_format, session_id="session-1"
1328    )
1329
1330    chunks = [chunk async for chunk in stream]
1331
1332    assert len(chunks) == 3
1333    assert streamdetails.duration == expected_duration
1334    assert queue_track.duration == expected_duration
1335    # the honest streamed amount is always recorded, only the duration is protected
1336    assert streamdetails.seconds_streamed == 3
1337    entry = queue_data.flow_mode_stream_log[0]
1338    assert entry.seconds_streamed == 3
1339    assert entry.duration == (None if source_cancelled else 3)
1340
1341
1342def _manager_context(
1343    *,
1344    alters_audio: bool = False,
1345) -> tuple[
1346    AudioProcessingManager,
1347    MagicMock,
1348    SimpleNamespace,
1349    StreamDetails,
1350    AudioOutputPlan,
1351    AudioOutputPlan,
1352]:
1353    """Return one prepared queue item and two output plan templates."""
1354    mass = MagicMock()
1355    streamdetails = _streamdetails()
1356    queue_item = SimpleNamespace(queue_item_id="item-1", streamdetails=streamdetails)
1357    queue = SimpleNamespace(
1358        queue_id="queue-1",
1359        current_item=queue_item,
1360        next_item=None,
1361        current_index=0,
1362    )
1363    queue_data = SimpleNamespace(session_id="session-1", items=[queue_item], queue=queue)
1364    mass.player_queues.get.return_value = queue
1365    mass.player_queues.get_active_queue.return_value = queue
1366    mass.player_queues.get_item.return_value = queue_item
1367    mass.player_queues.queue_data_or_none.return_value = queue_data
1368    manager = AudioProcessingManager(mass)
1369    pcm_format = _format(ContentType.PCM_S24LE, 96000, 24)
1370    manager.start_session("queue-1", "session-1")
1371    manager.update_item_context(
1372        "queue-1",
1373        "session-1",
1374        "item-1",
1375        AudioQueueProcessing(pcm_format=pcm_format),
1376        alters_audio=alters_audio,
1377    )
1378    lossless_plan = AudioOutputPlan(
1379        filter_params=[],
1380        output_details=AudioOutputDetails(
1381            dsp=AudioDSPDetails(state=DSPState.DISABLED),
1382            output_format=_format(ContentType.FLAC, 96000, 24),
1383        ),
1384        input_format=pcm_format,
1385    )
1386    lossy_plan = AudioOutputPlan(
1387        filter_params=[],
1388        output_details=AudioOutputDetails(
1389            dsp=AudioDSPDetails(state=DSPState.DISABLED),
1390            output_format=_format(ContentType.MP3, 48000, 16, bit_rate=128),
1391        ),
1392        input_format=pcm_format,
1393    )
1394    return manager, mass, queue_data, streamdetails, lossless_plan, lossy_plan
1395
1396
1397def _source_handled_soloist_item(
1398    output_format: AudioFormat,
1399) -> StreamDetails:
1400    """Prepare a Spotify-soloist-shaped item: a 24-bit tier delivered as 32-bit PCM."""
1401    manager, _mass, _queue_data, streamdetails, lossless_plan, _lossy_plan = _manager_context()
1402    streamdetails.audio_format = _format(ContentType.FLAC, 44100, 24)
1403    streamdetails.decoded_audio_format = _format(ContentType.PCM_S32LE, 44100, 32)
1404    pcm_format = _format(ContentType.PCM_S32LE, 44100, 32)
1405    manager.update_item_runtime(
1406        "queue-1",
1407        "session-1",
1408        "item-1",
1409        input_format=pcm_format,
1410        pcm_format=pcm_format,
1411        normalization=AudioNormalizationDetails(mode=VolumeNormalizationMode.SOURCE),
1412        playback_speed=1.0,
1413    )
1414    manager.update_item_context(
1415        "queue-1",
1416        "session-1",
1417        "item-1",
1418        AudioQueueProcessing(pcm_format=pcm_format, crossfade_mode=CrossfadeMode.SOURCE),
1419    )
1420    lossless_plan.input_format = pcm_format
1421    lossless_plan.output_details.output_format = output_format
1422    manager.update_output(
1423        "player-1",
1424        lossless_plan,
1425        queue_id="queue-1",
1426        session_id="session-1",
1427        queue_item_id="item-1",
1428    )
1429    return streamdetails
1430
1431
1432def _streamdetails(item_id: str = "item-1") -> StreamDetails:
1433    """Return hi-res lossless stream details."""
1434    return StreamDetails(
1435        provider="provider",
1436        item_id=item_id,
1437        audio_format=_format(ContentType.FLAC, 96000, 24, bit_rate=3200),
1438        media_type=MediaType.TRACK,
1439    )
1440
1441
1442class _OutputPlanRequested(Exception):
1443    """Signal that a stream handler reached output planning."""
1444
1445
1446def _native_stream_handler_context(
1447    monkeypatch: pytest.MonkeyPatch,
1448) -> tuple[Any, MagicMock, list[str]]:
1449    """Return a native HTTP stream handler prepared to stop at output planning."""
1450    mass = MagicMock()
1451    streamdetails = _streamdetails()
1452    queue_item = SimpleNamespace(
1453        queue_id="queue-1",
1454        queue_item_id="item-1",
1455        name="Track",
1456        duration=180,
1457        streamdetails=streamdetails,
1458        media_item=None,
1459        media_type=MediaType.TRACK,
1460        extra_attributes={},
1461        image=None,
1462    )
1463    queue = SimpleNamespace(
1464        queue_id="queue-1",
1465        display_name="Queue",
1466        current_item=queue_item,
1467        crossfade_enabled=False,
1468        overlay_enabled=False,
1469        overlay_source=None,
1470    )
1471    queue_data = SimpleNamespace(session_id="session-1")
1472    mass.player_queues.get.return_value = queue
1473    mass.player_queues.queue_data.return_value = queue_data
1474    mass.player_queues.get_item.return_value = queue_item
1475    mass.config.get_raw_core_config_value.return_value = 8
1476    mass.config.get_raw_player_config_value.return_value = "disabled"
1477
1478    group_members = ["player-1", "player-2"]
1479    player = MagicMock(player_id="player-1", protocol_parent_id=None)
1480    player.state.group_members = group_members
1481    player.state.supported_features = set()
1482    player.state.name = "Player"
1483    player.get_config_value.return_value = "default"
1484    mass.players.get_player.return_value = player
1485
1486    pcm_format = _format(ContentType.PCM_F32LE, 48000, 32)
1487    output_format = _format(ContentType.FLAC, 48000, 24)
1488    audio = MagicMock()
1489    audio.select_pcm_format = AsyncMock(return_value=pcm_format)
1490    audio.select_flow_pcm_format = AsyncMock(return_value=pcm_format)
1491    audio.get_output_format = AsyncMock(return_value=output_format)
1492    audio.get_player_output_plan.side_effect = _OutputPlanRequested
1493
1494    controller = cast("Any", object.__new__(StreamsController))
1495    controller.mass = mass
1496    controller.audio = audio
1497    controller.logger = MagicMock()
1498    controller._log_request = MagicMock()
1499    controller._update_audio_processing_context = MagicMock()
1500    controller._active_output_streams = 0
1501
1502    response = MagicMock()
1503    response.prepare = AsyncMock()
1504    monkeypatch.setattr(
1505        "music_assistant.controllers.streams.controller.web.StreamResponse",
1506        MagicMock(return_value=response),
1507    )
1508    request = MagicMock()
1509    request.method = "GET"
1510    request.headers = {}
1511    request.match_info = {
1512        "queue_id": "queue-1",
1513        "session_id": "session-1",
1514        "queue_item_id": "item-1",
1515        "player_id": "player-1",
1516        "fmt": "flac",
1517    }
1518    return controller, request, group_members
1519