/
/
/
1"""Tests for the is_realtime gate across the buffer, holdback, and stream paths."""
2
3from __future__ import annotations
4
5import asyncio
6from collections.abc import AsyncGenerator
7from types import SimpleNamespace
8from typing import Any, cast
9from unittest.mock import AsyncMock, MagicMock
10
11import pytest
12from music_assistant_models.enums import (
13 ContentType,
14 CrossfadeMode,
15 MediaType,
16 PlayerFeature,
17 StreamType,
18 VolumeNormalizationMode,
19)
20from music_assistant_models.errors import QueueEmpty
21from music_assistant_models.media_items import (
22 AudioFormat,
23 AudioSource,
24 ProviderMapping,
25 Radio,
26 Track,
27)
28from music_assistant_models.queue_item import QueueItem
29from music_assistant_models.streamdetails import StreamDetails
30
31from music_assistant.controllers.streams.audio import (
32 MIN_CROSSFADE_DURATION,
33 StreamsAudio,
34 _TailHold,
35)
36from music_assistant.controllers.streams.audio_buffer import AudioBuffer
37from music_assistant.controllers.streams.constants import BufferSize
38from music_assistant.controllers.streams.controller import StreamsController
39from music_assistant.controllers.streams.smart_fades.fades import StandardCrossFade
40from music_assistant.controllers.streams.smart_fades.helpers import SMART_CROSSFADE_DURATION
41
42# Standard test PCM format: 44100Hz, 16-bit, stereo
43TEST_PCM_FORMAT = AudioFormat(
44 content_type=ContentType.PCM_S16LE,
45 sample_rate=44100,
46 bit_depth=16,
47 channels=2,
48)
49
50# One second of silence in the test format
51ONE_SECOND_CHUNK = b"\x00" * TEST_PCM_FORMAT.pcm_sample_size
52
53
54def _make_stream_details(
55 media_type: MediaType,
56 *,
57 is_realtime: bool = False,
58 volume_normalization_mode: VolumeNormalizationMode | None = None,
59 queue_id: str | None = None,
60) -> StreamDetails:
61 """Build minimal stream details for AudioBuffer.get_buffer tests."""
62 return StreamDetails(
63 provider="builtin",
64 item_id="item-1",
65 audio_format=TEST_PCM_FORMAT,
66 media_type=media_type,
67 stream_type=StreamType.HTTP,
68 path="http://example.com/audio.mp3",
69 duration=180,
70 can_seek=True,
71 allow_seek=True,
72 queue_id=queue_id,
73 is_realtime=is_realtime,
74 volume_normalization_mode=volume_normalization_mode,
75 )
76
77
78async def _make_source(num_chunks: int) -> AsyncGenerator[bytes]:
79 """Create an async generator that yields one-second PCM chunks."""
80 for _ in range(num_chunks):
81 yield ONE_SECOND_CHUNK
82
83
84def _make_mass_for_get_buffer(
85 *, queue: Any | None = None
86) -> tuple[MagicMock, list[asyncio.Task[None]], list[float | None]]:
87 """Build a minimal mass stub for AudioBuffer.get_buffer tests."""
88 received_seek_positions: list[float | None] = []
89
90 def _get_media_stream(*_args: Any, **kwargs: Any) -> AsyncGenerator[bytes]:
91 received_seek_positions.append(kwargs.get("seek_position"))
92 return _make_source(1)
93
94 mass = MagicMock()
95 mass.config.get_raw_core_config_value.return_value = BufferSize.BALANCED.value
96 mass.player_queues.get.return_value = queue
97 mass.streams = SimpleNamespace(
98 audio_analysis=SimpleNamespace(start_analysis=AsyncMock(return_value=None)),
99 audio=SimpleNamespace(get_media_stream=_get_media_stream),
100 )
101 scheduled_tasks: list[asyncio.Task[None]] = []
102
103 def _create_task(coro: Any) -> asyncio.Task[None]:
104 task: asyncio.Task[None] = asyncio.ensure_future(coro)
105 scheduled_tasks.append(task)
106 return task
107
108 mass.create_task.side_effect = _create_task
109 return mass, scheduled_tasks, received_seek_positions
110
111
112def _streamdetails_for_crossfade(
113 audio_buffer: AudioBuffer | None, *, is_realtime: bool = False
114) -> StreamDetails:
115 """Build incoming track details with an optional prepared buffer."""
116 streamdetails = StreamDetails(
117 provider="test--1",
118 item_id="track-1",
119 audio_format=AudioFormat(content_type=ContentType.FLAC),
120 media_type=MediaType.TRACK,
121 stream_type=StreamType.HTTP,
122 path="http://test.invalid/track.flac",
123 duration=180,
124 is_realtime=is_realtime,
125 )
126 streamdetails.buffer = audio_buffer
127 return streamdetails
128
129
130async def _empty_mix(*_args: object, **_kwargs: object) -> AsyncGenerator[bytes]:
131 """Stand in for the mixer, producing no audio."""
132 no_audio: tuple[bytes, ...] = ()
133 for chunk in no_audio:
134 yield chunk
135
136
137def _buffer(duration_available: float, ready: bool, eof: bool = False) -> AudioBuffer:
138 """Build a valid buffer with the requested resident duration."""
139 audio_buffer = MagicMock(spec=AudioBuffer)
140 audio_buffer.has_error = False
141 audio_buffer.is_valid.return_value = True
142 audio_buffer.duration_available = duration_available
143 audio_buffer.eof = eof
144 audio_buffer.ready = MagicMock()
145 audio_buffer.ready.is_set.return_value = ready
146 return audio_buffer
147
148
149def _stream_details_provider(streamdetails: StreamDetails) -> StreamsAudio:
150 """Build a StreamsAudio whose single provider hands back the given streamdetails."""
151 provider = MagicMock()
152 provider.instance_id = "test--1"
153 provider.domain = "test"
154 provider.available = True
155 provider.is_streaming_provider = True
156 provider.get_stream_details = AsyncMock(return_value=streamdetails)
157 mass = MagicMock()
158 mass.get_provider.side_effect = lambda instance, **_kwargs: (
159 provider if instance == "test--1" else None
160 )
161 mass.providers = []
162 mass.player_queues.queue_data_or_none.return_value = None
163 mass.streams.get_config_value.return_value = -17
164 return StreamsAudio(mass)
165
166
167def _queue_item_with_mapping(media_item_cls: type) -> QueueItem:
168 """Build a queue item whose media item carries one matching provider mapping."""
169 mapping = ProviderMapping(item_id="item-1", provider_domain="test", provider_instance="test--1")
170 media_item = media_item_cls(
171 item_id="item-1", provider="test--1", name="Item", provider_mappings={mapping}
172 )
173 return QueueItem(
174 queue_id="q1", queue_item_id="qi1", name="Item", duration=None, media_item=media_item
175 )
176
177
178# -- AudioBuffer.get_buffer: ready threshold ladder --
179
180
181@pytest.mark.parametrize(
182 (
183 "is_realtime",
184 "crossfade_enabled",
185 "normalization_mode",
186 "media_type",
187 "expected_threshold",
188 ),
189 [
190 pytest.param(True, False, None, MediaType.RADIO, 1, id="realtime_base"),
191 pytest.param(True, False, None, MediaType.AUDIO_SOURCE, 1, id="realtime_audio_source"),
192 # the queue's crossfade setting buys nothing for a realtime source: its fade
193 # streams in as it arrives, so a second of audio here would only be a second
194 # of extra startup delay
195 pytest.param(True, True, None, MediaType.TRACK, 1, id="realtime_crossfade"),
196 pytest.param(
197 True,
198 False,
199 VolumeNormalizationMode.DYNAMIC,
200 MediaType.TRACK,
201 2,
202 id="realtime_dynamic_normalization",
203 ),
204 pytest.param(False, True, None, MediaType.TRACK, 8, id="non_realtime_crossfade"),
205 pytest.param(
206 False,
207 False,
208 VolumeNormalizationMode.DYNAMIC,
209 MediaType.RADIO,
210 3,
211 id="non_realtime_dynamic_radio",
212 ),
213 pytest.param(
214 False,
215 False,
216 VolumeNormalizationMode.DYNAMIC,
217 MediaType.TRACK,
218 5,
219 id="non_realtime_dynamic_track",
220 ),
221 pytest.param(False, False, None, MediaType.TRACK, 2, id="non_realtime_default"),
222 ],
223)
224async def test_ready_threshold_ladder(
225 is_realtime: bool,
226 crossfade_enabled: bool,
227 normalization_mode: VolumeNormalizationMode | None,
228 media_type: MediaType,
229 expected_threshold: int,
230) -> None:
231 """The buffered-ready threshold follows the realtime ladder, leaving the old one intact."""
232 # a realtime source is only ever raised above the floor by dynamic normalization,
233 # which genuinely needs its lookahead
234 queue = SimpleNamespace(crossfade_enabled=crossfade_enabled)
235 mass, scheduled_tasks, _seek_positions = _make_mass_for_get_buffer(queue=queue)
236 streamdetails = _make_stream_details(
237 media_type,
238 is_realtime=is_realtime,
239 volume_normalization_mode=normalization_mode,
240 queue_id="queue-1",
241 )
242
243 buffer = await AudioBuffer.get_buffer(mass, streamdetails, reason="test")
244
245 assert buffer._ready_threshold == expected_threshold
246 await asyncio.gather(*scheduled_tasks)
247 await buffer.clear()
248
249
250# -- AudioBuffer.get_buffer: seek handling --
251
252
253@pytest.mark.parametrize(
254 ("is_realtime", "seek_seconds", "expected_source_seek"),
255 [
256 pytest.param(True, 30, 30, id="realtime_short_seek_reaches_source"),
257 pytest.param(False, 30, 0, id="non_realtime_short_seek_buffers_from_start"),
258 pytest.param(False, 90, 90, id="non_realtime_long_seek_reaches_source"),
259 ],
260)
261async def test_get_buffer_seek_position_reaches_the_source(
262 is_realtime: bool, seek_seconds: int, expected_source_seek: int
263) -> None:
264 """A realtime source always seeks at the source; a non-realtime one only for a large seek."""
265 mass, scheduled_tasks, received_seek_positions = _make_mass_for_get_buffer()
266 streamdetails = _make_stream_details(MediaType.TRACK, is_realtime=is_realtime)
267
268 buffer = await AudioBuffer.get_buffer(
269 mass, streamdetails, seek_position_ms=seek_seconds * 1000, reason="test"
270 )
271
272 assert received_seek_positions == [expected_source_seek]
273 assert buffer._discarded_chunks == expected_source_seek
274 await asyncio.gather(*scheduled_tasks)
275 await buffer.clear()
276
277
278# -- AudioBuffer.eof --
279
280
281async def test_eof_reflects_producer_completion() -> None:
282 """The eof flag turns True only once the producer has delivered everything."""
283 buf = AudioBuffer(TEST_PCM_FORMAT)
284 assert not buf.eof
285 await buf._put(ONE_SECOND_CHUNK)
286 assert not buf.eof
287 await buf._set_eof()
288 assert buf.eof
289
290
291# -- _TailHold --
292
293
294async def test_tail_hold_grows_with_the_banked_surplus() -> None:
295 """The holdback takes half of what arrived beyond the wall clock plus a reserve."""
296 pcm_format = TEST_PCM_FORMAT
297 frame_size = (pcm_format.bit_depth // 8) * pcm_format.channels
298 audio_buffer = SimpleNamespace(eof=False, has_error=False, duration_available=2.0)
299 queue_item = SimpleNamespace(streamdetails=SimpleNamespace(buffer=audio_buffer))
300 hold = _TailHold(pcm_format, cast("Any", queue_item))
301
302 # nothing arrived yet: nothing may be held
303 assert hold.hold_target(8 * pcm_format.pcm_sample_size, frame_size) == 0
304
305 # 27s arrived in ~4s of wall time: 27 - 4 - 3 (reserve) = 20s is spare, half
306 # of which may be held (the rest keeps growing the player's lead)
307 hold.note_bytes(27 * pcm_format.pcm_sample_size)
308 hold._started = asyncio.get_event_loop().time() - 4.0
309 target = hold.hold_target(8 * pcm_format.pcm_sample_size, frame_size)
310 assert target == 8 * pcm_format.pcm_sample_size
311 larger = hold.hold_target(45 * pcm_format.pcm_sample_size, frame_size)
312 assert larger % frame_size == 0
313 assert int(9.5 * pcm_format.pcm_sample_size) < larger <= 10 * pcm_format.pcm_sample_size
314
315 # barely above realtime: within the reserve nothing may be held at all
316 fresh = _TailHold(pcm_format, cast("Any", queue_item))
317 fresh.note_bytes(6 * pcm_format.pcm_sample_size)
318 fresh._started = asyncio.get_event_loop().time() - 4.0
319 assert fresh.hold_target(8 * pcm_format.pcm_sample_size, frame_size) == 0
320
321 # once the source is done, the rest is resident: full window regardless
322 audio_buffer.eof = True
323 assert (
324 hold.hold_target(45 * pcm_format.pcm_sample_size, frame_size)
325 == 45 * pcm_format.pcm_sample_size
326 )
327
328
329async def test_tail_hold_sees_a_buffer_attached_after_it_was_created() -> None:
330 """Opening the stream is what creates the buffer, so its EOF must still be seen."""
331 pcm_format = TEST_PCM_FORMAT
332 frame_size = (pcm_format.bit_depth // 8) * pcm_format.channels
333 # the tracker is built before the stream is opened, so there is no buffer yet
334 streamdetails = SimpleNamespace(buffer=None)
335 hold = _TailHold(pcm_format, cast("Any", SimpleNamespace(streamdetails=streamdetails)))
336 hold.note_bytes(pcm_format.pcm_sample_size)
337 hold._started = asyncio.get_event_loop().time()
338
339 # a source that finished delivering releases the full window
340 streamdetails.buffer = SimpleNamespace(eof=True, has_error=False)
341
342 assert (
343 hold.hold_target(45 * pcm_format.pcm_sample_size, frame_size)
344 == 45 * pcm_format.pcm_sample_size
345 )
346
347
348async def test_tail_hold_counts_a_long_mix_as_listening_time() -> None:
349 """Bytes noted across a long overlap must not read as a suspension and bank a surplus."""
350 pcm_format = TEST_PCM_FORMAT
351 frame_size = (pcm_format.bit_depth // 8) * pcm_format.channels
352 queue_item = SimpleNamespace(
353 streamdetails=SimpleNamespace(buffer=SimpleNamespace(eof=False, has_error=False))
354 )
355 hold = _TailHold(pcm_format, cast("Any", queue_item))
356
357 # 20s of audio arrives over 20s of wall clock: the source is keeping pace, so
358 # there is no surplus to hold back
359 hold.note_bytes(pcm_format.pcm_sample_size)
360 now = asyncio.get_event_loop().time()
361 hold._started = now - 20.0
362 hold._last_noted = now
363 hold._received_bytes = 20 * pcm_format.pcm_sample_size
364
365 assert hold.hold_target(45 * pcm_format.pcm_sample_size, frame_size) == 0
366
367
368async def test_tail_hold_follows_a_capacity_reselection() -> None:
369 """A reselection hands the item different details; the tracker must follow them."""
370 pcm_format = TEST_PCM_FORMAT
371 frame_size = (pcm_format.bit_depth // 8) * pcm_format.channels
372 queue_item = SimpleNamespace(streamdetails=SimpleNamespace(buffer=None))
373 hold = _TailHold(pcm_format, cast("Any", queue_item))
374 hold.note_bytes(pcm_format.pcm_sample_size)
375 hold._started = asyncio.get_event_loop().time()
376
377 # the source was reselected: the item carries a different streamdetails now
378 queue_item.streamdetails = SimpleNamespace(buffer=SimpleNamespace(eof=True, has_error=False))
379
380 assert (
381 hold.hold_target(45 * pcm_format.pcm_sample_size, frame_size)
382 == 45 * pcm_format.pcm_sample_size
383 )
384
385
386async def test_tail_hold_releases_everything_for_a_failed_source() -> None:
387 """A failed source is skipped without a fade, so its remaining audio is played out."""
388 pcm_format = TEST_PCM_FORMAT
389 frame_size = (pcm_format.bit_depth // 8) * pcm_format.channels
390 audio_buffer = SimpleNamespace(eof=True, has_error=True, duration_available=30.0)
391 hold = _TailHold(
392 pcm_format, cast("Any", SimpleNamespace(streamdetails=SimpleNamespace(buffer=audio_buffer)))
393 )
394 hold.note_bytes(27 * pcm_format.pcm_sample_size)
395 hold._started = asyncio.get_event_loop().time() - 4.0
396
397 assert hold.hold_target(8 * pcm_format.pcm_sample_size, frame_size) == 0
398
399
400async def test_tail_hold_forgives_a_suspended_source() -> None:
401 """A pause is not elapsed listening, so it does not erase the banked surplus."""
402 pcm_format = TEST_PCM_FORMAT
403 frame_size = (pcm_format.bit_depth // 8) * pcm_format.channels
404 audio_buffer = SimpleNamespace(eof=False, has_error=False, duration_available=2.0)
405 hold = _TailHold(
406 pcm_format, cast("Any", SimpleNamespace(streamdetails=SimpleNamespace(buffer=audio_buffer)))
407 )
408
409 hold.note_bytes(27 * pcm_format.pcm_sample_size)
410 hold._started = asyncio.get_event_loop().time() - 4.0
411 # the source went quiet for a while, then resumed
412 hold._last_noted = asyncio.get_event_loop().time() - 30.0
413 hold.note_bytes(pcm_format.pcm_sample_size)
414
415 assert hold.hold_target(8 * pcm_format.pcm_sample_size, frame_size) > 0
416
417
418async def test_tail_hold_works_without_a_source_buffer() -> None:
419 """A source without a buffer still banks a holdback out of what it delivered."""
420 pcm_format = TEST_PCM_FORMAT
421 frame_size = (pcm_format.bit_depth // 8) * pcm_format.channels
422 hold = _TailHold(
423 pcm_format, cast("Any", SimpleNamespace(streamdetails=SimpleNamespace(buffer=None)))
424 )
425
426 hold.note_bytes(27 * pcm_format.pcm_sample_size)
427 hold._started = asyncio.get_event_loop().time() - 4.0
428
429 assert hold.hold_target(8 * pcm_format.pcm_sample_size, frame_size) > 0
430
431
432# -- StreamsAudio._select_buffered_crossfade --
433
434
435def test_the_held_tail_sizes_the_fade_the_configured_mode_picks() -> None:
436 """The mode decides which fade is applied; the held tail only sizes its window."""
437 audio = StreamsAudio(MagicMock())
438
439 # a realtime source barely delivers, yet the tail it banked carries the window:
440 # the incoming side streams in while the blend plays
441 mode, duration = audio._select_buffered_crossfade(
442 _streamdetails_for_crossfade(_buffer(2, ready=True), is_realtime=True),
443 CrossfadeMode.SMART_CROSSFADE,
444 standard_crossfade_duration=8,
445 fade_out_seconds=20,
446 )
447 assert (mode, duration) == (CrossfadeMode.SMART_CROSSFADE, 20)
448
449 # a shorter tail keeps the smart fade, on a shorter window
450 mode, duration = audio._select_buffered_crossfade(
451 _streamdetails_for_crossfade(_buffer(2, ready=True), is_realtime=True),
452 CrossfadeMode.SMART_CROSSFADE,
453 standard_crossfade_duration=8,
454 fade_out_seconds=6,
455 )
456 assert (mode, duration) == (CrossfadeMode.SMART_CROSSFADE, 6)
457
458 # a standard fade never exceeds the configured overlap
459 mode, duration = audio._select_buffered_crossfade(
460 _streamdetails_for_crossfade(_buffer(2, ready=True), is_realtime=True),
461 CrossfadeMode.STANDARD_CROSSFADE,
462 standard_crossfade_duration=8,
463 fade_out_seconds=20,
464 )
465 assert (mode, duration) == (CrossfadeMode.STANDARD_CROSSFADE, 8)
466
467
468def test_a_finished_incoming_source_caps_the_window_at_what_it_holds() -> None:
469 """A source that already ended has no more audio than what is resident."""
470 audio = StreamsAudio(MagicMock())
471
472 mode, duration = audio._select_buffered_crossfade(
473 _streamdetails_for_crossfade(_buffer(6, ready=True, eof=True), is_realtime=True),
474 CrossfadeMode.SMART_CROSSFADE,
475 standard_crossfade_duration=8,
476 fade_out_seconds=45,
477 )
478
479 assert (mode, duration) == (CrossfadeMode.SMART_CROSSFADE, 6)
480
481
482def test_a_short_incoming_track_caps_the_window() -> None:
483 """A long tail cannot claim more overlap than the next track can supply."""
484 audio = StreamsAudio(MagicMock())
485 streamdetails = _streamdetails_for_crossfade(_buffer(2, ready=True), is_realtime=True)
486 streamdetails.duration = 20
487
488 mode, duration = audio._select_buffered_crossfade(
489 streamdetails,
490 CrossfadeMode.SMART_CROSSFADE,
491 standard_crossfade_duration=8,
492 fade_out_seconds=45,
493 )
494
495 assert (mode, duration) == (CrossfadeMode.SMART_CROSSFADE, 10)
496
497
498def test_a_tail_too_short_to_blend_skips_the_fade() -> None:
499 """Below the minimum overlap the tail plays out and the boundary is a hard cut."""
500 audio = StreamsAudio(MagicMock())
501
502 mode, duration = audio._select_buffered_crossfade(
503 _streamdetails_for_crossfade(_buffer(20, ready=True), is_realtime=True),
504 CrossfadeMode.SMART_CROSSFADE,
505 standard_crossfade_duration=8,
506 fade_out_seconds=MIN_CROSSFADE_DURATION - 0.5,
507 )
508
509 assert mode == CrossfadeMode.DISABLED
510 assert duration == 0
511
512
513def test_realtime_incoming_source_not_yet_delivering_skips_the_fade() -> None:
514 """A realtime source whose buffer is not ready yet means the boundary plays clean."""
515 audio = StreamsAudio(MagicMock())
516
517 mode, duration = audio._select_buffered_crossfade(
518 _streamdetails_for_crossfade(_buffer(0, ready=False), is_realtime=True),
519 CrossfadeMode.STANDARD_CROSSFADE,
520 standard_crossfade_duration=8,
521 fade_out_seconds=8,
522 )
523
524 assert mode == CrossfadeMode.DISABLED
525 assert duration == 0
526
527
528# -- Path level: get_queue_item_stream_with_smartfade --
529
530
531async def test_smartfade_realtime_current_item_fades_once_its_source_is_done(
532 monkeypatch: pytest.MonkeyPatch,
533) -> None:
534 """A realtime item whose source finished delivering holds its tail and fades."""
535 pcm_format = AudioFormat(
536 content_type=ContentType.PCM_S16LE,
537 sample_rate=8000,
538 bit_depth=16,
539 channels=2,
540 )
541 # the source is done delivering, which is what arms the realtime holdback
542 current_details = SimpleNamespace(
543 duration=16,
544 seek_position=0,
545 seconds_streamed=0,
546 uri="test://current",
547 buffer=SimpleNamespace(
548 eof=True, cancelled=False, has_error=False, max_size_seconds=300, duration_available=0.0
549 ),
550 is_realtime=True,
551 )
552 next_details = SimpleNamespace(
553 audio_format=pcm_format,
554 buffer=_buffer(SMART_CROSSFADE_DURATION, ready=True),
555 duration=16,
556 seek_position=0,
557 uri="test://next",
558 is_realtime=False,
559 volume_normalization_mode=None,
560 )
561 current_item = SimpleNamespace(
562 queue_id="queue-1",
563 queue_item_id="current",
564 name="Current",
565 streamdetails=current_details,
566 extra_attributes={},
567 )
568 next_item = SimpleNamespace(
569 queue_id="queue-1",
570 queue_item_id="next",
571 name="Next",
572 streamdetails=next_details,
573 extra_attributes={},
574 available=True,
575 )
576 queue = SimpleNamespace(
577 queue_id="queue-1",
578 display_name="Queue",
579 index_in_buffer=0,
580 )
581 player = SimpleNamespace(player_id="player-1", name="Player")
582 mass = MagicMock()
583 mass.player_queues.get.return_value = queue
584 mass.player_queues.load_next_queue_item = AsyncMock(return_value=next_item)
585 mass.player_queues.index_by_id.return_value = 1
586 audio = StreamsAudio(cast("Any", mass))
587 audio.setup()
588 audio.select_pcm_format = AsyncMock(return_value=pcm_format) # type: ignore[method-assign]
589 audio.crossfade_allowed = MagicMock(return_value=True) # type: ignore[method-assign]
590 build = AsyncMock(
591 return_value=SimpleNamespace(
592 timing_info=SimpleNamespace(
593 fadein_trimmed_duration=0.0,
594 crossfade_duration=8.0,
595 pre_crossfade_duration=0.0,
596 )
597 )
598 )
599 monkeypatch.setattr(audio.smart_fades_mixer, "build", build)
600
601 async def _concat_mix(
602 _smart_fade: object,
603 *,
604 fade_in_part: AsyncGenerator[bytes],
605 fade_out_part: bytes,
606 **_kwargs: object,
607 ) -> AsyncGenerator[bytes]:
608 yield fade_out_part
609 async for fade_in_chunk in fade_in_part:
610 yield fade_in_chunk
611
612 monkeypatch.setattr(audio.smart_fades_mixer, "mix", _concat_mix)
613
614 async def _item_stream(
615 _queue_item: object,
616 *_args: object,
617 **_kwargs: object,
618 ) -> AsyncGenerator[bytes]:
619 yield bytes(pcm_format.pcm_sample_size * 8)
620 yield bytes(pcm_format.pcm_sample_size * 8)
621
622 monkeypatch.setattr(audio, "get_queue_item_stream", _item_stream)
623 stream = audio.get_queue_item_stream_with_smartfade(
624 cast("Any", player),
625 cast("Any", current_item),
626 pcm_format,
627 crossfade_mode=CrossfadeMode.STANDARD_CROSSFADE,
628 standard_crossfade_duration=8,
629 )
630
631 output = b"".join([chunk async for chunk in stream])
632
633 # 8s warmup + 8s of mix output (pre+overlap); the incoming share of the mix
634 # is buffered as crossfade data for the next item's own stream
635 assert len(output) == pcm_format.pcm_sample_size * 16
636 build.assert_awaited_once()
637 crossfade_data = audio._crossfade_data.get("queue-1")
638 assert crossfade_data is not None
639 assert crossfade_data.queue_item_id == "next"
640
641
642async def test_smartfade_still_filling_source_fades_from_what_it_banked(
643 monkeypatch: pytest.MonkeyPatch,
644) -> None:
645 """A source still delivering fades from the audio it banked ahead of playback."""
646 pcm_format = AudioFormat(
647 content_type=ContentType.PCM_S16LE,
648 sample_rate=8000,
649 bit_depth=16,
650 channels=2,
651 )
652 current_details = SimpleNamespace(
653 duration=16,
654 seek_position=0,
655 seconds_streamed=0,
656 uri="test://current",
657 buffer=SimpleNamespace(eof=False, cancelled=False, has_error=False, max_size_seconds=300),
658 is_realtime=False,
659 )
660 next_details = SimpleNamespace(
661 audio_format=pcm_format,
662 buffer=_buffer(SMART_CROSSFADE_DURATION, ready=True),
663 duration=16,
664 seek_position=0,
665 uri="test://next",
666 volume_normalization_mode=None,
667 is_realtime=False,
668 )
669 current_item = SimpleNamespace(
670 queue_id="queue-1",
671 queue_item_id="current",
672 name="Current",
673 streamdetails=current_details,
674 extra_attributes={},
675 )
676 next_item = SimpleNamespace(
677 queue_id="queue-1",
678 queue_item_id="next",
679 name="Next",
680 streamdetails=next_details,
681 extra_attributes={},
682 available=True,
683 )
684 queue = SimpleNamespace(
685 queue_id="queue-1",
686 display_name="Queue",
687 index_in_buffer=0,
688 )
689 player = SimpleNamespace(player_id="player-1", name="Player")
690 mass = MagicMock()
691 mass.player_queues.get.return_value = queue
692 mass.player_queues.load_next_queue_item = AsyncMock(return_value=next_item)
693 mass.player_queues.index_by_id.return_value = 1
694 audio = StreamsAudio(cast("Any", mass))
695 audio.setup()
696 audio.select_pcm_format = AsyncMock(return_value=pcm_format) # type: ignore[method-assign]
697 audio.crossfade_allowed = MagicMock(return_value=True) # type: ignore[method-assign]
698 build = AsyncMock(
699 return_value=SimpleNamespace(
700 timing_info=SimpleNamespace(
701 fadein_trimmed_duration=0.0,
702 crossfade_duration=8.0,
703 pre_crossfade_duration=0.0,
704 )
705 )
706 )
707 monkeypatch.setattr(audio.smart_fades_mixer, "build", build)
708
709 async def _concat_mix(
710 _smart_fade: object,
711 *,
712 fade_in_part: AsyncGenerator[bytes],
713 fade_out_part: bytes,
714 **_kwargs: object,
715 ) -> AsyncGenerator[bytes]:
716 yield fade_out_part
717 async for fade_in_chunk in fade_in_part:
718 yield fade_in_chunk
719
720 monkeypatch.setattr(audio.smart_fades_mixer, "mix", _concat_mix)
721
722 async def _item_stream(
723 _queue_item: object,
724 *_args: object,
725 **_kwargs: object,
726 ) -> AsyncGenerator[bytes]:
727 yield bytes(pcm_format.pcm_sample_size * 8)
728 yield bytes(pcm_format.pcm_sample_size * 8)
729
730 monkeypatch.setattr(audio, "get_queue_item_stream", _item_stream)
731 stream = audio.get_queue_item_stream_with_smartfade(
732 cast("Any", player),
733 cast("Any", current_item),
734 pcm_format,
735 crossfade_mode=CrossfadeMode.STANDARD_CROSSFADE,
736 standard_crossfade_duration=8,
737 )
738
739 output = b"".join([chunk async for chunk in stream])
740
741 # how much tail the holdback banked depends on the wall clock, so only the
742 # invariants are asserted: the source's own audio is all there, and it faded
743 assert len(output) >= pcm_format.pcm_sample_size * 16
744 build.assert_awaited_once()
745 crossfade_data = audio._crossfade_data.get("queue-1")
746 assert crossfade_data is not None
747 assert crossfade_data.queue_item_id == "next"
748
749
750# -- Path level: get_queue_flow_stream --
751
752
753async def test_flow_realtime_item_yields_all_audio_as_plain_concatenation(
754 monkeypatch: pytest.MonkeyPatch,
755) -> None:
756 """A realtime item's flow audio is passed straight through and simply concatenated."""
757 pcm_format = AudioFormat(
758 content_type=ContentType.PCM_S16LE,
759 sample_rate=8000,
760 bit_depth=16,
761 channels=2,
762 )
763 # the source is done delivering, so only the realtime flag can deny the holdback
764 realtime_details = SimpleNamespace(
765 audio_format=pcm_format,
766 buffer=SimpleNamespace(eof=True, cancelled=False, has_error=False, max_size_seconds=300),
767 fade_in=False,
768 stream_error=False,
769 uri="test://realtime",
770 seek_position=0,
771 seconds_streamed=0,
772 duration=20,
773 is_realtime=True,
774 )
775 realtime_item = SimpleNamespace(
776 queue_id="queue-1",
777 queue_item_id="item-1",
778 name="Realtime",
779 media_type=MediaType.TRACK,
780 media_item=None,
781 streamdetails=realtime_details,
782 duration=20,
783 extra_attributes={},
784 )
785 next_details = SimpleNamespace(
786 audio_format=pcm_format,
787 buffer=None,
788 fade_in=False,
789 stream_error=False,
790 uri="test://next",
791 seek_position=0,
792 seconds_streamed=0,
793 duration=20,
794 is_realtime=False,
795 )
796 next_item = SimpleNamespace(
797 queue_id="queue-1",
798 queue_item_id="item-2",
799 name="Next",
800 media_type=MediaType.TRACK,
801 media_item=None,
802 streamdetails=next_details,
803 duration=20,
804 extra_attributes={},
805 )
806 queue = SimpleNamespace(
807 queue_id="queue-1",
808 display_name="Queue",
809 flow_mode=False,
810 overlay_enabled=False,
811 overlay_source=None,
812 )
813 queue_data = SimpleNamespace(session_id="session-1", flow_mode_stream_log=[])
814 mass = MagicMock()
815 mass.player_queues.queue_data.return_value = queue_data
816 mass.player_queues.load_next_queue_item = AsyncMock(side_effect=[next_item, QueueEmpty])
817 mass.player_queues.get.return_value = queue
818 mass.streams.get_crossfade_mode.return_value = CrossfadeMode.STANDARD_CROSSFADE
819 mass.config.get_raw_core_config_value.return_value = 8
820 mass.streams.audio_processing.update_item_context = MagicMock()
821 mass.player_queues.queue_buffer_completed = MagicMock()
822 player = MagicMock()
823 player.config.get_value.return_value = "fixed_48000"
824 player.get_supported_sample_rates.return_value = []
825 mass.players.get_player.return_value = player
826 audio = StreamsAudio(cast("Any", mass))
827 audio.setup()
828 build = AsyncMock()
829 monkeypatch.setattr(audio.smart_fades_mixer, "build", build)
830
831 realtime_chunks = [
832 bytes(pcm_format.pcm_sample_size * 8),
833 bytes(pcm_format.pcm_sample_size * 8),
834 ]
835 next_chunks = [bytes(pcm_format.pcm_sample_size * 2)]
836
837 async def _item_stream(
838 queue_item: SimpleNamespace, *_args: object, **_kwargs: object
839 ) -> AsyncGenerator[bytes]:
840 chunks = realtime_chunks if queue_item is realtime_item else next_chunks
841 for chunk in chunks:
842 yield chunk
843
844 monkeypatch.setattr(audio, "get_queue_item_stream", _item_stream)
845 select_crossfade = MagicMock(wraps=audio._select_buffered_crossfade)
846 monkeypatch.setattr(audio, "_select_buffered_crossfade", select_crossfade)
847 stream = audio.get_queue_flow_stream(
848 cast("Any", queue), cast("Any", realtime_item), pcm_format, session_id="session-1"
849 )
850
851 output = b"".join([chunk async for chunk in stream])
852
853 assert output == b"".join(realtime_chunks) + b"".join(next_chunks)
854 build.assert_not_awaited()
855 # no tail was held back, so the next item is never asked to fade into anything
856 select_crossfade.assert_not_called()
857 mass.player_queues.queue_buffer_completed.assert_called_once()
858
859
860async def test_smartfade_unaligned_chunks_still_crossfade(
861 monkeypatch: pytest.MonkeyPatch,
862) -> None:
863 """A source whose chunks are not whole seconds still collects a complete fade tail."""
864 pcm_format = AudioFormat(
865 content_type=ContentType.PCM_S16LE,
866 sample_rate=8000,
867 bit_depth=16,
868 channels=2,
869 )
870 current_details = SimpleNamespace(
871 duration=30,
872 seek_position=0,
873 seconds_streamed=0,
874 uri="test://current",
875 buffer=SimpleNamespace(eof=True, cancelled=False, has_error=False, max_size_seconds=300),
876 is_realtime=False,
877 )
878 next_details = SimpleNamespace(
879 audio_format=pcm_format,
880 buffer=_buffer(SMART_CROSSFADE_DURATION, ready=True),
881 duration=30,
882 seek_position=0,
883 uri="test://next",
884 is_realtime=False,
885 volume_normalization_mode=None,
886 )
887 current_item = SimpleNamespace(
888 queue_id="queue-1",
889 queue_item_id="current",
890 name="Current",
891 streamdetails=current_details,
892 extra_attributes={},
893 )
894 next_item = SimpleNamespace(
895 queue_id="queue-1",
896 queue_item_id="next",
897 name="Next",
898 streamdetails=next_details,
899 extra_attributes={},
900 available=True,
901 )
902 queue = SimpleNamespace(queue_id="queue-1", display_name="Queue", index_in_buffer=0)
903 player = SimpleNamespace(player_id="player-1", name="Player")
904 mass = MagicMock()
905 mass.player_queues.get.return_value = queue
906 mass.player_queues.load_next_queue_item = AsyncMock(return_value=next_item)
907 mass.player_queues.index_by_id.return_value = 1
908 audio = StreamsAudio(cast("Any", mass))
909 audio.setup()
910 audio.select_pcm_format = AsyncMock(return_value=pcm_format) # type: ignore[method-assign]
911 audio.crossfade_allowed = MagicMock(return_value=True) # type: ignore[method-assign]
912 build = AsyncMock(
913 return_value=SimpleNamespace(
914 timing_info=SimpleNamespace(
915 pre_crossfade_duration=2,
916 crossfade_duration=6,
917 fadein_trimmed_duration=0,
918 )
919 )
920 )
921 monkeypatch.setattr(audio.smart_fades_mixer, "build", build)
922 monkeypatch.setattr(audio.smart_fades_mixer, "mix", _empty_mix)
923
924 async def _current_stream(
925 queue_item: object, *_args: object, **_kwargs: object
926 ) -> AsyncGenerator[bytes]:
927 if queue_item is not current_item:
928 return
929 # a whole second, then chunks that never line up with a second boundary
930 yield bytes(pcm_format.pcm_sample_size * 8)
931 for _ in range(30):
932 yield bytes(pcm_format.pcm_sample_size // 3)
933
934 monkeypatch.setattr(audio, "get_queue_item_stream", _current_stream)
935 stream = audio.get_queue_item_stream_with_smartfade(
936 cast("Any", player),
937 cast("Any", current_item),
938 pcm_format,
939 crossfade_mode=CrossfadeMode.STANDARD_CROSSFADE,
940 standard_crossfade_duration=8,
941 )
942
943 async for _chunk in stream:
944 pass
945
946 build.assert_awaited_once()
947
948
949async def test_smartfade_short_remainder_still_crossfades(
950 monkeypatch: pytest.MonkeyPatch,
951) -> None:
952 """Less audio left than the configured overlap still fades with what is there."""
953 pcm_format = AudioFormat(
954 content_type=ContentType.PCM_S16LE,
955 sample_rate=8000,
956 bit_depth=16,
957 channels=2,
958 )
959 current_details = SimpleNamespace(
960 duration=180,
961 seek_position=146,
962 seconds_streamed=0,
963 uri="test://current",
964 buffer=SimpleNamespace(eof=True, cancelled=False, has_error=False, max_size_seconds=300),
965 is_realtime=False,
966 )
967 next_details = SimpleNamespace(
968 audio_format=pcm_format,
969 buffer=_buffer(SMART_CROSSFADE_DURATION, ready=True),
970 duration=180,
971 seek_position=0,
972 uri="test://next",
973 is_realtime=False,
974 volume_normalization_mode=None,
975 )
976 current_item = SimpleNamespace(
977 queue_id="queue-1",
978 queue_item_id="current",
979 name="Current",
980 streamdetails=current_details,
981 extra_attributes={},
982 )
983 next_item = SimpleNamespace(
984 queue_id="queue-1",
985 queue_item_id="next",
986 name="Next",
987 streamdetails=next_details,
988 extra_attributes={},
989 available=True,
990 )
991 queue = SimpleNamespace(queue_id="queue-1", display_name="Queue", index_in_buffer=0)
992 player = SimpleNamespace(player_id="player-1", name="Player")
993 mass = MagicMock()
994 mass.player_queues.get.return_value = queue
995 mass.player_queues.load_next_queue_item = AsyncMock(return_value=next_item)
996 mass.player_queues.index_by_id.return_value = 1
997 audio = StreamsAudio(cast("Any", mass))
998 audio.setup()
999 audio.select_pcm_format = AsyncMock(return_value=pcm_format) # type: ignore[method-assign]
1000 audio.crossfade_allowed = MagicMock(return_value=True) # type: ignore[method-assign]
1001 build = AsyncMock(
1002 return_value=SimpleNamespace(
1003 timing_info=SimpleNamespace(
1004 pre_crossfade_duration=2,
1005 crossfade_duration=6,
1006 fadein_trimmed_duration=0,
1007 )
1008 )
1009 )
1010 monkeypatch.setattr(audio.smart_fades_mixer, "build", build)
1011 monkeypatch.setattr(audio.smart_fades_mixer, "mix", _empty_mix)
1012
1013 async def _current_stream(
1014 queue_item: object, *_args: object, **_kwargs: object
1015 ) -> AsyncGenerator[bytes]:
1016 if queue_item is not current_item:
1017 return
1018 # a seek near the end leaves 34s, less than the 45s smart overlap
1019 yield bytes(pcm_format.pcm_sample_size * 8)
1020 yield bytes(pcm_format.pcm_sample_size * 26)
1021
1022 monkeypatch.setattr(audio, "get_queue_item_stream", _current_stream)
1023 stream = audio.get_queue_item_stream_with_smartfade(
1024 cast("Any", player),
1025 cast("Any", current_item),
1026 pcm_format,
1027 crossfade_mode=CrossfadeMode.SMART_CROSSFADE,
1028 standard_crossfade_duration=8,
1029 )
1030
1031 async for _chunk in stream:
1032 pass
1033
1034 build.assert_awaited_once()
1035 assert build.await_args is not None
1036 fade_out_seconds = len(build.await_args.kwargs["fade_out_data"]) / pcm_format.pcm_sample_size
1037 assert fade_out_seconds == pytest.approx(26, abs=1)
1038
1039
1040async def test_smartfade_stub_remainder_does_not_crossfade(
1041 monkeypatch: pytest.MonkeyPatch,
1042) -> None:
1043 """A remainder too short to overlap with is played out instead of faded."""
1044 pcm_format = AudioFormat(
1045 content_type=ContentType.PCM_S16LE,
1046 sample_rate=8000,
1047 bit_depth=16,
1048 channels=2,
1049 )
1050 current_details = SimpleNamespace(
1051 duration=180,
1052 seek_position=176,
1053 seconds_streamed=0,
1054 uri="test://current",
1055 buffer=SimpleNamespace(eof=True, cancelled=False, has_error=False, max_size_seconds=300),
1056 is_realtime=False,
1057 )
1058 next_details = SimpleNamespace(
1059 audio_format=pcm_format,
1060 buffer=_buffer(SMART_CROSSFADE_DURATION, ready=True),
1061 duration=180,
1062 seek_position=0,
1063 uri="test://next",
1064 is_realtime=False,
1065 volume_normalization_mode=None,
1066 )
1067 current_item = SimpleNamespace(
1068 queue_id="queue-1",
1069 queue_item_id="current",
1070 name="Current",
1071 streamdetails=current_details,
1072 extra_attributes={},
1073 )
1074 next_item = SimpleNamespace(
1075 queue_id="queue-1",
1076 queue_item_id="next",
1077 name="Next",
1078 streamdetails=next_details,
1079 extra_attributes={},
1080 available=True,
1081 )
1082 queue = SimpleNamespace(queue_id="queue-1", display_name="Queue", index_in_buffer=0)
1083 player = SimpleNamespace(player_id="player-1", name="Player")
1084 mass = MagicMock()
1085 mass.player_queues.get.return_value = queue
1086 mass.player_queues.load_next_queue_item = AsyncMock(return_value=next_item)
1087 mass.player_queues.index_by_id.return_value = 1
1088 audio = StreamsAudio(cast("Any", mass))
1089 audio.setup()
1090 audio.select_pcm_format = AsyncMock(return_value=pcm_format) # type: ignore[method-assign]
1091 audio.crossfade_allowed = MagicMock(return_value=True) # type: ignore[method-assign]
1092 build = AsyncMock()
1093 monkeypatch.setattr(audio.smart_fades_mixer, "build", build)
1094
1095 async def _current_stream(
1096 queue_item: object, *_args: object, **_kwargs: object
1097 ) -> AsyncGenerator[bytes]:
1098 if queue_item is not current_item:
1099 return
1100 yield bytes(pcm_format.pcm_sample_size * 8)
1101 yield bytes(pcm_format.pcm_sample_size * 2)
1102
1103 monkeypatch.setattr(audio, "get_queue_item_stream", _current_stream)
1104 stream = audio.get_queue_item_stream_with_smartfade(
1105 cast("Any", player),
1106 cast("Any", current_item),
1107 pcm_format,
1108 crossfade_mode=CrossfadeMode.SMART_CROSSFADE,
1109 standard_crossfade_duration=8,
1110 )
1111
1112 output = b"".join([chunk async for chunk in stream])
1113
1114 assert len(output) == pcm_format.pcm_sample_size * 10
1115 build.assert_not_awaited()
1116
1117
1118async def test_flow_reports_no_fade_for_a_realtime_item_until_one_renders(
1119 monkeypatch: pytest.MonkeyPatch,
1120) -> None:
1121 """
1122 A realtime item is not credited with any fade up front.
1123
1124 A fade is only reported once one is really rendered at its boundary; the
1125 source-delegation reporting is gone along with the delegation itself.
1126 """
1127 pcm_format = AudioFormat(
1128 content_type=ContentType.PCM_S16LE,
1129 sample_rate=8000,
1130 bit_depth=16,
1131 channels=2,
1132 )
1133 realtime_details = SimpleNamespace(
1134 audio_format=pcm_format,
1135 buffer=SimpleNamespace(eof=True, cancelled=False, has_error=False, max_size_seconds=300),
1136 fade_in=False,
1137 stream_error=False,
1138 uri="test://realtime",
1139 seek_position=0,
1140 seconds_streamed=0,
1141 duration=20,
1142 is_realtime=True,
1143 )
1144 realtime_item = SimpleNamespace(
1145 queue_id="queue-1",
1146 queue_item_id="item-1",
1147 name="Realtime",
1148 media_type=MediaType.TRACK,
1149 media_item=None,
1150 streamdetails=realtime_details,
1151 duration=20,
1152 extra_attributes={},
1153 )
1154 queue = SimpleNamespace(
1155 queue_id="queue-1",
1156 display_name="Queue",
1157 flow_mode=False,
1158 overlay_enabled=False,
1159 overlay_source=None,
1160 )
1161 mass = MagicMock()
1162 mass.player_queues.queue_data.return_value = SimpleNamespace(
1163 session_id="session-1", flow_mode_stream_log=[]
1164 )
1165 mass.player_queues.load_next_queue_item = AsyncMock(side_effect=QueueEmpty)
1166 mass.player_queues.get.return_value = queue
1167 mass.streams.get_crossfade_mode.return_value = CrossfadeMode.SMART_CROSSFADE
1168 mass.config.get_raw_core_config_value.return_value = 8
1169 update_item_context = MagicMock()
1170 mass.streams.audio_processing.update_item_context = update_item_context
1171 player = MagicMock()
1172 player.config.get_value.return_value = "fixed_48000"
1173 player.get_supported_sample_rates.return_value = []
1174 mass.players.get_player.return_value = player
1175 audio = StreamsAudio(cast("Any", mass))
1176 audio.setup()
1177
1178 async def _item_stream(*_args: object, **_kwargs: object) -> AsyncGenerator[bytes]:
1179 yield bytes(pcm_format.pcm_sample_size * 4)
1180
1181 monkeypatch.setattr(audio, "get_queue_item_stream", _item_stream)
1182 stream = audio.get_queue_flow_stream(
1183 cast("Any", queue), cast("Any", realtime_item), pcm_format, session_id="session-1"
1184 )
1185
1186 async for _chunk in stream:
1187 pass
1188
1189 update_item_context.assert_called()
1190 reported = update_item_context.call_args.kwargs["queue_processing"]
1191 assert reported.crossfade_mode == CrossfadeMode.DISABLED
1192
1193
1194async def test_flow_standard_fade_only_holds_back_its_overlap(
1195 monkeypatch: pytest.MonkeyPatch,
1196) -> None:
1197 """A standard transition waits for its overlap, not for the whole requested window."""
1198 pcm_format = AudioFormat(
1199 content_type=ContentType.PCM_S16LE,
1200 sample_rate=8000,
1201 bit_depth=16,
1202 channels=2,
1203 )
1204 first_details = SimpleNamespace(
1205 audio_format=pcm_format,
1206 buffer=SimpleNamespace(eof=True, cancelled=False, has_error=False, max_size_seconds=300),
1207 fade_in=False,
1208 stream_error=False,
1209 uri="test://first",
1210 seek_position=0,
1211 seconds_streamed=0,
1212 duration=300,
1213 is_realtime=False,
1214 )
1215 second_details = SimpleNamespace(
1216 audio_format=pcm_format,
1217 buffer=_buffer(SMART_CROSSFADE_DURATION, ready=True),
1218 fade_in=False,
1219 stream_error=False,
1220 uri="test://second",
1221 seek_position=0,
1222 seconds_streamed=0,
1223 duration=300,
1224 is_realtime=False,
1225 volume_normalization_mode=None,
1226 )
1227 first_item = SimpleNamespace(
1228 queue_id="queue-1",
1229 queue_item_id="item-1",
1230 name="First",
1231 media_type=MediaType.TRACK,
1232 media_item=None,
1233 streamdetails=first_details,
1234 duration=300,
1235 extra_attributes={},
1236 )
1237 second_item = SimpleNamespace(
1238 queue_id="queue-1",
1239 queue_item_id="item-2",
1240 name="Second",
1241 media_type=MediaType.TRACK,
1242 media_item=None,
1243 streamdetails=second_details,
1244 duration=300,
1245 extra_attributes={},
1246 )
1247 queue = SimpleNamespace(
1248 queue_id="queue-1",
1249 display_name="Queue",
1250 flow_mode=False,
1251 overlay_enabled=False,
1252 overlay_source=None,
1253 )
1254 mass = MagicMock()
1255 mass.player_queues.queue_data.return_value = SimpleNamespace(
1256 session_id="session-1", flow_mode_stream_log=[]
1257 )
1258 mass.player_queues.load_next_queue_item = AsyncMock(side_effect=[second_item, QueueEmpty])
1259 mass.player_queues.get.return_value = queue
1260 mass.streams.get_crossfade_mode.return_value = CrossfadeMode.SMART_CROSSFADE
1261 mass.config.get_raw_core_config_value.return_value = 8
1262 player = MagicMock()
1263 player.config.get_value.return_value = "fixed_48000"
1264 player.get_supported_sample_rates.return_value = []
1265 mass.players.get_player.return_value = player
1266 audio = StreamsAudio(cast("Any", mass))
1267 audio.setup()
1268 audio.crossfade_allowed = MagicMock(return_value=True) # type: ignore[method-assign]
1269 # the incoming analysis is not ready, so the mixer degrades to a standard fade
1270 standard = StandardCrossFade(logger=MagicMock(), crossfade_duration=8)
1271 standard.build(
1272 pcm_format.pcm_sample_size * SMART_CROSSFADE_DURATION,
1273 pcm_format.pcm_sample_size * SMART_CROSSFADE_DURATION,
1274 pcm_format,
1275 )
1276 monkeypatch.setattr(audio.smart_fades_mixer, "build", AsyncMock(return_value=standard))
1277 monkeypatch.setattr(audio.smart_fades_mixer, "mix", _empty_mix)
1278
1279 consumed: dict[str, int] = {"second": 0}
1280
1281 async def _item_stream(
1282 queue_item: SimpleNamespace, *_args: object, **_kwargs: object
1283 ) -> AsyncGenerator[bytes]:
1284 if queue_item is first_item:
1285 for _ in range(60):
1286 yield bytes(pcm_format.pcm_sample_size)
1287 return
1288 for _ in range(SMART_CROSSFADE_DURATION + 20):
1289 consumed["second"] += 1
1290 yield bytes(pcm_format.pcm_sample_size)
1291
1292 monkeypatch.setattr(audio, "get_queue_item_stream", _item_stream)
1293 stream = audio.get_queue_flow_stream(
1294 cast("Any", queue), cast("Any", first_item), pcm_format, session_id="session-1"
1295 )
1296
1297 seconds_before_transition: int | None = None
1298 async for _chunk in stream:
1299 if seconds_before_transition is None and consumed["second"]:
1300 seconds_before_transition = consumed["second"]
1301
1302 # the overlap is 8s, so the transition must not wait for the full 45s window
1303 assert seconds_before_transition is not None
1304 assert seconds_before_transition <= SMART_CROSSFADE_DURATION / 2
1305
1306
1307# -- StreamsController.serve_queue_item_stream steering --
1308
1309
1310class _PcmFormatRequested(Exception):
1311 """Raised to stop the handler once it has decided on crossfading."""
1312
1313
1314def _single_item_handler(*, is_realtime: bool) -> tuple[Any, MagicMock, dict[str, Any]]:
1315 """Return a single-item stream handler that stops once the PCM format is picked."""
1316 streamdetails = _make_stream_details(MediaType.TRACK, is_realtime=is_realtime)
1317 queue_item = SimpleNamespace(
1318 queue_id="queue-1",
1319 queue_item_id="item-1",
1320 name="Track",
1321 duration=180,
1322 streamdetails=streamdetails,
1323 media_item=None,
1324 media_type=MediaType.TRACK,
1325 extra_attributes={},
1326 image=None,
1327 )
1328 queue = SimpleNamespace(
1329 queue_id="queue-1",
1330 display_name="Queue",
1331 current_item=queue_item,
1332 crossfade_enabled=True,
1333 overlay_enabled=False,
1334 overlay_source=None,
1335 )
1336 mass = MagicMock()
1337 mass.player_queues.get.return_value = queue
1338 mass.player_queues.queue_data.return_value = SimpleNamespace(session_id="session-1")
1339 mass.player_queues.get_item.return_value = queue_item
1340 mass.config.get_raw_core_config_value.return_value = 8
1341 player = MagicMock(player_id="player-1", protocol_parent_id=None)
1342 player.state.supported_features = {PlayerFeature.GAPLESS_PLAYBACK}
1343 player.state.name = "Player"
1344 mass.players.get_player.return_value = player
1345
1346 seen: dict[str, Any] = {}
1347
1348 async def _select_pcm_format(**kwargs: Any) -> None:
1349 seen["crossfade_enabled"] = kwargs["crossfade_enabled"]
1350 raise _PcmFormatRequested
1351
1352 audio = MagicMock()
1353 audio.select_pcm_format = _select_pcm_format
1354 controller = cast("Any", object.__new__(StreamsController))
1355 controller.mass = mass
1356 controller.audio = audio
1357 controller.logger = MagicMock()
1358 controller._log_request = MagicMock()
1359 controller.get_crossfade_mode = MagicMock(return_value=CrossfadeMode.SMART_CROSSFADE)
1360 request = MagicMock()
1361 request.method = "GET"
1362 request.match_info = {
1363 "queue_id": "queue-1",
1364 "player_id": "player-1",
1365 "session_id": "session-1",
1366 "queue_item_id": "item-1",
1367 }
1368 return controller, request, seen
1369
1370
1371async def test_single_item_handler_keeps_crossfade_for_a_realtime_item() -> None:
1372 """A realtime item whose source does not fade keeps the queue's crossfade."""
1373 controller, request, seen = _single_item_handler(is_realtime=True)
1374
1375 with pytest.raises(_PcmFormatRequested):
1376 await controller.serve_queue_item_stream(request)
1377
1378 assert seen["crossfade_enabled"] is True
1379 controller.get_crossfade_mode.assert_called_once()
1380
1381
1382async def test_single_item_handler_keeps_crossfade_for_a_buffered_item() -> None:
1383 """A buffered item still gets the queue's configured crossfade."""
1384 controller, request, seen = _single_item_handler(is_realtime=False)
1385
1386 with pytest.raises(_PcmFormatRequested):
1387 await controller.serve_queue_item_stream(request)
1388
1389 assert seen["crossfade_enabled"] is True
1390 controller.get_crossfade_mode.assert_called_once()
1391
1392
1393# -- StreamsAudio.get_stream_details --
1394
1395
1396@pytest.mark.parametrize(
1397 ("media_item_cls", "media_type", "expected_is_realtime"),
1398 [
1399 pytest.param(Radio, MediaType.RADIO, True, id="radio"),
1400 pytest.param(AudioSource, MediaType.AUDIO_SOURCE, True, id="audio_source"),
1401 pytest.param(Track, MediaType.TRACK, False, id="track"),
1402 ],
1403)
1404async def test_get_stream_details_sets_is_realtime_by_media_type(
1405 media_item_cls: type, media_type: MediaType, expected_is_realtime: bool
1406) -> None:
1407 """RADIO and AUDIO_SOURCE streams are marked realtime; a TRACK's flag is left alone."""
1408 provider_streamdetails = StreamDetails(
1409 provider="test--1",
1410 item_id="item-1",
1411 audio_format=AudioFormat(content_type=ContentType.MP3),
1412 media_type=media_type,
1413 stream_type=StreamType.CUSTOM,
1414 duration=180 if media_type == MediaType.TRACK else None,
1415 )
1416 audio = _stream_details_provider(provider_streamdetails)
1417
1418 streamdetails = await audio.get_stream_details(
1419 queue_item=_queue_item_with_mapping(media_item_cls)
1420 )
1421
1422 assert streamdetails.is_realtime is expected_is_realtime
1423