music-assistant-server

1"""Tests for effective audio processing plans and stream details."""
2
3from __future__ import annotations
4
5from copy import deepcopy
6from types import SimpleNamespace
7from typing import Any, cast
8from unittest.mock import AsyncMock, MagicMock
9
10import pytest
11from music_assistant_models.audio_processing import (
12    AudioDSPDetails,
13    AudioFidelity,
14    AudioNormalizationDetails,
15    AudioOutputDetails,
16    AudioProcessingChain,
17    AudioQuality,
18    AudioQueueProcessing,
19)
20from music_assistant_models.dsp import (
21    AudioChannel,
22    ConvolutionFilter,
23    DSPConfig,
24    DSPState,
25    ToneControlFilter,
26)
27from music_assistant_models.enums import (
28    ContentType,
29    CrossfadeMode,
30    MediaType,
31    VolumeNormalizationMode,
32)
33from music_assistant_models.media_items import AudioFormat
34from music_assistant_models.streamdetails import StreamDetails
35
36from music_assistant.controllers.streams.audio import StreamsAudio
37from music_assistant.controllers.streams.audio_processing import (
38    AudioOutputPlan,
39    AudioProcessingManager,
40    get_audio_quality,
41    get_normalization_details,
42)
43from music_assistant.controllers.streams.controller import StreamsController
44from music_assistant.helpers.dsp import ComplexFilter
45
46
47def _format(
48    content_type: ContentType,
49    sample_rate: int = 44100,
50    bit_depth: int = 16,
51    *,
52    channels: int = 2,
53    bit_rate: int | None = None,
54) -> AudioFormat:
55    """Return an AudioFormat with matching container and codec."""
56    return AudioFormat(
57        content_type=content_type,
58        codec_type=content_type,
59        sample_rate=sample_rate,
60        bit_depth=bit_depth,
61        channels=channels,
62        bit_rate=bit_rate,
63    )
64
65
66@pytest.mark.parametrize(
67    ("audio_format", "expected"),
68    [
69        (_format(ContentType.FLAC, 44100, 16), AudioQuality.LOSSLESS),
70        (_format(ContentType.FLAC, 96000, 24), AudioQuality.HI_RES),
71        (_format(ContentType.MP3, bit_rate=320), AudioQuality.STANDARD),
72        (_format(ContentType.AAC, bit_rate=128), AudioQuality.LOW),
73        (_format(ContentType.MP3, bit_rate=128000), AudioQuality.LOW),
74        (_format(ContentType.AAC), AudioQuality.UNKNOWN),
75    ],
76)
77def test_get_audio_quality(audio_format: AudioFormat, expected: AudioQuality) -> None:
78    """Quality classification uses codec, resolution and normalized bitrate."""
79    assert get_audio_quality(audio_format) == expected
80
81
82def test_get_normalization_details_uses_album_measurement() -> None:
83    """Album normalization reports the selected measurement and applied gain."""
84    streamdetails = _streamdetails()
85    streamdetails.volume_normalization_mode = VolumeNormalizationMode.MEASUREMENT_ONLY
86    streamdetails.prefer_album_loudness = True
87    streamdetails.loudness = -12.0
88    streamdetails.loudness_album = -14.5
89    streamdetails.target_loudness = -17.0
90
91    details = get_normalization_details(streamdetails, applied_gain_db=-2.5)
92
93    assert details is not None
94    assert details.measurement_source.value == "album"
95    assert details.measured_lufs == -14.5
96    assert details.target_lufs == -17.0
97    assert details.applied_gain_db == -2.5
98
99
100def test_audio_processing_manager_attaches_grouped_chain() -> None:
101    """A complete chain is attached to StreamDetails with grouped outputs."""
102    manager, _mass, _queue_data, streamdetails, lossless_plan, lossy_plan = _manager_context()
103    assert streamdetails.audio_processing is None
104
105    assert manager.update_output(
106        "player-2",
107        lossless_plan,
108        queue_id="queue-1",
109        session_id="session-1",
110        queue_item_id="item-1",
111    )
112    assert manager.update_output(
113        "player-1",
114        lossless_plan,
115        queue_id="queue-1",
116        session_id="session-1",
117        queue_item_id="item-1",
118    )
119    assert manager.update_output(
120        "player-3",
121        lossy_plan,
122        queue_id="queue-1",
123        session_id="session-1",
124        queue_item_id="item-1",
125    )
126
127    chain = cast("AudioProcessingChain", streamdetails.audio_processing)
128    assert chain.input_fidelity.quality == AudioQuality.HI_RES
129    assert chain.queue_processing is not None
130    assert chain.outputs[0].player_ids == ["player-1", "player-2"]
131    assert chain.outputs[0].fidelity == AudioFidelity(
132        quality=AudioQuality.HI_RES,
133        bit_perfect=True,
134    )
135    assert chain.outputs[1].player_ids == ["player-3"]
136    assert chain.outputs[1].fidelity == AudioFidelity(
137        quality=AudioQuality.LOW,
138        bit_perfect=False,
139    )
140
141
142def test_lossy_source_can_have_bit_perfect_lossless_output() -> None:
143    """Lossy source quality does not prevent preserving its decoded PCM samples."""
144    manager, _mass, _queue_data, streamdetails, lossless_plan, lossy_plan = _manager_context()
145    streamdetails.audio_format = AudioFormat(
146        content_type=ContentType.OGG,
147        codec_type=ContentType.VORBIS,
148        sample_rate=44100,
149        bit_depth=16,
150        channels=2,
151        bit_rate=320,
152    )
153    pcm_format = _format(ContentType.PCM_S16LE)
154    manager.update_item_runtime(
155        "queue-1",
156        "session-1",
157        "item-1",
158        input_format=pcm_format,
159        pcm_format=pcm_format,
160        normalization=None,
161        playback_speed=1.0,
162    )
163    lossless_plan.input_format = pcm_format
164    lossless_plan.output_details.output_format = _format(ContentType.FLAC)
165    lossy_plan.input_format = pcm_format
166    lossy_plan.output_details.output_format = _format(ContentType.MP3, bit_rate=320)
167
168    manager.update_output(
169        "lossless-player",
170        lossless_plan,
171        queue_id="queue-1",
172        session_id="session-1",
173        queue_item_id="item-1",
174    )
175    manager.update_output(
176        "lossy-player",
177        lossy_plan,
178        queue_id="queue-1",
179        session_id="session-1",
180        queue_item_id="item-1",
181    )
182
183    chain = cast("AudioProcessingChain", streamdetails.audio_processing)
184    outputs = {output.player_ids[0]: output for output in chain.outputs}
185    assert chain.input_fidelity.quality == AudioQuality.STANDARD
186    assert outputs["lossless-player"].fidelity == AudioFidelity(
187        quality=AudioQuality.STANDARD,
188        bit_perfect=True,
189    )
190    assert outputs["lossy-player"].fidelity == AudioFidelity(
191        quality=AudioQuality.STANDARD,
192        bit_perfect=False,
193    )
194    serialized = streamdetails.to_dict()
195    serialized_outputs = {
196        output["player_ids"][0]: output for output in serialized["audio_processing"]["outputs"]
197    }
198    assert serialized_outputs["lossless-player"]["fidelity"]["bit_perfect"] is True
199
200
201def test_shared_output_destinations_are_registered_atomically() -> None:
202    """One shared output publishes all destinations in a single queue update."""
203    manager, mass, _queue_data, streamdetails, output_plan, _lossy_plan = _manager_context()
204    mass.player_queues.signal_update.reset_mock()
205
206    assert manager.update_output(
207        "leader",
208        output_plan,
209        shared_player_ids={"leader", "sync-child"},
210        queue_id="queue-1",
211        session_id="session-1",
212        queue_item_id="item-1",
213    )
214
215    assert streamdetails.audio_processing is not None
216    assert len(streamdetails.audio_processing.outputs) == 1
217    assert streamdetails.audio_processing.outputs[0].player_ids == ["leader", "sync-child"]
218    mass.player_queues.signal_update.assert_called_once_with("queue-1")
219
220    mass.player_queues.signal_update.reset_mock()
221    assert not manager.update_output(
222        "leader",
223        output_plan,
224        shared_player_ids={"sync-child"},
225        queue_id="queue-1",
226        session_id="session-1",
227        queue_item_id="item-1",
228    )
229    mass.player_queues.signal_update.assert_not_called()
230
231
232def test_shared_output_adds_member_without_stream_restart() -> None:
233    """A late native-sync member inherits the active shared output path."""
234    manager, mass, _queue_data, streamdetails, output_plan, _lossy_plan = _manager_context()
235    manager.update_output(
236        "leader",
237        output_plan,
238        shared_player_ids=(),
239        queue_id="queue-1",
240        session_id="session-1",
241        queue_item_id="item-1",
242    )
243    mass.player_queues.signal_update.reset_mock()
244
245    assert manager.retain_outputs("queue-1", {"queue-1", "leader", "late-member"})
246
247    assert streamdetails.audio_processing is not None
248    assert streamdetails.audio_processing.outputs[0].player_ids == ["late-member", "leader"]
249    mass.player_queues.signal_update.assert_called_once_with("queue-1")
250
251
252def test_independent_output_does_not_add_group_member() -> None:
253    """Membership changes do not expand an independently processed output."""
254    manager, mass, _queue_data, streamdetails, output_plan, _lossy_plan = _manager_context()
255    manager.update_output(
256        "leader",
257        output_plan,
258        queue_id="queue-1",
259        session_id="session-1",
260        queue_item_id="item-1",
261    )
262    mass.player_queues.signal_update.reset_mock()
263
264    assert not manager.retain_outputs("queue-1", {"leader", "independent-member"})
265
266    assert streamdetails.audio_processing is not None
267    assert streamdetails.audio_processing.outputs[0].player_ids == ["leader"]
268    mass.player_queues.signal_update.assert_not_called()
269
270
271def test_preset_identity_update_republishes_current_chain() -> None:
272    """Preset updates follow a changed config owner even when output details match."""
273    manager, mass, _queue_data, streamdetails, output_plan, _lossy_plan = _manager_context()
274    output_plan.dsp_config_id = "old-config-player"
275    output_plan.output_details.dsp.preset_id = "night"
276    manager.update_output(
277        "player-1",
278        output_plan,
279        queue_id="queue-1",
280        session_id="session-1",
281        queue_item_id="item-1",
282    )
283    replacement = deepcopy(output_plan)
284    replacement.dsp_config_id = "configured-player"
285    assert manager.update_output(
286        "player-1",
287        replacement,
288        queue_id="queue-1",
289        session_id="session-1",
290        queue_item_id="item-1",
291    )
292    mass.player_queues.signal_update.reset_mock()
293
294    manager.update_player_dsp_preset("configured-player", None)
295
296    assert streamdetails.audio_processing is not None
297    assert streamdetails.audio_processing.outputs[0].dsp.preset_id is None
298    mass.player_queues.signal_update.assert_called_once_with("queue-1")
299
300
301def test_retain_outputs_signals_current_chain_change() -> None:
302    """Pruning a current output publishes the reduced chain."""
303    manager, mass, _queue_data, streamdetails, output_plan, _lossy_plan = _manager_context()
304    manager.update_output(
305        "player-1",
306        output_plan,
307        shared_player_ids={"player-2"},
308        queue_id="queue-1",
309        session_id="session-1",
310        queue_item_id="item-1",
311    )
312    mass.player_queues.signal_update.reset_mock()
313
314    assert manager.retain_outputs("queue-1", {"player-1"})
315
316    assert streamdetails.audio_processing is not None
317    assert streamdetails.audio_processing.outputs[0].player_ids == ["player-1"]
318    mass.player_queues.signal_update.assert_called_once_with("queue-1")
319
320
321def test_prefetched_output_does_not_replace_current_chain() -> None:
322    """An output prepared for the next item does not change the current item."""
323    manager, mass, queue_data, streamdetails, lossless_plan, lossy_plan = _manager_context()
324    next_streamdetails = _streamdetails(item_id="item-2")
325    next_item = SimpleNamespace(queue_item_id="item-2", streamdetails=next_streamdetails)
326    queue_data.items.append(next_item)
327    mass.player_queues.get_item.side_effect = lambda _queue_id, item_id: (
328        next_item if item_id == "item-2" else queue_data.items[0]
329    )
330    manager.update_output(
331        "player-1",
332        lossless_plan,
333        queue_id="queue-1",
334        session_id="session-1",
335        queue_item_id="item-1",
336    )
337    current_chain = streamdetails.audio_processing
338
339    manager.update_item_context(
340        "queue-1",
341        "session-1",
342        "item-2",
343        AudioQueueProcessing(pcm_format=lossless_plan.input_format),
344    )
345    manager.update_output(
346        "player-1",
347        lossy_plan,
348        queue_id="queue-1",
349        session_id="session-1",
350        queue_item_id="item-2",
351    )
352
353    assert streamdetails.audio_processing == current_chain
354    assert next_streamdetails.audio_processing is not None
355    assert next_streamdetails.audio_processing.outputs[0].fidelity.quality == AudioQuality.LOW
356
357
358def test_context_refresh_preserves_runtime_normalization() -> None:
359    """A second consumer does not erase normalization resolved at stream time."""
360    manager, _mass, _queue_data, streamdetails, lossless_plan, _lossy_plan = _manager_context()
361    normalization = AudioNormalizationDetails(
362        mode=VolumeNormalizationMode.DYNAMIC,
363        target_lufs=-17.0,
364    )
365    manager.update_item_runtime(
366        "queue-1",
367        "session-1",
368        "item-1",
369        input_format=lossless_plan.input_format,
370        pcm_format=lossless_plan.input_format,
371        normalization=normalization,
372        playback_speed=1.0,
373    )
374    manager.update_output(
375        "player-1",
376        lossless_plan,
377        queue_id="queue-1",
378        session_id="session-1",
379        queue_item_id="item-1",
380    )
381
382    manager.update_item_context(
383        "queue-1",
384        "session-1",
385        "item-1",
386        AudioQueueProcessing(
387            pcm_format=lossless_plan.input_format,
388            crossfade_mode=CrossfadeMode.SMART_CROSSFADE,
389        ),
390    )
391
392    chain = cast("AudioProcessingChain", streamdetails.audio_processing)
393    assert chain.queue_processing is not None
394    assert chain.queue_processing.normalization == normalization
395    assert chain.outputs[0].fidelity.bit_perfect is False
396
397
398def test_manager_rejects_superseded_and_cleared_sessions() -> None:
399    """Late producers cannot update or recreate a replacement queue session."""
400    manager, mass, queue_data, streamdetails, lossless_plan, _lossy_plan = _manager_context()
401    manager.update_output(
402        "player-1",
403        lossless_plan,
404        queue_id="queue-1",
405        session_id="session-1",
406        queue_item_id="item-1",
407    )
408    assert streamdetails.to_dict()["audio_processing"] is not None
409
410    mass.player_queues.signal_update.reset_mock()
411    manager.start_session("queue-1", "session-2")
412    queue_data.session_id = "session-2"
413    assert streamdetails.to_dict()["audio_processing"] is None
414    mass.player_queues.signal_update.assert_called_once_with("queue-1")
415    assert not manager.update_output(
416        "stale-player",
417        lossless_plan,
418        shared_player_ids={"stale-child"},
419        queue_id="queue-1",
420        session_id="session-1",
421    )
422    assert streamdetails.audio_processing is None
423
424    manager.update_item_context(
425        "queue-1",
426        "session-2",
427        "item-1",
428        AudioQueueProcessing(pcm_format=lossless_plan.input_format),
429    )
430    manager.update_output(
431        "current-player",
432        lossless_plan,
433        queue_id="queue-1",
434        session_id="session-2",
435        queue_item_id="item-1",
436    )
437    assert streamdetails.to_dict()["audio_processing"] is not None
438    mass.player_queues.signal_update.reset_mock()
439    manager.clear("queue-1", "session-2")
440    assert streamdetails.to_dict()["audio_processing"] is None
441    mass.player_queues.signal_update.assert_called_once_with("queue-1")
442    assert not manager.update_output(
443        "late-player",
444        lossless_plan,
445        queue_id="queue-1",
446        session_id="session-2",
447    )
448
449
450def test_manager_prunes_played_item_chains() -> None:
451    """Advancing the queue drops processing state from completed items."""
452    manager, mass, queue_data, streamdetails, lossless_plan, _lossy_plan = _manager_context()
453    manager.update_output(
454        "player-1",
455        lossless_plan,
456        queue_id="queue-1",
457        session_id="session-1",
458        queue_item_id="item-1",
459    )
460    assert streamdetails.audio_processing is not None
461    next_streamdetails = _streamdetails(item_id="item-2")
462    next_item = SimpleNamespace(queue_item_id="item-2", streamdetails=next_streamdetails)
463    queue_data.items.append(next_item)
464    queue_data.queue.current_index = 1
465    queue_data.queue.current_item = next_item
466    mass.player_queues.get_item.side_effect = lambda _queue_id, item_id: (
467        next_item if item_id == "item-2" else queue_data.items[0]
468    )
469
470    manager.update_item_context(
471        "queue-1",
472        "session-1",
473        "item-2",
474        AudioQueueProcessing(pcm_format=lossless_plan.input_format),
475    )
476    manager.update_item_runtime(
477        "queue-1",
478        "session-1",
479        "item-1",
480        input_format=lossless_plan.input_format,
481        pcm_format=lossless_plan.input_format,
482        normalization=None,
483        playback_speed=1.0,
484    )
485
486    assert streamdetails.audio_processing is None
487
488
489def test_hidden_and_intermediate_processing_prevents_bit_perfect_claim() -> None:
490    """Hidden fades and lower-resolution handoffs prevent bit-perfect output."""
491    manager, _mass, _queue_data, streamdetails, output_plan, _lossy_plan = _manager_context(
492        alters_audio=True
493    )
494    output_plan.handoff_format = _format(ContentType.PCM_S24LE, 48000, 24)
495
496    manager.update_output(
497        "player-1",
498        output_plan,
499        queue_id="queue-1",
500        session_id="session-1",
501        queue_item_id="item-1",
502    )
503
504    assert streamdetails.audio_processing is not None
505    assert streamdetails.audio_processing.outputs[0].fidelity.bit_perfect is False
506
507
508def test_player_output_plan_matches_ffmpeg_filters() -> None:
509    """Typed output details describe the FFmpeg filters returned to callers."""
510    mass = MagicMock()
511    mass.players.get_player.return_value = None
512    mass.config.get_player_dsp_config.return_value = DSPConfig(
513        enabled=True,
514        input_gain=-1.0,
515        filters=[ToneControlFilter(enabled=True, bass_level=2.0)],
516        output_gain=-0.5,
517        preset_id="night",
518    )
519    mass.config.get_raw_player_config_value.return_value = "left"
520    audio = StreamsAudio(cast("Any", mass))
521    input_format = _format(ContentType.PCM_F32LE, 96000, 32)
522    output_format = _format(ContentType.FLAC, 48000, 16, channels=1)
523
524    plan = audio.get_player_output_plan(
525        "player-1",
526        input_format,
527        output_format,
528        queue_id="queue-1",
529        session_id="session-1",
530        queue_item_id="item-1",
531    )
532
533    assert plan.filter_params[0] == "volume=-1.0dB"
534    assert plan.filter_params[-1] == "pan=mono|c0=FL"
535    assert plan.output_details.dsp == AudioDSPDetails(
536        state=DSPState.ENABLED,
537        input_gain=-1.0,
538        filters=[ToneControlFilter(enabled=True, bass_level=2.0)],
539        output_gain=-0.5,
540        preset_id="night",
541    )
542    assert plan.dsp_config_id == "player-1"
543    assert plan.output_details.source_channel == AudioChannel.FL
544    assert plan.output_details.output_format == output_format
545    mass.streams.audio_processing.update_output.assert_called_once_with(
546        "player-1",
547        plan,
548        shared_player_ids=None,
549        queue_id="queue-1",
550        session_id="session-1",
551        queue_item_id="item-1",
552    )
553
554
555def test_player_output_plan_downmixes_to_mono() -> None:
556    """The mono output mode folds both source channels into a single channel."""
557    mass = MagicMock()
558    mass.players.get_player.return_value = None
559    mass.config.get_player_dsp_config.return_value = DSPConfig(enabled=False)
560    mass.config.get_raw_player_config_value.return_value = "mono"
561    audio = StreamsAudio(cast("Any", mass))
562    input_format = _format(ContentType.PCM_F32LE, 48000, 32)
563    output_format = _format(ContentType.FLAC, 48000, 16, channels=1)
564
565    plan = audio.get_player_output_plan(
566        "player-1",
567        input_format,
568        output_format,
569        queue_id="queue-1",
570        session_id="session-1",
571        queue_item_id="item-1",
572    )
573
574    assert plan.filter_params == ["pan=mono|c0=0.5*FL+0.5*FR"]
575    assert plan.output_details.source_channel == AudioChannel.ALL
576
577
578def test_player_output_plan_feeds_every_output_channel() -> None:
579    """A stereo output carries the downmix on both channels instead of being upmixed."""
580    mass = MagicMock()
581    mass.players.get_player.return_value = None
582    mass.config.get_player_dsp_config.return_value = DSPConfig(enabled=False)
583    mass.config.get_raw_player_config_value.return_value = "mono"
584    audio = StreamsAudio(cast("Any", mass))
585    audio_format = _format(ContentType.PCM_F32LE, 48000, 32)
586
587    plan = audio.get_player_output_plan(
588        "player-1",
589        audio_format,
590        audio_format,
591        queue_id="queue-1",
592        session_id="session-1",
593        queue_item_id="item-1",
594    )
595
596    assert plan.filter_params == ["pan=stereo|c0=0.5*FL+0.5*FR|c1=0.5*FL+0.5*FR"]
597    assert plan.output_details.source_channel == AudioChannel.ALL
598
599
600def test_player_output_plan_skips_channel_selection_for_mono_source() -> None:
601    """A single channel source has no channels to select, so it is left untouched."""
602    mass = MagicMock()
603    mass.players.get_player.return_value = None
604    mass.config.get_player_dsp_config.return_value = DSPConfig(enabled=False)
605    mass.config.get_raw_player_config_value.return_value = "mono"
606    audio = StreamsAudio(cast("Any", mass))
607    audio_format = _format(ContentType.PCM_F32LE, 48000, 32, channels=1)
608
609    plan = audio.get_player_output_plan(
610        "player-1",
611        audio_format,
612        audio_format,
613        queue_id="queue-1",
614        session_id="session-1",
615        queue_item_id="item-1",
616    )
617
618    assert plan.filter_params == []
619    assert plan.output_details.source_channel is None
620
621
622def test_player_output_plan_pans_for_the_handoff_format() -> None:
623    """The pan follows the format FFmpeg emits, not a later provider side encode."""
624    mass = MagicMock()
625    mass.players.get_player.return_value = None
626    mass.config.get_player_dsp_config.return_value = DSPConfig(enabled=False)
627    mass.config.get_raw_player_config_value.return_value = "mono"
628    audio = StreamsAudio(cast("Any", mass))
629    pcm_format = _format(ContentType.PCM_F32LE, 48000, 32)
630
631    plan = audio.get_player_output_plan(
632        "player-1",
633        pcm_format,
634        _format(ContentType.FLAC, 48000, 16, channels=1),
635        handoff_format=pcm_format,
636        queue_id="queue-1",
637        session_id="session-1",
638        queue_item_id="item-1",
639    )
640
641    assert plan.filter_params == ["pan=stereo|c0=0.5*FL+0.5*FR|c1=0.5*FL+0.5*FR"]
642
643
644def test_mono_downmix_prevents_bit_perfect_claim() -> None:
645    """A mono downmix alters the samples, even when every format stays stereo."""
646    manager, _mass, _queue_data, streamdetails, output_plan, _lossy_plan = _manager_context()
647    output_plan.output_details.source_channel = AudioChannel.ALL
648
649    manager.update_output(
650        "player-1",
651        output_plan,
652        queue_id="queue-1",
653        session_id="session-1",
654        queue_item_id="item-1",
655    )
656
657    assert streamdetails.audio_processing is not None
658    assert streamdetails.audio_processing.outputs[0].fidelity.bit_perfect is False
659
660
661def test_player_output_plan_excludes_neutral_filters() -> None:
662    """A filter that emits no FFmpeg params is left out of the reported chain."""
663    mass = MagicMock()
664    mass.players.get_player.return_value = None
665    mass.config.get_player_dsp_config.return_value = DSPConfig(
666        enabled=True,
667        filters=[ToneControlFilter(enabled=True)],
668    )
669    mass.config.get_raw_player_config_value.return_value = "stereo"
670    audio = StreamsAudio(cast("Any", mass))
671    audio_format = _format(ContentType.PCM_F32LE, 48000, 32)
672
673    plan = audio.get_player_output_plan(
674        "player-1",
675        audio_format,
676        audio_format,
677        queue_id="queue-1",
678        session_id="session-1",
679        queue_item_id="item-1",
680    )
681
682    assert plan.output_details.dsp.filters == []
683    assert not any(
684        isinstance(param, str) and param.startswith("equalizer=") for param in plan.filter_params
685    )
686
687
688def _convolution_plan(known_ir_ids: list[str]) -> AudioOutputPlan:
689    """Build an output plan for a player convolving with impulse response "abc123"."""
690    mass = MagicMock()
691    mass.players.get_player.return_value = None
692    mass.storage_path = "/storage"
693    mass.config.get_player_dsp_config.return_value = DSPConfig(
694        enabled=True,
695        filters=[ConvolutionFilter(enabled=True, ir_id="abc123")],
696    )
697    mass.config.get_dsp_irs.return_value = [{"ir_id": ir_id} for ir_id in known_ir_ids]
698    mass.config.get_raw_player_config_value.return_value = "stereo"
699    audio = StreamsAudio(cast("Any", mass))
700    audio_format = _format(ContentType.PCM_F32LE, 48000, 32)
701    return audio.get_player_output_plan(
702        "player-1",
703        audio_format,
704        audio_format,
705        queue_id="queue-1",
706        session_id="session-1",
707        queue_item_id="item-1",
708    )
709
710
711def test_player_output_plan_drops_convolution_with_unknown_ir() -> None:
712    """An impulse response with no stored record is left out rather than failing ffmpeg."""
713    plan = _convolution_plan(known_ir_ids=["other"])
714
715    assert plan.output_details.dsp.filters == []
716    assert not any(isinstance(param, ComplexFilter) for param in plan.filter_params)
717
718
719def test_player_output_plan_keeps_convolution_with_known_ir() -> None:
720    """An impulse response that is still stored convolves as configured."""
721    plan = _convolution_plan(known_ir_ids=["abc123"])
722
723    assert len(plan.output_details.dsp.filters) == 1
724    complex_filters = [param for param in plan.filter_params if isinstance(param, ComplexFilter)]
725    assert [f.inputs[0].path for f in complex_filters] == ["/storage/dsp_irs/abc123.wav"]
726
727
728def test_player_output_plan_prefers_rendering_player_channels() -> None:
729    """Output channels stored on the rendering player win over the parent's value."""
730    mass = MagicMock()
731    player = MagicMock(player_id="child-1", protocol_parent_id="parent-1")
732    player.state.active_group = None
733    player.state.synced_to = None
734    mass.players.get_player.return_value = player
735    mass.config.get_player_dsp_config.return_value = DSPConfig(enabled=False)
736    mass.config.get_raw_player_config_value.side_effect = lambda player_id, _key, default: (
737        "left" if player_id == "child-1" else default
738    )
739    audio = StreamsAudio(cast("Any", mass))
740    audio_format = _format(ContentType.PCM_F32LE, 48000, 32)
741
742    plan = audio.get_player_output_plan(
743        "child-1",
744        audio_format,
745        audio_format,
746        queue_id="queue-1",
747        session_id="session-1",
748        queue_item_id="item-1",
749    )
750
751    assert plan.output_details.source_channel == AudioChannel.FL
752    assert "pan=stereo|c0=FL|c1=FL" in plan.filter_params
753    # processing attribution still points at the visible parent player
754    assert mass.streams.audio_processing.update_output.call_args.args[0] == "parent-1"
755
756
757@pytest.mark.asyncio
758async def test_output_format_prefers_rendering_player_channels() -> None:
759    """The output format channel count follows the rendering player's own stored value."""
760    mass = MagicMock()
761    player = MagicMock(player_id="child-1", protocol_parent_id="parent-1")
762    player.get_supported_sample_rates.return_value = [(48000, 24)]
763    mass.config.get_raw_player_config_value.side_effect = lambda player_id, _key, default: (
764        "left" if player_id == "child-1" else default
765    )
766    audio = StreamsAudio(cast("Any", mass))
767
768    fmt = await audio.get_output_format("flac", player, 48000, 24, MediaType.TRACK)
769
770    assert fmt.channels == 1
771
772
773@pytest.mark.asyncio
774async def test_single_stream_handler_shares_native_group_members(
775    monkeypatch: pytest.MonkeyPatch,
776) -> None:
777    """The regular single-item HTTP stream registers native group members."""
778    controller, request, group_members = _native_stream_handler_context(monkeypatch)
779
780    with pytest.raises(_OutputPlanRequested):
781        await controller.serve_queue_item_stream(request)
782
783    assert controller.audio.get_player_output_plan.call_args.kwargs["shared_player_ids"] is (
784        group_members
785    )
786
787
788@pytest.mark.asyncio
789async def test_flow_stream_handler_shares_native_group_members(
790    monkeypatch: pytest.MonkeyPatch,
791) -> None:
792    """The regular flow HTTP stream registers native group members."""
793    controller, request, group_members = _native_stream_handler_context(monkeypatch)
794
795    with pytest.raises(_OutputPlanRequested):
796        await controller.serve_queue_flow_stream(request)
797
798    assert controller.audio.get_player_output_plan.call_args.kwargs["shared_player_ids"] is (
799        group_members
800    )
801
802
803def test_protocol_output_uses_parent_settings(monkeypatch: pytest.MonkeyPatch) -> None:
804    """Protocol output details use the user-facing parent configuration."""
805    mass = MagicMock()
806    player = MagicMock(player_id="protocol-1", protocol_parent_id="player-1")
807    player.state.active_group = None
808    player.state.synced_to = None
809    shared_player = MagicMock(player_id="protocol-2", protocol_parent_id="player-2")
810    mass.players.get_player.side_effect = lambda player_id: {
811        "protocol-1": player,
812        "protocol-2": shared_player,
813    }.get(player_id)
814    mass.config.get_player_dsp_config.return_value = DSPConfig(enabled=False)
815    mass.config.get_raw_player_config_value.side_effect = lambda _player_id, _key, default: (
816        False if isinstance(default, bool) else "right"
817    )
818    audio = StreamsAudio(cast("Any", mass))
819    monkeypatch.setattr(
820        audio,
821        "_resolve_player_dsp_config",
822        lambda _player: DSPConfig(preset_id="parent-preset"),
823    )
824    pcm_format = _format(ContentType.PCM_S16LE, 44100, 16)
825
826    plan = audio.get_player_output_plan(
827        "protocol-1",
828        pcm_format,
829        pcm_format,
830        shared_player_ids={"protocol-1", "protocol-2"},
831        queue_id="queue-1",
832        session_id="session-1",
833    )
834
835    assert plan.output_details.player_ids == ["player-1", "player-2"]
836    assert plan.output_details.dsp.preset_id == "parent-preset"
837    assert plan.dsp_config_id == "player-1"
838    assert plan.output_details.source_channel == AudioChannel.FR
839    # the output channels are looked up on the rendering player first (no value
840    # stored there in this scenario), then resolved from the user-facing parent
841    assert {call.args[0] for call in mass.config.get_raw_player_config_value.call_args_list} == {
842        "player-1",
843        "protocol-1",
844    }
845    mass.streams.audio_processing.update_output.assert_called_once_with(
846        "player-1",
847        plan,
848        shared_player_ids={"player-2"},
849        queue_id="queue-1",
850        session_id="session-1",
851        queue_item_id=None,
852    )
853
854
855def test_single_member_group_uses_child_dsp_preset() -> None:
856    """A single-member player group reports the child's effective preset."""
857    mass = MagicMock()
858    player = MagicMock(player_id="group-1", protocol_parent_id=None)
859    player.provider.domain = "player_group"
860    player.state.active_group = None
861    player.state.synced_to = None
862    player.state.group_members = ["child-1"]
863    player.state.supported_features = set()
864    child = MagicMock(player_id="child-1")
865    mass.players.get_player.side_effect = lambda player_id: (
866        player if player_id == "group-1" else child
867    )
868    mass.config.get_player_dsp_config.side_effect = lambda player_id: (
869        DSPConfig(enabled=True, preset_id="child-preset")
870        if player_id == "child-1"
871        else DSPConfig(enabled=False)
872    )
873    mass.config.get_raw_player_config_value.return_value = "stereo"
874    audio = StreamsAudio(cast("Any", mass))
875    pcm_format = _format(ContentType.PCM_F32LE, 48000, 32)
876
877    plan = audio.get_player_output_plan("group-1", pcm_format, pcm_format)
878
879    assert plan.dsp_config_id == "child-1"
880    assert plan.output_details.dsp.state == DSPState.ENABLED
881    assert plan.output_details.dsp.preset_id == "child-preset"
882
883
884def test_unsupported_group_preserves_configured_preset() -> None:
885    """Runtime DSP suppression retains the selected preset identity."""
886    mass = MagicMock()
887    player = MagicMock(player_id="leader-1", protocol_parent_id=None)
888    player.provider.domain = "test"
889    player.state.active_group = None
890    player.state.synced_to = None
891    player.state.group_members = ["child-1"]
892    player.state.supported_features = set()
893    mass.players.get_player.return_value = player
894    mass.config.get_player_dsp_config.side_effect = lambda _player_id: DSPConfig(
895        enabled=True,
896        preset_id="group-preset",
897    )
898    mass.config.get_raw_player_config_value.return_value = "stereo"
899    audio = StreamsAudio(cast("Any", mass))
900    pcm_format = _format(ContentType.PCM_F32LE, 48000, 32)
901
902    plan = audio.get_player_output_plan("leader-1", pcm_format, pcm_format)
903
904    assert plan.dsp_config_id == "leader-1"
905    assert plan.output_details.dsp.state == DSPState.DISABLED_BY_UNSUPPORTED_GROUP
906    assert plan.output_details.dsp.preset_id == "group-preset"
907
908
909@pytest.mark.asyncio
910async def test_stale_flow_generator_does_not_mutate_active_session() -> None:
911    """A deferred flow generator exits before clearing newer session state."""
912    mass = MagicMock()
913    queue_data = SimpleNamespace(session_id="session-2", flow_mode_stream_log=["current"])
914    mass.player_queues.queue_data.return_value = queue_data
915    audio = StreamsAudio(cast("Any", mass))
916    queue = SimpleNamespace(queue_id="queue-1", display_name="Queue", flow_mode=False)
917    stream = audio.get_queue_flow_stream(
918        cast("Any", queue),
919        MagicMock(),
920        _format(ContentType.PCM_F32LE, 48000, 32),
921        session_id="session-1",
922    )
923
924    with pytest.raises(StopAsyncIteration):
925        await anext(stream)
926
927    assert not queue.flow_mode
928    assert queue_data.flow_mode_stream_log == ["current"]
929
930
931def _manager_context(
932    *,
933    alters_audio: bool = False,
934) -> tuple[
935    AudioProcessingManager,
936    MagicMock,
937    SimpleNamespace,
938    StreamDetails,
939    AudioOutputPlan,
940    AudioOutputPlan,
941]:
942    """Return one prepared queue item and two output plan templates."""
943    mass = MagicMock()
944    streamdetails = _streamdetails()
945    queue_item = SimpleNamespace(queue_item_id="item-1", streamdetails=streamdetails)
946    queue = SimpleNamespace(
947        queue_id="queue-1",
948        current_item=queue_item,
949        next_item=None,
950        current_index=0,
951    )
952    queue_data = SimpleNamespace(session_id="session-1", items=[queue_item], queue=queue)
953    mass.player_queues.get.return_value = queue
954    mass.player_queues.get_active_queue.return_value = queue
955    mass.player_queues.get_item.return_value = queue_item
956    mass.player_queues.queue_data_or_none.return_value = queue_data
957    manager = AudioProcessingManager(mass)
958    pcm_format = _format(ContentType.PCM_S24LE, 96000, 24)
959    manager.start_session("queue-1", "session-1")
960    manager.update_item_context(
961        "queue-1",
962        "session-1",
963        "item-1",
964        AudioQueueProcessing(pcm_format=pcm_format),
965        alters_audio=alters_audio,
966    )
967    lossless_plan = AudioOutputPlan(
968        filter_params=[],
969        output_details=AudioOutputDetails(
970            dsp=AudioDSPDetails(state=DSPState.DISABLED),
971            output_format=_format(ContentType.FLAC, 96000, 24),
972        ),
973        input_format=pcm_format,
974    )
975    lossy_plan = AudioOutputPlan(
976        filter_params=[],
977        output_details=AudioOutputDetails(
978            dsp=AudioDSPDetails(state=DSPState.DISABLED),
979            output_format=_format(ContentType.MP3, 48000, 16, bit_rate=128),
980        ),
981        input_format=pcm_format,
982    )
983    return manager, mass, queue_data, streamdetails, lossless_plan, lossy_plan
984
985
986def _streamdetails(item_id: str = "item-1") -> StreamDetails:
987    """Return hi-res lossless stream details."""
988    return StreamDetails(
989        provider="provider",
990        item_id=item_id,
991        audio_format=_format(ContentType.FLAC, 96000, 24, bit_rate=3200),
992        media_type=MediaType.TRACK,
993    )
994
995
996class _OutputPlanRequested(Exception):
997    """Signal that a stream handler reached output planning."""
998
999
1000def _native_stream_handler_context(
1001    monkeypatch: pytest.MonkeyPatch,
1002) -> tuple[Any, MagicMock, list[str]]:
1003    """Return a native HTTP stream handler prepared to stop at output planning."""
1004    mass = MagicMock()
1005    streamdetails = _streamdetails()
1006    queue_item = SimpleNamespace(
1007        queue_id="queue-1",
1008        queue_item_id="item-1",
1009        name="Track",
1010        duration=180,
1011        streamdetails=streamdetails,
1012        media_item=None,
1013        media_type=MediaType.TRACK,
1014        extra_attributes={},
1015        image=None,
1016    )
1017    queue = SimpleNamespace(
1018        queue_id="queue-1",
1019        display_name="Queue",
1020        current_item=queue_item,
1021        crossfade_enabled=False,
1022        overlay_enabled=False,
1023        overlay_source=None,
1024    )
1025    queue_data = SimpleNamespace(session_id="session-1")
1026    mass.player_queues.get.return_value = queue
1027    mass.player_queues.queue_data.return_value = queue_data
1028    mass.player_queues.get_item.return_value = queue_item
1029    mass.config.get_raw_core_config_value.return_value = 8
1030    mass.config.get_raw_player_config_value.return_value = "disabled"
1031
1032    group_members = ["player-1", "player-2"]
1033    player = MagicMock(player_id="player-1", protocol_parent_id=None)
1034    player.state.group_members = group_members
1035    player.state.supported_features = set()
1036    player.state.name = "Player"
1037    player.get_config_value.return_value = "default"
1038    mass.players.get_player.return_value = player
1039
1040    pcm_format = _format(ContentType.PCM_F32LE, 48000, 32)
1041    output_format = _format(ContentType.FLAC, 48000, 24)
1042    audio = MagicMock()
1043    audio.select_pcm_format = AsyncMock(return_value=pcm_format)
1044    audio.select_flow_pcm_format = AsyncMock(return_value=pcm_format)
1045    audio.get_output_format = AsyncMock(return_value=output_format)
1046    audio.get_player_output_plan.side_effect = _OutputPlanRequested
1047
1048    controller = cast("Any", object.__new__(StreamsController))
1049    controller.mass = mass
1050    controller.audio = audio
1051    controller.logger = MagicMock()
1052    controller._log_request = MagicMock()
1053    controller._update_audio_processing_context = MagicMock()
1054    controller._active_output_streams = 0
1055
1056    response = MagicMock()
1057    response.prepare = AsyncMock()
1058    monkeypatch.setattr(
1059        "music_assistant.controllers.streams.controller.web.StreamResponse",
1060        MagicMock(return_value=response),
1061    )
1062    request = MagicMock()
1063    request.method = "GET"
1064    request.headers = {}
1065    request.match_info = {
1066        "queue_id": "queue-1",
1067        "session_id": "session-1",
1068        "queue_item_id": "item-1",
1069        "player_id": "player-1",
1070        "fmt": "flac",
1071    }
1072    return controller, request, group_members
1073