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