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                is_realtime=False,
953                audio_format=_format(ContentType.PCM_F32LE, 48000, 32),
954            ),
955        )
956
957    items = {item_id: _flow_item(item_id) for item_id in ("item-1", "item-2")}
958
959    async def _load_next(_queue_id: str, current_id: str) -> Any:
960        if current_id == "item-1":
961            return items["item-2"]
962        raise QueueEmpty
963
964    mass = MagicMock()
965    queue_data = SimpleNamespace(session_id="session-1", flow_mode_stream_log=[])
966    mass.player_queues.queue_data.return_value = queue_data
967    mass.player_queues.load_next_queue_item = _load_next
968    mass.streams.get_crossfade_mode.return_value = CrossfadeMode.DISABLED
969    mass.config.get_raw_core_config_value.return_value = 0
970    mass.player_queues.get_active_queue.return_value = None
971    audio = StreamsAudio(cast("Any", mass))
972
973    async def _one_chunk(*_args: object, **_kwargs: object) -> AsyncGenerator[bytes]:
974        yield b"\x00" * 16
975
976    audio.get_queue_item_stream = _one_chunk  # type: ignore[method-assign]
977    queue = cast(
978        "Any",
979        SimpleNamespace(
980            queue_id="queue-1",
981            display_name="Queue",
982            flow_mode=False,
983            overlay_enabled=False,
984            overlay_source=None,
985        ),
986    )
987    pcm_format = _format(ContentType.PCM_F32LE, 48000, 32)
988
989    # the probing connection opens the flow url and logs its first track
990    first = audio.get_queue_flow_stream(queue, items["item-1"], pcm_format, session_id="session-1")
991    await anext(first)
992    assert [entry.queue_item_id for entry in queue_data.flow_mode_stream_log] == ["item-1"]
993
994    # the connection that really plays opens the same url and publishes its own play log
995    second = audio.get_queue_flow_stream(queue, items["item-1"], pcm_format, session_id="session-1")
996    await anext(second)
997    live_log = queue_data.flow_mode_stream_log
998
999    # the first producer moves on to its next track; that entry must not reach the live log
1000    await anext(first)
1001    assert queue_data.flow_mode_stream_log is live_log
1002    assert [entry.queue_item_id for entry in live_log] == ["item-1"]
1003
1004    await first.aclose()
1005    await second.aclose()
1006
1007
1008@pytest.mark.asyncio
1009async def test_flow_source_error_skips_item_without_completing_it() -> None:
1010    """An item-stream error skips to the next queue item; the flow itself continues."""
1011    mass = MagicMock()
1012    streamdetails = SimpleNamespace(
1013        fade_in=False,
1014        stream_error=False,
1015        uri="audiobookshelf://book",
1016        seek_position=0,
1017        duration=3600,
1018        is_realtime=False,
1019    )
1020    queue_item = SimpleNamespace(
1021        queue_item_id="item-1",
1022        name="book",
1023        media_type=MediaType.AUDIOBOOK,
1024        streamdetails=streamdetails,
1025        extra_attributes={},
1026    )
1027    queue_data = SimpleNamespace(session_id="session-1", flow_mode_stream_log=[])
1028    mass.player_queues.queue_data.return_value = queue_data
1029    mass.player_queues.load_next_queue_item.side_effect = QueueEmpty
1030    mass.streams.get_crossfade_mode.return_value = CrossfadeMode.DISABLED
1031    mass.config.get_raw_core_config_value.return_value = 0
1032    mass.streams.audio_processing.update_item_context = MagicMock()
1033    mass.player_queues.queue_buffer_completed = MagicMock()
1034    mass.player_queues.get_active_queue.return_value = None
1035    audio = StreamsAudio(cast("Any", mass))
1036
1037    async def _failed_stream(*_args: object, **_kwargs: object) -> AsyncGenerator[bytes]:
1038        yield b"buffered audio"
1039        streamdetails.stream_error = True
1040
1041    audio.get_queue_item_stream = _failed_stream  # type: ignore[method-assign]
1042    stream = audio.get_queue_flow_stream(
1043        cast(
1044            "Any",
1045            SimpleNamespace(
1046                queue_id="queue-1",
1047                display_name="Queue",
1048                flow_mode=False,
1049                overlay_enabled=False,
1050                overlay_source=None,
1051            ),
1052        ),
1053        cast("Any", queue_item),
1054        _format(ContentType.PCM_F32LE, 48000, 32),
1055        session_id="session-1",
1056    )
1057
1058    chunks = [chunk async for chunk in stream]
1059
1060    assert chunks == [b"buffered audio"]
1061    # the flow ran to natural completion (next item lookup raised QueueEmpty)
1062    mass.player_queues.queue_buffer_completed.assert_called_once()
1063    # the play log entry is kept, honest about the partial amount actually sent
1064    assert len(queue_data.flow_mode_stream_log) == 1
1065    entry = queue_data.flow_mode_stream_log[0]
1066    assert entry.queue_item_id == "item-1"
1067    assert entry.seconds_streamed is not None
1068    assert entry.seconds_streamed > 0
1069
1070
1071@pytest.mark.asyncio
1072async def test_flow_zero_audio_skip_restores_seek_position(
1073    monkeypatch: pytest.MonkeyPatch,
1074) -> None:
1075    """A zero-audio item keeps its original seek position when its crossfade is skipped."""
1076    mass = MagicMock()
1077    pcm_format = _format(ContentType.PCM_S16LE, 8000, 16)
1078    first_streamdetails = SimpleNamespace(
1079        audio_format=pcm_format,
1080        fade_in=False,
1081        stream_error=False,
1082        uri="test://first",
1083        seek_position=0,
1084        seconds_streamed=0,
1085        duration=120,
1086        buffer=SimpleNamespace(eof=True, cancelled=False, has_error=False, max_size_seconds=300),
1087        is_realtime=False,
1088    )
1089    first_item = SimpleNamespace(
1090        queue_id="queue-1",
1091        queue_item_id="item-1",
1092        name="first",
1093        media_type=MediaType.TRACK,
1094        media_item=None,
1095        streamdetails=first_streamdetails,
1096        extra_attributes={},
1097    )
1098    raw_seek_position = 12
1099    skipped_streamdetails = SimpleNamespace(
1100        audio_format=pcm_format,
1101        buffer=SimpleNamespace(
1102            has_error=False,
1103            is_valid=lambda *_args: True,
1104            duration_available=16,
1105            ready=SimpleNamespace(is_set=lambda: True),
1106        ),
1107        fade_in=False,
1108        stream_error=False,
1109        uri="test://skipped",
1110        seek_position=raw_seek_position,
1111        seconds_streamed=0,
1112        duration=120,
1113        is_realtime=False,
1114    )
1115    skipped_item = SimpleNamespace(
1116        queue_id="queue-1",
1117        queue_item_id="item-2",
1118        name="skipped",
1119        media_type=MediaType.TRACK,
1120        media_item=None,
1121        streamdetails=skipped_streamdetails,
1122        extra_attributes={"playback_speed": 2.0},
1123    )
1124    queue = SimpleNamespace(
1125        queue_id="queue-1",
1126        display_name="Queue",
1127        flow_mode=False,
1128        overlay_enabled=False,
1129        overlay_source=None,
1130    )
1131    queue_data = SimpleNamespace(session_id="session-1", flow_mode_stream_log=[])
1132    mass.player_queues.queue_data.return_value = queue_data
1133    mass.player_queues.load_next_queue_item = AsyncMock(side_effect=[skipped_item, QueueEmpty])
1134    mass.player_queues.get.return_value = queue
1135    mass.player_queues.get_next_item.return_value = skipped_item
1136    mass.streams.get_crossfade_mode.return_value = CrossfadeMode.STANDARD_CROSSFADE
1137    mass.config.get_raw_core_config_value.return_value = 8
1138    mass.streams.audio_processing.update_item_context = MagicMock()
1139    mass.player_queues.queue_buffer_completed = MagicMock()
1140    player = MagicMock()
1141    player.config.get_value.return_value = "fixed_48000"
1142    player.get_supported_sample_rates.return_value = []
1143    mass.players.get_player.return_value = player
1144    audio = StreamsAudio(cast("Any", mass))
1145    audio.setup()
1146    build = AsyncMock(
1147        return_value=SimpleNamespace(
1148            timing_info=SimpleNamespace(
1149                fadein_trimmed_duration=2,
1150                crossfade_duration=8,
1151            )
1152        )
1153    )
1154    monkeypatch.setattr(audio.smart_fades_mixer, "build", build)
1155    eager_seek_positions: list[float] = []
1156
1157    async def _item_stream(
1158        queue_item: SimpleNamespace,
1159        *_args: object,
1160        **_kwargs: object,
1161    ) -> AsyncGenerator[bytes]:
1162        if queue_item is first_item:
1163            # warmup worth of audio, then a full crossfade tail
1164            yield bytes(pcm_format.pcm_sample_size * 8)
1165            yield bytes(pcm_format.pcm_sample_size * 8)
1166        else:
1167            eager_seek_positions.append(queue_item.streamdetails.seek_position)
1168
1169    monkeypatch.setattr(audio, "get_queue_item_stream", _item_stream)
1170    stream = audio.get_queue_flow_stream(
1171        cast("Any", queue),
1172        cast("Any", first_item),
1173        pcm_format,
1174        session_id="session-1",
1175    )
1176
1177    async for _ in stream:
1178        pass
1179
1180    build.assert_awaited_once()
1181    # a source that hands over nothing is reopened, and both opens see the eager position
1182    assert eager_seek_positions == [32, 32]
1183    assert skipped_streamdetails.seek_position == raw_seek_position
1184
1185
1186@pytest.mark.parametrize(
1187    ("source_cancelled", "expected_duration"),
1188    [(True, 300), (False, 3)],
1189    ids=["aborted_source", "clean_source"],
1190)
1191@pytest.mark.asyncio
1192async def test_flow_does_not_write_back_a_duration_for_an_aborted_source(
1193    monkeypatch: pytest.MonkeyPatch, source_cancelled: bool, expected_duration: int
1194) -> None:
1195    """An externally cancelled buffer ends in a clean EOF that must not shorten the item."""
1196    mass = MagicMock()
1197    pcm_format = _format(ContentType.PCM_S16LE, 8000, 16)
1198    streamdetails = SimpleNamespace(
1199        audio_format=pcm_format,
1200        buffer=SimpleNamespace(cancelled=source_cancelled),
1201        fade_in=False,
1202        stream_error=False,
1203        uri="test://track",
1204        seek_position=0,
1205        seconds_streamed=0,
1206        duration=300,
1207        is_realtime=False,
1208    )
1209    queue_track = SimpleNamespace(
1210        queue_id="queue-1",
1211        queue_item_id="item-1",
1212        name="track",
1213        media_type=MediaType.TRACK,
1214        media_item=None,
1215        streamdetails=streamdetails,
1216        duration=300,
1217        extra_attributes={},
1218    )
1219    queue = SimpleNamespace(
1220        queue_id="queue-1",
1221        display_name="Queue",
1222        flow_mode=False,
1223        overlay_enabled=False,
1224        overlay_source=None,
1225    )
1226    queue_data = SimpleNamespace(session_id="session-1", flow_mode_stream_log=[])
1227    mass.player_queues.queue_data.return_value = queue_data
1228    mass.player_queues.load_next_queue_item = AsyncMock(side_effect=QueueEmpty)
1229    mass.player_queues.get.return_value = queue
1230    mass.streams.get_crossfade_mode.return_value = CrossfadeMode.DISABLED
1231    mass.config.get_raw_core_config_value.return_value = 8
1232    mass.streams.audio_processing.update_item_context = MagicMock()
1233    mass.player_queues.queue_buffer_completed = MagicMock()
1234    player = MagicMock()
1235    player.config.get_value.return_value = "fixed_48000"
1236    player.get_supported_sample_rates.return_value = []
1237    mass.players.get_player.return_value = player
1238    audio = StreamsAudio(cast("Any", mass))
1239    audio.setup()
1240
1241    async def _item_stream(*_args: object, **_kwargs: object) -> AsyncGenerator[bytes]:
1242        # a cancelled buffer stops yielding without an error, exactly like a real EOF
1243        for _ in range(3):
1244            yield bytes(pcm_format.pcm_sample_size)
1245
1246    monkeypatch.setattr(audio, "get_queue_item_stream", _item_stream)
1247    stream = audio.get_queue_flow_stream(
1248        cast("Any", queue), cast("Any", queue_track), pcm_format, session_id="session-1"
1249    )
1250
1251    chunks = [chunk async for chunk in stream]
1252
1253    assert len(chunks) == 3
1254    assert streamdetails.duration == expected_duration
1255    assert queue_track.duration == expected_duration
1256    # the honest streamed amount is always recorded, only the duration is protected
1257    assert streamdetails.seconds_streamed == 3
1258    entry = queue_data.flow_mode_stream_log[0]
1259    assert entry.seconds_streamed == 3
1260    assert entry.duration == (None if source_cancelled else 3)
1261
1262
1263def _manager_context(
1264    *,
1265    alters_audio: bool = False,
1266) -> tuple[
1267    AudioProcessingManager,
1268    MagicMock,
1269    SimpleNamespace,
1270    StreamDetails,
1271    AudioOutputPlan,
1272    AudioOutputPlan,
1273]:
1274    """Return one prepared queue item and two output plan templates."""
1275    mass = MagicMock()
1276    streamdetails = _streamdetails()
1277    queue_item = SimpleNamespace(queue_item_id="item-1", streamdetails=streamdetails)
1278    queue = SimpleNamespace(
1279        queue_id="queue-1",
1280        current_item=queue_item,
1281        next_item=None,
1282        current_index=0,
1283    )
1284    queue_data = SimpleNamespace(session_id="session-1", items=[queue_item], queue=queue)
1285    mass.player_queues.get.return_value = queue
1286    mass.player_queues.get_active_queue.return_value = queue
1287    mass.player_queues.get_item.return_value = queue_item
1288    mass.player_queues.queue_data_or_none.return_value = queue_data
1289    manager = AudioProcessingManager(mass)
1290    pcm_format = _format(ContentType.PCM_S24LE, 96000, 24)
1291    manager.start_session("queue-1", "session-1")
1292    manager.update_item_context(
1293        "queue-1",
1294        "session-1",
1295        "item-1",
1296        AudioQueueProcessing(pcm_format=pcm_format),
1297        alters_audio=alters_audio,
1298    )
1299    lossless_plan = AudioOutputPlan(
1300        filter_params=[],
1301        output_details=AudioOutputDetails(
1302            dsp=AudioDSPDetails(state=DSPState.DISABLED),
1303            output_format=_format(ContentType.FLAC, 96000, 24),
1304        ),
1305        input_format=pcm_format,
1306    )
1307    lossy_plan = AudioOutputPlan(
1308        filter_params=[],
1309        output_details=AudioOutputDetails(
1310            dsp=AudioDSPDetails(state=DSPState.DISABLED),
1311            output_format=_format(ContentType.MP3, 48000, 16, bit_rate=128),
1312        ),
1313        input_format=pcm_format,
1314    )
1315    return manager, mass, queue_data, streamdetails, lossless_plan, lossy_plan
1316
1317
1318def _streamdetails(item_id: str = "item-1") -> StreamDetails:
1319    """Return hi-res lossless stream details."""
1320    return StreamDetails(
1321        provider="provider",
1322        item_id=item_id,
1323        audio_format=_format(ContentType.FLAC, 96000, 24, bit_rate=3200),
1324        media_type=MediaType.TRACK,
1325    )
1326
1327
1328class _OutputPlanRequested(Exception):
1329    """Signal that a stream handler reached output planning."""
1330
1331
1332def _native_stream_handler_context(
1333    monkeypatch: pytest.MonkeyPatch,
1334) -> tuple[Any, MagicMock, list[str]]:
1335    """Return a native HTTP stream handler prepared to stop at output planning."""
1336    mass = MagicMock()
1337    streamdetails = _streamdetails()
1338    queue_item = SimpleNamespace(
1339        queue_id="queue-1",
1340        queue_item_id="item-1",
1341        name="Track",
1342        duration=180,
1343        streamdetails=streamdetails,
1344        media_item=None,
1345        media_type=MediaType.TRACK,
1346        extra_attributes={},
1347        image=None,
1348    )
1349    queue = SimpleNamespace(
1350        queue_id="queue-1",
1351        display_name="Queue",
1352        current_item=queue_item,
1353        crossfade_enabled=False,
1354        overlay_enabled=False,
1355        overlay_source=None,
1356    )
1357    queue_data = SimpleNamespace(session_id="session-1")
1358    mass.player_queues.get.return_value = queue
1359    mass.player_queues.queue_data.return_value = queue_data
1360    mass.player_queues.get_item.return_value = queue_item
1361    mass.config.get_raw_core_config_value.return_value = 8
1362    mass.config.get_raw_player_config_value.return_value = "disabled"
1363
1364    group_members = ["player-1", "player-2"]
1365    player = MagicMock(player_id="player-1", protocol_parent_id=None)
1366    player.state.group_members = group_members
1367    player.state.supported_features = set()
1368    player.state.name = "Player"
1369    player.get_config_value.return_value = "default"
1370    mass.players.get_player.return_value = player
1371
1372    pcm_format = _format(ContentType.PCM_F32LE, 48000, 32)
1373    output_format = _format(ContentType.FLAC, 48000, 24)
1374    audio = MagicMock()
1375    audio.select_pcm_format = AsyncMock(return_value=pcm_format)
1376    audio.select_flow_pcm_format = AsyncMock(return_value=pcm_format)
1377    audio.get_output_format = AsyncMock(return_value=output_format)
1378    audio.get_player_output_plan.side_effect = _OutputPlanRequested
1379
1380    controller = cast("Any", object.__new__(StreamsController))
1381    controller.mass = mass
1382    controller.audio = audio
1383    controller.logger = MagicMock()
1384    controller._log_request = MagicMock()
1385    controller._update_audio_processing_context = MagicMock()
1386    controller._active_output_streams = 0
1387
1388    response = MagicMock()
1389    response.prepare = AsyncMock()
1390    monkeypatch.setattr(
1391        "music_assistant.controllers.streams.controller.web.StreamResponse",
1392        MagicMock(return_value=response),
1393    )
1394    request = MagicMock()
1395    request.method = "GET"
1396    request.headers = {}
1397    request.match_info = {
1398        "queue_id": "queue-1",
1399        "session_id": "session-1",
1400        "queue_item_id": "item-1",
1401        "player_id": "player-1",
1402        "fmt": "flac",
1403    }
1404    return controller, request, group_members
1405