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