/
/
/
1"""Tests for the AudioBuffer class."""
2
3from __future__ import annotations
4
5import asyncio
6import logging
7import time
8from collections.abc import AsyncGenerator
9from contextlib import suppress
10from types import SimpleNamespace
11from typing import Any, cast
12from unittest.mock import AsyncMock, MagicMock, patch
13
14import pytest
15from music_assistant_models.enums import ContentType, MediaType, StreamType
16from music_assistant_models.errors import AudioError
17from music_assistant_models.media_items import AudioFormat
18from music_assistant_models.queue_item import QueueItem
19from music_assistant_models.streamdetails import StreamDetails
20
21import music_assistant.controllers.streams.audio as audio_mod
22from music_assistant.controllers.streams.audio import StreamsAudio
23from music_assistant.controllers.streams.audio_buffer import (
24 AudioBuffer,
25 AudioBufferDiscarded,
26 AudioBufferEOF,
27)
28from music_assistant.controllers.streams.constants import (
29 BUFFER_SIZE_MAP,
30 RADIO_BUFFER_SIZE,
31 SEEK_WAIT_THRESHOLD,
32 BufferMode,
33 BufferSize,
34)
35from music_assistant.mass import MusicAssistant
36from music_assistant.models.music_provider import MusicProvider
37
38# Standard test PCM format: 44100Hz, 16-bit, stereo
39TEST_PCM_FORMAT = AudioFormat(
40 content_type=ContentType.PCM_S16LE,
41 sample_rate=44100,
42 bit_depth=16,
43 channels=2,
44)
45
46# One second of silence in the test format
47ONE_SECOND_CHUNK = b"\x00" * TEST_PCM_FORMAT.pcm_sample_size
48
49
50def _make_chunk(value: int = 0) -> bytes:
51 """Create a 1-second PCM chunk filled with a byte value."""
52 return bytes([value % 256]) * TEST_PCM_FORMAT.pcm_sample_size
53
54
55async def _make_source(num_chunks: int) -> AsyncGenerator[bytes]:
56 """Create an async generator that yields numbered chunks."""
57 for i in range(num_chunks):
58 yield _make_chunk(i)
59
60
61def _make_stream_details(
62 media_type: MediaType,
63 *,
64 duration: int | None,
65 allow_seek: bool,
66 queue_id: str | None = None,
67) -> StreamDetails:
68 """Build minimal stream details for AudioBuffer.get_buffer tests."""
69 return StreamDetails(
70 provider="builtin",
71 item_id="item-1",
72 audio_format=TEST_PCM_FORMAT,
73 media_type=media_type,
74 stream_type=StreamType.HTTP,
75 path="http://example.com/audio.mp3",
76 duration=duration,
77 can_seek=allow_seek,
78 allow_seek=allow_seek,
79 queue_id=queue_id,
80 )
81
82
83def _make_mass_for_get_buffer(
84 *, queue: Any | None = None
85) -> tuple[MagicMock, AsyncMock, list[asyncio.Task[None]]]:
86 """Build a minimal mass stub for AudioBuffer.get_buffer tests."""
87
88 def _get_media_stream(*_args: Any, **_kwargs: Any) -> AsyncGenerator[bytes]:
89 return _make_source(1)
90
91 mass = MagicMock()
92 mass.config.get_raw_core_config_value.return_value = BufferSize.BALANCED.value
93 mass.player_queues.get.return_value = queue
94 start_analysis = AsyncMock(return_value=None)
95 mass.streams = SimpleNamespace(
96 audio_analysis=SimpleNamespace(start_analysis=start_analysis),
97 audio=SimpleNamespace(get_media_stream=_get_media_stream),
98 )
99 scheduled_tasks: list[asyncio.Task[None]] = []
100
101 def _create_task(coro: Any) -> asyncio.Task[None]:
102 task = asyncio.create_task(coro)
103 scheduled_tasks.append(task)
104 return task
105
106 mass.create_task.side_effect = _create_task
107 return mass, start_analysis, scheduled_tasks
108
109
110# -- Init and properties --
111
112
113def test_init_defaults() -> None:
114 """AudioBuffer initializes with correct defaults."""
115 buf = AudioBuffer(TEST_PCM_FORMAT)
116 assert buf.pcm_format == TEST_PCM_FORMAT
117 assert buf.mode == BufferMode.SEEKABLE
118 assert buf.max_size_seconds == BUFFER_SIZE_MAP[BufferSize.BALANCED]
119 assert buf.size_seconds == 0
120 assert buf.seconds_available == 0
121 assert buf.duration_available == 0
122 assert not buf.cancelled
123 assert not buf.has_error
124 assert not buf.ready.is_set()
125
126
127@pytest.mark.asyncio
128async def test_realtime_buffer_refuses_a_seek_past_what_it_produced() -> None:
129 """
130 A realtime source is re-seeked rather than waited for on a forward seek.
131
132 Such a source hands its audio over at playback pace, so covering the gap
133 costs exactly the gap; a fresh producer starts at the position right away.
134 """
135 live = AudioBuffer(TEST_PCM_FORMAT, is_realtime=True)
136 recorded = AudioBuffer(TEST_PCM_FORMAT)
137 for buf in (live, recorded):
138 await buf._put(ONE_SECOND_CHUNK)
139
140 # already produced, so both serve it from what they hold
141 assert live.is_valid(0)
142 assert recorded.is_valid(0)
143
144 # a second past the head: the recorded source catches up, the live one cannot
145 assert not live.is_valid(2000)
146 assert recorded.is_valid(2000)
147 assert not recorded.is_valid((SEEK_WAIT_THRESHOLD + 2) * 1000)
148
149
150@pytest.mark.asyncio
151async def test_realtime_buffer_stays_valid_before_it_holds_anything() -> None:
152 """A buffer that has not produced its first second yet is still the right one."""
153 live = AudioBuffer(TEST_PCM_FORMAT, is_realtime=True)
154
155 assert live.is_valid(0)
156
157
158@pytest.mark.asyncio
159async def test_realtime_buffer_serves_the_position_it_was_seeded_at() -> None:
160 """A buffer built for a seek answers for that position before it holds anything."""
161 live = AudioBuffer(TEST_PCM_FORMAT, is_realtime=True)
162 live._discarded_chunks = 65
163
164 assert live.is_valid(65000)
165 assert live.is_valid(65999)
166 # before its own start, and past the second it is about to produce
167 assert not live.is_valid(64000)
168 assert not live.is_valid(66000)
169
170 await live._put(ONE_SECOND_CHUNK)
171
172 assert live.is_valid(66000)
173 assert not live.is_valid(67000)
174
175
176def test_init_minimal_buffer() -> None:
177 """AudioBuffer with MINIMAL preset has correct max size."""
178 buf = AudioBuffer(TEST_PCM_FORMAT, buffer_size=BufferSize.MINIMAL)
179 assert buf.max_size_seconds == BUFFER_SIZE_MAP[BufferSize.MINIMAL]
180
181
182def test_init_rolling_mode() -> None:
183 """ROLLING mode uses radio buffer size regardless of preset."""
184 buf = AudioBuffer(TEST_PCM_FORMAT, buffer_size=BufferSize.MAXIMUM, mode=BufferMode.ROLLING)
185 assert buf.max_size_seconds == RADIO_BUFFER_SIZE
186
187
188# -- Put and get --
189
190
191async def test_duration_available_uses_exact_resident_byte_count() -> None:
192 """A partial EOF chunk contributes its exact PCM duration."""
193 audio_buffer = AudioBuffer(TEST_PCM_FORMAT)
194 await audio_buffer._put(ONE_SECOND_CHUNK)
195 await audio_buffer._put(ONE_SECOND_CHUNK[: len(ONE_SECOND_CHUNK) // 2])
196
197 assert audio_buffer.seconds_available == 2
198 assert audio_buffer.duration_available == 1.5
199
200
201@pytest.mark.asyncio
202async def test_put_and_get() -> None:
203 """Basic put/get cycle works correctly."""
204 buf = AudioBuffer(TEST_PCM_FORMAT, buffer_size=BufferSize.MINIMAL)
205 await buf._put(ONE_SECOND_CHUNK)
206 result = await buf._get(chunk_number=0)
207 assert result == ONE_SECOND_CHUNK
208
209
210@pytest.mark.asyncio
211async def test_put_sets_ready_default_threshold() -> None:
212 """Ready event is set after 1 chunk with default threshold."""
213 buf = AudioBuffer(TEST_PCM_FORMAT)
214 assert not buf.ready.is_set()
215 await buf._put(ONE_SECOND_CHUNK)
216 assert buf.ready.is_set()
217
218
219@pytest.mark.asyncio
220async def test_put_sets_ready_custom_threshold() -> None:
221 """Ready event is set after ready_threshold chunks are buffered."""
222 buf = AudioBuffer(TEST_PCM_FORMAT, ready_threshold=3)
223 assert not buf.ready.is_set()
224 await buf._put(ONE_SECOND_CHUNK)
225 assert not buf.ready.is_set()
226 await buf._put(ONE_SECOND_CHUNK)
227 assert not buf.ready.is_set()
228 await buf._put(ONE_SECOND_CHUNK)
229 assert buf.ready.is_set()
230
231
232@pytest.mark.asyncio
233async def test_eof_sets_ready_below_threshold() -> None:
234 """EOF sets ready even when fewer than threshold chunks are buffered."""
235 buf = AudioBuffer(TEST_PCM_FORMAT, ready_threshold=5)
236 await buf._put(ONE_SECOND_CHUNK)
237 assert not buf.ready.is_set()
238 await buf._set_eof()
239 assert buf.ready.is_set()
240
241
242@pytest.mark.asyncio
243async def test_get_waits_for_data() -> None:
244 """Get waits until data is available."""
245 buf = AudioBuffer(TEST_PCM_FORMAT)
246
247 async def _delayed_put() -> None:
248 await asyncio.sleep(0.05)
249 await buf._put(ONE_SECOND_CHUNK)
250
251 asyncio.get_event_loop().create_task(_delayed_put())
252 result = await buf._get(chunk_number=0)
253 assert result == ONE_SECOND_CHUNK
254
255
256@pytest.mark.asyncio
257async def test_get_raises_on_eof() -> None:
258 """Get raises AudioBufferEOF when EOF is set and chunk not available."""
259 buf = AudioBuffer(TEST_PCM_FORMAT)
260 await buf._set_eof()
261 with pytest.raises(AudioBufferEOF):
262 await buf._get(chunk_number=0)
263
264
265@pytest.mark.asyncio
266async def test_get_after_cancel() -> None:
267 """Get raises AudioBufferEOF when buffer is cleared."""
268 buf = AudioBuffer(TEST_PCM_FORMAT)
269 await buf._put(ONE_SECOND_CHUNK)
270 await buf.clear()
271 with pytest.raises(AudioBufferEOF):
272 await buf._get(chunk_number=0)
273
274
275# -- Fill and stream --
276
277
278@pytest.mark.asyncio
279async def test_fill_and_raw_stream() -> None:
280 """Fill from async generator and iterate via get_raw_stream."""
281 buf = AudioBuffer(TEST_PCM_FORMAT, buffer_size=BufferSize.MINIMAL)
282 buf.fill(_make_source(5), source_name="test")
283
284 # wait for fill to complete
285 await asyncio.sleep(0.1)
286
287 chunks = []
288 async for chunk in buf.get_raw_stream():
289 chunks.append(chunk)
290
291 assert len(chunks) == 5
292 # verify chunk content matches what we generated
293 for i, chunk in enumerate(chunks):
294 assert chunk == _make_chunk(i)
295
296
297@pytest.mark.asyncio
298async def test_fill_sets_eof() -> None:
299 """Fill sets EOF when the source generator completes."""
300 buf = AudioBuffer(TEST_PCM_FORMAT, buffer_size=BufferSize.MINIMAL)
301 buf.fill(_make_source(3), source_name="test")
302 await asyncio.sleep(0.1)
303 assert buf._eof_received
304
305
306@pytest.mark.asyncio
307async def test_fill_error_propagation() -> None:
308 """When the source errors after producing data, valid chunks are still delivered."""
309
310 async def _failing_source() -> AsyncGenerator[bytes]:
311 yield ONE_SECOND_CHUNK
312 msg = "test error"
313 raise RuntimeError(msg)
314
315 buf = AudioBuffer(TEST_PCM_FORMAT, buffer_size=BufferSize.MINIMAL)
316 buf.fill(_failing_source(), source_name="test")
317 await asyncio.sleep(0.1)
318
319 assert buf.has_error
320
321 # consumer should receive the valid chunk before the source error surfaces.
322 result: list[bytes] = []
323
324 async def _consume() -> None:
325 async for chunk in buf.get_raw_stream():
326 result.append(chunk)
327
328 with pytest.raises(RuntimeError, match="test error"):
329 await _consume()
330 assert result == [ONE_SECOND_CHUNK]
331
332
333@pytest.mark.asyncio
334async def test_fill_error_surfaces_to_analysis_reader() -> None:
335 """An aborted source raises its error to the analysis reader instead of a clean EOF."""
336
337 async def _failing_source() -> AsyncGenerator[bytes]:
338 yield ONE_SECOND_CHUNK
339 msg = "test error"
340 raise RuntimeError(msg)
341
342 buf = AudioBuffer(TEST_PCM_FORMAT, buffer_size=BufferSize.MINIMAL)
343 buf.fill(_failing_source(), source_name="test")
344
345 # buffered chunks are still delivered before the error surfaces
346 assert await buf.read_chunk_for_analysis(0) == ONE_SECOND_CHUNK
347 with pytest.raises(RuntimeError, match="test error"):
348 await buf.read_chunk_for_analysis(1)
349
350
351@pytest.mark.asyncio
352async def test_fill_error_no_data() -> None:
353 """When the source errors without producing any data, the error propagates."""
354
355 async def _failing_source() -> AsyncGenerator[bytes]:
356 msg = "test error"
357 raise RuntimeError(msg)
358 yield # type: ignore[unreachable]
359
360 buf = AudioBuffer(TEST_PCM_FORMAT, buffer_size=BufferSize.MINIMAL)
361 buf.fill(_failing_source(), source_name="test")
362 await asyncio.sleep(0.1)
363
364 assert buf.has_error
365
366 async def _consume() -> list[bytes]:
367 result = []
368 async for chunk in buf.get_raw_stream():
369 result.append(chunk)
370 return result
371
372 with pytest.raises(RuntimeError, match="test error"):
373 await _consume()
374
375
376async def _silent_source() -> AsyncGenerator[bytes]:
377 """Create an async generator that never delivers audio."""
378 await asyncio.sleep(10)
379 yield ONE_SECOND_CHUNK
380
381
382@pytest.mark.asyncio
383async def test_logs_time_to_first_playable_audio(caplog: pytest.LogCaptureFixture) -> None:
384 """A buffer reports how long its first playable audio took."""
385 buf = AudioBuffer(TEST_PCM_FORMAT, buffer_size=BufferSize.MINIMAL)
386 with caplog.at_level(logging.DEBUG, logger="music_assistant.audio_buffer"):
387 buf.fill(_make_source(2), source_name="test://item")
388 await buf.ready.wait()
389
390 assert "test://item became ready after" in caplog.text
391
392
393@pytest.mark.asyncio
394async def test_logs_time_to_ready_for_a_stream_below_its_threshold(
395 caplog: pytest.LogCaptureFixture,
396) -> None:
397 """A stream that ends before reaching its ready threshold still reports its startup time."""
398 buf = AudioBuffer(TEST_PCM_FORMAT, buffer_size=BufferSize.MINIMAL, ready_threshold=8)
399 with caplog.at_level(logging.DEBUG, logger="music_assistant.audio_buffer"):
400 buf.fill(_make_source(2), source_name="test://short")
401 await buf.ready.wait()
402
403 assert "test://short became ready after" in caplog.text
404
405
406@pytest.mark.asyncio
407async def test_empty_stream_is_not_reported_as_ready(caplog: pytest.LogCaptureFixture) -> None:
408 """A source that ends without delivering audio never became playable."""
409 buf = AudioBuffer(TEST_PCM_FORMAT, buffer_size=BufferSize.MINIMAL)
410 with caplog.at_level(logging.DEBUG, logger="music_assistant.audio_buffer"):
411 buf.fill(_make_source(0), source_name="test://empty")
412 await buf.ready.wait()
413
414 assert "became ready after" not in caplog.text
415
416
417@pytest.mark.asyncio
418async def test_failed_stream_below_threshold_is_not_reported_as_ready(
419 caplog: pytest.LogCaptureFixture,
420) -> None:
421 """A source that fails before reaching its threshold never became playable."""
422
423 async def _failing_source() -> AsyncGenerator[bytes]:
424 yield ONE_SECOND_CHUNK
425 msg = "test error"
426 raise RuntimeError(msg)
427
428 buf = AudioBuffer(TEST_PCM_FORMAT, buffer_size=BufferSize.MINIMAL, ready_threshold=8)
429 with caplog.at_level(logging.DEBUG, logger="music_assistant.audio_buffer"):
430 buf.fill(_failing_source(), source_name="test://failing")
431 await buf.ready.wait()
432
433 assert "became ready after" not in caplog.text
434
435
436@pytest.mark.asyncio
437async def test_ready_timeout_reports_the_wait(caplog: pytest.LogCaptureFixture) -> None:
438 """The readiness deadline reports the caller, provider and how much audio arrived."""
439 streamdetails = _make_stream_details(MediaType.TRACK, duration=100, allow_seek=True)
440 buf = AudioBuffer(TEST_PCM_FORMAT, buffer_size=BufferSize.MINIMAL)
441 streamdetails.buffer = buf
442 buf.fill(_silent_source(), source_name=streamdetails.uri)
443
444 with (
445 caplog.at_level(logging.WARNING),
446 pytest.raises(AudioError, match="Timeout waiting for audio data"),
447 ):
448 await buf._wait_until_ready(streamdetails, 0.05, "get_buffer[prepare]")
449
450 assert "get_buffer[prepare]: Gave up on builtin" in caplog.text
451 assert ", 0s buffered" in caplog.text
452 assert streamdetails.buffer is None
453
454
455@pytest.mark.asyncio
456async def test_ready_timeout_stays_quiet_for_a_released_buffer(
457 caplog: pytest.LogCaptureFixture,
458) -> None:
459 """A buffer released to free a stream slot is not reported as a provider stall."""
460 streamdetails = _make_stream_details(MediaType.TRACK, duration=100, allow_seek=True)
461 buf = AudioBuffer(TEST_PCM_FORMAT, buffer_size=BufferSize.MINIMAL)
462 streamdetails.buffer = buf
463 buf.fill(_silent_source(), source_name=streamdetails.uri)
464
465 with caplog.at_level(logging.WARNING):
466 waiter = asyncio.create_task(
467 buf._wait_until_ready(streamdetails, 0.05, "get_buffer[prepare_next]")
468 )
469 await asyncio.sleep(0)
470 await buf.clear()
471 with pytest.raises(AudioError, match="Timeout waiting for audio data"):
472 await waiter
473
474 assert "Gave up on" not in caplog.text
475
476
477@pytest.mark.asyncio
478async def test_ready_timeout_stays_quiet_while_the_source_is_finalizing(
479 caplog: pytest.LogCaptureFixture,
480) -> None:
481 """A release is not reported as a stall while the source is still cleaning up."""
482 finalized = asyncio.Event()
483
484 async def _slow_to_finalize_source() -> AsyncGenerator[bytes]:
485 try:
486 await asyncio.sleep(10)
487 yield ONE_SECOND_CHUNK
488 finally:
489 await finalized.wait()
490
491 streamdetails = _make_stream_details(MediaType.TRACK, duration=100, allow_seek=True)
492 buf = AudioBuffer(TEST_PCM_FORMAT, buffer_size=BufferSize.MINIMAL)
493 streamdetails.buffer = buf
494 buf.fill(_slow_to_finalize_source(), source_name=streamdetails.uri)
495
496 with caplog.at_level(logging.WARNING):
497 waiter = asyncio.create_task(
498 buf._wait_until_ready(streamdetails, 0.05, "get_buffer[prepare_next]")
499 )
500 await asyncio.sleep(0)
501 assert buf._producer_task is not None
502 buf._producer_task.cancel()
503 # the deadline expires while the source is still inside its cleanup
504 await asyncio.sleep(0.1)
505 finalized.set()
506 with pytest.raises(AudioError, match="Timeout waiting for audio data"):
507 await waiter
508
509 assert "Gave up on" not in caplog.text
510
511
512@pytest.mark.asyncio
513@pytest.mark.parametrize("media_type", [MediaType.SOUND_EFFECT, MediaType.AUDIO_SOURCE])
514async def test_get_buffer_skips_analysis_for_non_analyzed_types(media_type: MediaType) -> None:
515 """get_buffer skips audio analysis for sound effects and audio sources."""
516 mass, start_analysis, scheduled_tasks = _make_mass_for_get_buffer()
517 streamdetails = _make_stream_details(
518 media_type,
519 duration=30 if media_type == MediaType.SOUND_EFFECT else None,
520 allow_seek=media_type == MediaType.SOUND_EFFECT,
521 )
522
523 buffer = await AudioBuffer.get_buffer(mass, streamdetails, reason="test")
524
525 assert scheduled_tasks == []
526 start_analysis.assert_not_called()
527 await buffer.clear()
528
529
530@pytest.mark.asyncio
531async def test_get_buffer_still_starts_analysis_for_track() -> None:
532 """get_buffer still schedules audio analysis for tracks."""
533 mass, start_analysis, scheduled_tasks = _make_mass_for_get_buffer()
534 streamdetails = _make_stream_details(MediaType.TRACK, duration=180, allow_seek=True)
535
536 buffer = await AudioBuffer.get_buffer(mass, streamdetails, reason="test")
537
538 assert len(scheduled_tasks) == 1
539 await asyncio.gather(*scheduled_tasks)
540 start_analysis.assert_awaited_once()
541 await buffer.clear()
542
543
544@pytest.mark.asyncio
545async def test_get_buffer_sound_effect_uses_default_ready_threshold_without_crossfade() -> None:
546 """Sound effects should not use the larger crossfade buffering threshold."""
547 mass, start_analysis, scheduled_tasks = _make_mass_for_get_buffer(
548 queue=SimpleNamespace(crossfade_enabled=True)
549 )
550 streamdetails = _make_stream_details(
551 MediaType.SOUND_EFFECT,
552 duration=30,
553 allow_seek=True,
554 queue_id="queue-1",
555 )
556
557 buffer = await AudioBuffer.get_buffer(mass, streamdetails, reason="test")
558
559 assert buffer._ready_at_chunk == 2
560 assert scheduled_tasks == []
561 start_analysis.assert_not_called()
562 await buffer.clear()
563
564
565@pytest.mark.parametrize(
566 ("max_concurrent_streams", "has_free_slot", "expect_released"),
567 [(1, False, True), (1, True, False), (None, True, False)],
568 ids=["slot_limited_saturated", "slot_limited_with_free_slot", "unlimited"],
569)
570@pytest.mark.asyncio
571async def test_get_buffer_releases_a_slot_limited_producer_before_replacing_it(
572 max_concurrent_streams: int | None, has_free_slot: bool, expect_released: bool
573) -> None:
574 """The superseded producer only gives up its slot when the provider has none to spare."""
575 mass, _start_analysis, _scheduled_tasks = _make_mass_for_get_buffer()
576 provider = MagicMock(spec=MusicProvider)
577 provider.max_concurrent_streams = max_concurrent_streams
578 provider.has_available_stream_slot = has_free_slot
579 mass.get_provider.return_value = provider
580 streamdetails = _make_stream_details(MediaType.TRACK, duration=600, allow_seek=True)
581 blocked = asyncio.Event()
582
583 async def _never_ending_source() -> AsyncGenerator[bytes]:
584 yield _make_chunk(0)
585 await blocked.wait()
586
587 stale_buffer = AudioBuffer(TEST_PCM_FORMAT)
588 stale_buffer.fill(_never_ending_source())
589 await asyncio.sleep(0)
590 await asyncio.sleep(0)
591 streamdetails.buffer = stale_buffer
592 # the producer is still charging a source slot and the consumer is active right now,
593 # so the 30s inactivity heuristic must not be what decides this
594 assert stale_buffer.is_buffering
595 assert time.time() - stale_buffer._last_access_time < 30
596
597 # a forward seek far past the buffered window can not be served by this buffer
598 assert not stale_buffer.is_valid((SEEK_WAIT_THRESHOLD + 60) * 1000)
599 replacement = await AudioBuffer.get_buffer(
600 mass,
601 streamdetails,
602 seek_position_ms=(SEEK_WAIT_THRESHOLD + 60) * 1000,
603 reason="test",
604 )
605
606 assert replacement is not stale_buffer
607 assert stale_buffer.cancelled is expect_released
608 assert stale_buffer.is_buffering is not expect_released
609 blocked.set()
610 await stale_buffer.clear()
611 await replacement.clear()
612
613
614@pytest.mark.asyncio
615async def test_get_buffer_replaces_a_live_buffer_on_a_short_forward_seek() -> None:
616 """A live source is restarted at the position rather than waited out."""
617 mass, _start_analysis, _scheduled_tasks = _make_mass_for_get_buffer()
618 provider = MagicMock(spec=MusicProvider)
619 provider.max_concurrent_streams = None
620 provider.has_available_stream_slot = True
621 mass.get_provider.return_value = provider
622 streamdetails = _make_stream_details(MediaType.TRACK, duration=600, allow_seek=True)
623 streamdetails.is_realtime = True
624 blocked = asyncio.Event()
625
626 async def _never_ending_source() -> AsyncGenerator[bytes]:
627 yield _make_chunk(0)
628 await blocked.wait()
629
630 live_buffer = AudioBuffer(TEST_PCM_FORMAT, is_realtime=True)
631 live_buffer.fill(_never_ending_source())
632 await asyncio.sleep(0)
633 await asyncio.sleep(0)
634 streamdetails.buffer = live_buffer
635
636 # well inside the span a recorded source would have been waited out for
637 replacement = await AudioBuffer.get_buffer(
638 mass, streamdetails, seek_position_ms=5000, reason="test"
639 )
640
641 assert replacement is not live_buffer
642 # the replacement carries the flag on, and starts at the position asked for
643 assert replacement.is_realtime
644 assert replacement._discarded_chunks == 5
645 blocked.set()
646 await live_buffer.clear()
647 await replacement.clear()
648
649
650@pytest.mark.asyncio
651async def test_fill_closes_source_on_cancel() -> None:
652 """The source generator is finalized immediately when the fill task is cancelled."""
653 source_closed = asyncio.Event()
654
655 async def _endless_source() -> AsyncGenerator[bytes]:
656 try:
657 while True:
658 yield ONE_SECOND_CHUNK
659 finally:
660 source_closed.set()
661
662 buf = AudioBuffer(TEST_PCM_FORMAT, buffer_size=BufferSize.MINIMAL)
663 buf.fill(_endless_source(), source_name="test")
664 # let the fill task run until it blocks on the full buffer
665 await asyncio.sleep(0.1)
666
667 # clear() cancels the fill task, which must close the source generator
668 await buf.clear()
669 await asyncio.wait_for(source_closed.wait(), timeout=1)
670
671
672# -- Seek and is_valid --
673
674
675@pytest.mark.asyncio
676async def test_is_valid_basic() -> None:
677 """is_valid returns True for buffered positions."""
678 buf = AudioBuffer(TEST_PCM_FORMAT, buffer_size=BufferSize.MINIMAL)
679 for _ in range(10):
680 await buf._put(ONE_SECOND_CHUNK)
681
682 assert buf.is_valid(seek_position_ms=0)
683 assert buf.is_valid(seek_position_ms=5000)
684 assert buf.is_valid(seek_position_ms=9000)
685
686
687@pytest.mark.asyncio
688async def test_is_valid_cancelled() -> None:
689 """is_valid returns False for cancelled buffer."""
690 buf = AudioBuffer(TEST_PCM_FORMAT)
691 await buf._put(ONE_SECOND_CHUNK)
692 await buf.clear()
693 assert not buf.is_valid()
694
695
696@pytest.mark.asyncio
697async def test_is_valid_seek_ahead_within_threshold() -> None:
698 """is_valid returns True when seek is within SEEK_WAIT_THRESHOLD of buffered data."""
699 buf = AudioBuffer(TEST_PCM_FORMAT, buffer_size=BufferSize.MINIMAL)
700 for _ in range(10):
701 await buf._put(ONE_SECOND_CHUNK)
702
703 # 10 chunks buffered, seek to 10+SEEK_WAIT_THRESHOLD seconds should be valid
704 seek_ms = (10 + SEEK_WAIT_THRESHOLD) * 1000
705 assert buf.is_valid(seek_position_ms=seek_ms)
706
707
708@pytest.mark.asyncio
709async def test_is_valid_seek_ahead_beyond_threshold() -> None:
710 """is_valid returns False when seek is beyond SEEK_WAIT_THRESHOLD."""
711 buf = AudioBuffer(TEST_PCM_FORMAT, buffer_size=BufferSize.MINIMAL)
712 for _ in range(10):
713 await buf._put(ONE_SECOND_CHUNK)
714
715 seek_ms = (10 + SEEK_WAIT_THRESHOLD + 1) * 1000
716 assert not buf.is_valid(seek_position_ms=seek_ms)
717
718
719@pytest.mark.asyncio
720async def test_is_valid_with_eof() -> None:
721 """is_valid returns True for any position when EOF is received."""
722 buf = AudioBuffer(TEST_PCM_FORMAT, buffer_size=BufferSize.MINIMAL)
723 for _ in range(5):
724 await buf._put(ONE_SECOND_CHUNK)
725 await buf._set_eof()
726
727 # even beyond buffered data, is_valid returns True with EOF
728 assert buf.is_valid(seek_position_ms=100_000)
729
730
731@pytest.mark.asyncio
732async def test_seek_in_raw_stream() -> None:
733 """get_raw_stream with seek_position_ms skips to correct chunk."""
734 buf = AudioBuffer(TEST_PCM_FORMAT, buffer_size=BufferSize.MINIMAL)
735 buf.fill(_make_source(10), source_name="test")
736 await asyncio.sleep(0.1)
737
738 chunks = []
739 async for chunk in buf.get_raw_stream(seek_position_ms=5000):
740 chunks.append(chunk)
741
742 assert len(chunks) == 5
743 # first chunk should be chunk #5
744 assert chunks[0] == _make_chunk(5)
745
746
747async def test_exact_raw_seek_preserves_millisecond_position() -> None:
748 """Crossfade continuation does not round its media-time resume backward."""
749 audio_buffer = AudioBuffer(TEST_PCM_FORMAT)
750 await audio_buffer._put(ONE_SECOND_CHUNK)
751 await audio_buffer._set_eof()
752
753 regular_stream = audio_buffer.get_raw_stream(seek_position_ms=250)
754 exact_stream = audio_buffer.get_raw_stream(seek_position_ms=250, exact_seek=True)
755 regular_chunk = await anext(regular_stream)
756 exact_chunk = await anext(exact_stream)
757 await regular_stream.aclose()
758 await exact_stream.aclose()
759
760 assert len(regular_chunk) == int(len(ONE_SECOND_CHUNK) * 0.8)
761 assert len(exact_chunk) == int(len(ONE_SECOND_CHUNK) * 0.75)
762
763
764# -- Analysis reader (read_chunk_for_analysis) --
765
766
767@pytest.mark.asyncio
768async def test_read_chunk_for_analysis_returns_buffered_chunk() -> None:
769 """A passive reader gets a retained chunk without discarding it."""
770 buf = AudioBuffer(TEST_PCM_FORMAT, buffer_size=BufferSize.MINIMAL)
771 await buf._put(_make_chunk(0))
772 await buf._put(_make_chunk(1))
773
774 assert await buf.read_chunk_for_analysis(0) == _make_chunk(0)
775 assert await buf.read_chunk_for_analysis(1) == _make_chunk(1)
776 # Reading must not have discarded anything — both chunks are still buffered.
777 assert buf.seconds_available == 2
778 assert buf.first_buffered_chunk == 0
779
780
781@pytest.mark.asyncio
782async def test_read_chunk_for_analysis_waits_then_returns() -> None:
783 """A reader ahead of the filled position waits until the chunk is produced."""
784 buf = AudioBuffer(TEST_PCM_FORMAT, buffer_size=BufferSize.MINIMAL)
785 reader = asyncio.ensure_future(buf.read_chunk_for_analysis(0))
786 await asyncio.sleep(0.05)
787 assert not reader.done() # nothing buffered yet
788
789 await buf._put(_make_chunk(0))
790 assert await asyncio.wait_for(reader, timeout=1.0) == _make_chunk(0)
791
792
793@pytest.mark.asyncio
794async def test_read_chunk_for_analysis_raises_eof_past_end() -> None:
795 """Reading past the last chunk of an ended stream raises AudioBufferEOF."""
796 buf = AudioBuffer(TEST_PCM_FORMAT, buffer_size=BufferSize.MINIMAL)
797 await buf._put(_make_chunk(0))
798 await buf._set_eof()
799
800 assert await buf.read_chunk_for_analysis(0) == _make_chunk(0)
801 with pytest.raises(AudioBufferEOF):
802 await buf.read_chunk_for_analysis(1)
803
804
805@pytest.mark.asyncio
806async def test_read_chunk_for_analysis_raises_discarded_when_evicted() -> None:
807 """Requesting a chunk that has been evicted from the window raises AudioBufferDiscarded."""
808 buf = AudioBuffer(TEST_PCM_FORMAT, buffer_size=BufferSize.MINIMAL)
809 await buf._put(_make_chunk(0))
810 # Simulate the playback consumer sliding the window past chunk 0.
811 buf._chunks.popleft()
812 buf._discarded_chunks += 1
813 assert buf.first_buffered_chunk == 1
814
815 with pytest.raises(AudioBufferDiscarded):
816 await buf.read_chunk_for_analysis(0)
817
818
819@pytest.mark.asyncio
820async def test_read_chunk_for_analysis_raises_discarded_on_clear() -> None:
821 """A reader blocked on a torn-down buffer is released with AudioBufferDiscarded."""
822 buf = AudioBuffer(TEST_PCM_FORMAT, buffer_size=BufferSize.MINIMAL)
823 reader = asyncio.ensure_future(buf.read_chunk_for_analysis(0))
824 await asyncio.sleep(0.05)
825 await buf.clear()
826 with pytest.raises(AudioBufferDiscarded):
827 await asyncio.wait_for(reader, timeout=1.0)
828
829
830# -- Buffer size limits --
831
832
833@pytest.mark.asyncio
834async def test_rolling_buffer_fifo() -> None:
835 """ROLLING mode works as a FIFO — get pops the oldest chunk."""
836 buf = AudioBuffer(TEST_PCM_FORMAT, mode=BufferMode.ROLLING)
837
838 for i in range(5):
839 await buf._put(_make_chunk(i))
840
841 assert buf.size_seconds == 5
842
843 # get pops the oldest chunk and frees space
844 result = await buf._get(chunk_number=0)
845 assert result == _make_chunk(0)
846 assert buf.size_seconds == 4
847 assert buf._discarded_chunks == 1
848
849 # next get returns the next chunk
850 result = await buf._get(chunk_number=1)
851 assert result == _make_chunk(1)
852 assert buf.size_seconds == 3
853 assert buf._discarded_chunks == 2
854
855
856@pytest.mark.asyncio
857async def test_rolling_buffer_drained_surfaces_producer_error() -> None:
858 """A drained rolling buffer raises the producer error instead of a clean EOF."""
859
860 async def _failing_source() -> AsyncGenerator[bytes]:
861 yield ONE_SECOND_CHUNK
862 msg = "test error"
863 raise RuntimeError(msg)
864
865 buf = AudioBuffer(TEST_PCM_FORMAT, mode=BufferMode.ROLLING)
866 buf.fill(_failing_source(), source_name="test")
867 while not buf.has_error:
868 await asyncio.sleep(0.01)
869
870 # the buffered chunk is still delivered before the error surfaces
871 assert await buf._get() == ONE_SECOND_CHUNK
872 with pytest.raises(RuntimeError, match="test error"):
873 await buf._get()
874
875
876@pytest.mark.asyncio
877async def test_seekable_buffer_backpressure() -> None:
878 """SEEKABLE mode waits on put when full, consumer frees space on get."""
879 buf = AudioBuffer(TEST_PCM_FORMAT, buffer_size=BufferSize.MINIMAL)
880 max_size = buf.max_size_seconds
881
882 # use fill() so there's an active producer task (eviction only happens
883 # when the producer is running and needs space)
884 buf.fill(_make_source(max_size + 5), source_name="test")
885 async with asyncio.timeout(5):
886 async with buf._data_available:
887 await buf._data_available.wait_for(lambda: buf.size_seconds == max_size)
888
889 assert buf.size_seconds == max_size
890
891 # reading from a full buffer frees space for the producer
892 chunk = await buf._get(chunk_number=0)
893 assert chunk == _make_chunk(0)
894 assert buf._discarded_chunks == 1
895
896
897@pytest.mark.asyncio
898async def test_seekable_no_eviction_after_eof() -> None:
899 """After EOF, reads from a full buffer do not evict chunks."""
900 buf = AudioBuffer(TEST_PCM_FORMAT, buffer_size=BufferSize.MINIMAL)
901 max_size = buf.max_size_seconds
902
903 buf.fill(_make_source(max_size), source_name="test")
904 await asyncio.sleep(0.1)
905
906 assert buf._eof_received
907 assert buf.size_seconds == max_size
908
909 # read should NOT evict since producer is done
910 chunk = await buf._get(chunk_number=0)
911 assert chunk == _make_chunk(0)
912 assert buf._discarded_chunks == 0
913 assert buf.size_seconds == max_size
914
915
916# -- get_stream passthrough --
917
918
919@pytest.mark.asyncio
920async def test_get_stream_no_filters() -> None:
921 """get_stream without filters passes through raw data."""
922 buf = AudioBuffer(TEST_PCM_FORMAT, buffer_size=BufferSize.MINIMAL)
923 buf.fill(_make_source(3), source_name="test")
924 await asyncio.sleep(0.1)
925
926 chunks = []
927 async for chunk in buf.get_stream(output_format=TEST_PCM_FORMAT):
928 chunks.append(chunk)
929
930 assert len(chunks) == 3
931 assert chunks[0] == _make_chunk(0)
932
933
934# -- Rolling mode --
935
936
937@pytest.mark.asyncio
938async def test_rolling_mode_max_size() -> None:
939 """ROLLING mode uses RADIO_BUFFER_SIZE."""
940 buf = AudioBuffer(TEST_PCM_FORMAT, mode=BufferMode.ROLLING)
941 assert buf.max_size_seconds == RADIO_BUFFER_SIZE
942
943
944# -- Ready threshold with seek offset --
945
946
947@pytest.mark.asyncio
948async def test_ready_accounts_for_seek_offset() -> None:
949 """Ready fires only after enough data past the seek point is buffered."""
950 buf = AudioBuffer(TEST_PCM_FORMAT, ready_threshold=3)
951 # simulate get_buffer setting the offset for a seek to 100s
952 buf._discarded_chunks = 100
953 buf._ready_at_chunk = 100 + 3 # seek_chunk + threshold
954
955 await buf._put(ONE_SECOND_CHUNK) # chunk 100
956 assert not buf.ready.is_set()
957 await buf._put(ONE_SECOND_CHUNK) # chunk 101
958 assert not buf.ready.is_set()
959 await buf._put(ONE_SECOND_CHUNK) # chunk 102
960 assert buf.ready.is_set()
961
962
963@pytest.mark.asyncio
964async def test_chunk_numbering_with_seek_offset() -> None:
965 """Chunks are numbered correctly when buffer starts at a seek offset."""
966 buf = AudioBuffer(TEST_PCM_FORMAT, buffer_size=BufferSize.MINIMAL)
967 # simulate a buffer created for a seek to 300s
968 buf._discarded_chunks = 300
969
970 for i in range(5):
971 await buf._put(_make_chunk(i))
972
973 # chunk 300 should be the first chunk (value 0)
974 result = await buf._get(chunk_number=300)
975 assert result == _make_chunk(0)
976 # chunk 304 should be the fifth chunk (value 4)
977 result = await buf._get(chunk_number=304)
978 assert result == _make_chunk(4)
979
980
981@pytest.mark.asyncio
982async def test_is_valid_with_seek_offset() -> None:
983 """is_valid works correctly with a seek offset."""
984 buf = AudioBuffer(TEST_PCM_FORMAT, buffer_size=BufferSize.MINIMAL)
985 buf._discarded_chunks = 300
986
987 for _ in range(10):
988 await buf._put(ONE_SECOND_CHUNK)
989
990 # positions before the offset are invalid (discarded)
991 assert not buf.is_valid(seek_position_ms=299_000)
992 # positions within the buffer are valid
993 assert buf.is_valid(seek_position_ms=300_000)
994 assert buf.is_valid(seek_position_ms=305_000)
995
996
997@pytest.mark.asyncio
998async def test_raw_stream_with_seek_offset() -> None:
999 """get_raw_stream works correctly when buffer has a seek offset."""
1000 buf = AudioBuffer(TEST_PCM_FORMAT, buffer_size=BufferSize.MINIMAL)
1001 buf._discarded_chunks = 300
1002
1003 for i in range(5):
1004 await buf._put(_make_chunk(i))
1005 await buf._set_eof()
1006
1007 chunks = []
1008 async for chunk in buf.get_raw_stream(seek_position_ms=300_000):
1009 chunks.append(chunk)
1010
1011 assert len(chunks) == 5
1012 assert chunks[0] == _make_chunk(0)
1013 assert chunks[4] == _make_chunk(4)
1014
1015
1016# -- Callback error isolation --
1017
1018
1019@pytest.mark.asyncio
1020async def test_clear_fires_cancel_callbacks() -> None:
1021 """clear() fires registered cancel callbacks before removing them."""
1022 cancel_called = False
1023
1024 def _cancel_callback() -> None:
1025 nonlocal cancel_called
1026 cancel_called = True
1027
1028 buf = AudioBuffer(TEST_PCM_FORMAT, buffer_size=BufferSize.MINIMAL)
1029 buf.register_cancel_callback(_cancel_callback)
1030 await buf._put(ONE_SECOND_CHUNK)
1031
1032 await buf.clear()
1033 assert cancel_called is True
1034 assert len(buf._cancel_callbacks) == 0
1035
1036
1037# -- Inactivity monitor --
1038
1039
1040@pytest.mark.asyncio
1041async def test_inactivity_monitor_releases_drained_buffer() -> None:
1042 """
1043 A buffer that has drained to empty is still released by the inactivity monitor.
1044
1045 Regression test: the monitor previously only cleared when chunks remained, so an
1046 abandoned rolling buffer that drained to zero chunks looped forever and leaked it
1047 (and its producer/ffmpeg) until the process exited.
1048 """
1049 buf = AudioBuffer(TEST_PCM_FORMAT, mode=BufferMode.ROLLING)
1050 # no chunks buffered and last access long ago -> the buffer is inactive
1051 assert buf.size_seconds == 0
1052 buf._last_access_time = time.time() - 10_000
1053
1054 await buf._monitor_inactivity(inactivity_timeout=0.01, check_interval=0.01)
1055
1056 assert buf.cancelled is True
1057
1058
1059@pytest.mark.asyncio
1060async def test_inactivity_monitor_keeps_active_buffer() -> None:
1061 """A buffer that is still being accessed is not cleared by the inactivity monitor."""
1062 buf = AudioBuffer(TEST_PCM_FORMAT, mode=BufferMode.ROLLING)
1063 buf._last_access_time = time.time()
1064
1065 monitor = asyncio.create_task(
1066 buf._monitor_inactivity(inactivity_timeout=5, check_interval=0.01)
1067 )
1068 await asyncio.sleep(0.05)
1069
1070 assert not monitor.done()
1071 assert buf.cancelled is False
1072
1073 monitor.cancel()
1074 with suppress(asyncio.CancelledError):
1075 await monitor
1076
1077
1078# -- Pre-buffering of the next queue item --
1079
1080
1081@pytest.fixture
1082async def mass_minimal(mass_minimal: MusicAssistant) -> MusicAssistant:
1083 """Extend the base fixture with the player_queues/streams stand-ins get_queue_item_stream needs."""
1084 mass_minimal.player_queues = SimpleNamespace( # type: ignore[assignment]
1085 get_active_queue=lambda _queue_id: None,
1086 prepare_next_audio_buffer=lambda _queue_id: None,
1087 queue_data_or_none=lambda _queue_id: None,
1088 )
1089 mass_minimal.streams = MagicMock()
1090 return mass_minimal
1091
1092
1093class _FakeAudioBuffer:
1094 """AudioBuffer test double that streams a fixed run of 1-second chunks."""
1095
1096 has_error = False
1097 pcm_format = TEST_PCM_FORMAT
1098
1099 @classmethod
1100 async def get_buffer(cls, **_kwargs: Any) -> _FakeAudioBuffer:
1101 return cls()
1102
1103 async def get_stream(self, **_kwargs: Any) -> AsyncGenerator[bytes]:
1104 async for chunk in _make_source(90):
1105 yield chunk
1106
1107
1108async def _stream_until_prebuffer_window(
1109 mass: MusicAssistant,
1110 *,
1111 next_item_media_type: MediaType,
1112 queue_id: str,
1113 is_realtime: bool = False,
1114) -> None:
1115 """
1116 Drive get_queue_item_stream for a 90s current TRACK item past the pre-buffer trigger point.
1117
1118 Sets up a queue whose next item has ``next_item_media_type`` and streams the current
1119 item to completion, so the pre-buffer trigger condition (evaluated once more than
1120 duration - 60 seconds of PCM has been yielded) gets a chance to fire.
1121
1122 :param is_realtime: Whether the current item's source hands over its audio
1123 just-in-time, which moves the trigger to the source itself.
1124 """
1125 streamdetails = _make_stream_details(MediaType.TRACK, duration=90, allow_seek=True)
1126 streamdetails.is_realtime = is_realtime
1127 streamdetails.loudness = -10.0 # skip the audio-analysis hydration call
1128 current_item = QueueItem(
1129 queue_id=queue_id,
1130 queue_item_id="current",
1131 name="Current",
1132 duration=90,
1133 streamdetails=streamdetails,
1134 )
1135 next_item = SimpleNamespace(queue_item_id="next", media_type=next_item_media_type)
1136 queue = SimpleNamespace(next_item=next_item)
1137 mass.player_queues.get_active_queue = lambda _player_id: queue # type: ignore[method-assign, assignment, return-value]
1138
1139 controller = StreamsAudio(mass)
1140 with patch.object(audio_mod, "AudioBuffer", _FakeAudioBuffer):
1141 async for _chunk in controller.get_queue_item_stream(current_item, TEST_PCM_FORMAT):
1142 pass
1143
1144
1145@pytest.mark.asyncio
1146async def test_sound_effect_next_item_triggers_prebuffer(mass_minimal: MusicAssistant) -> None:
1147 """A SOUND_EFFECT next item is pre-buffered like a track."""
1148 calls: list[str] = []
1149 mass_minimal.player_queues.prepare_next_audio_buffer = ( # type: ignore[method-assign]
1150 lambda queue_id: calls.append(queue_id)
1151 )
1152
1153 await _stream_until_prebuffer_window(
1154 mass_minimal, next_item_media_type=MediaType.SOUND_EFFECT, queue_id="player_a"
1155 )
1156
1157 assert calls == ["player_a"]
1158
1159
1160@pytest.mark.asyncio
1161async def test_audio_source_next_item_is_not_prebuffered(mass_minimal: MusicAssistant) -> None:
1162 """A live AUDIO_SOURCE next item is still excluded from pre-buffering."""
1163 calls: list[str] = []
1164 mass_minimal.player_queues.prepare_next_audio_buffer = ( # type: ignore[method-assign]
1165 lambda queue_id: calls.append(queue_id)
1166 )
1167
1168 await _stream_until_prebuffer_window(
1169 mass_minimal, next_item_media_type=MediaType.AUDIO_SOURCE, queue_id="player_a"
1170 )
1171
1172 assert calls == []
1173
1174
1175@pytest.mark.asyncio
1176async def test_realtime_source_leaves_the_prebuffer_to_the_source(
1177 mass_minimal: MusicAssistant,
1178) -> None:
1179 """A realtime source triggers the next item itself, so the blind trigger stays quiet."""
1180 calls: list[str] = []
1181 mass_minimal.player_queues.prepare_next_audio_buffer = ( # type: ignore[method-assign]
1182 lambda queue_id: calls.append(queue_id)
1183 )
1184
1185 await _stream_until_prebuffer_window(
1186 mass_minimal,
1187 next_item_media_type=MediaType.TRACK,
1188 queue_id="player_a",
1189 is_realtime=True,
1190 )
1191
1192 # the next item's audio does not exist yet while this one plays, so triggering here
1193 # would only open a source that times out and gets discarded
1194 assert calls == []
1195
1196
1197@pytest.mark.asyncio
1198async def test_real_buffer_producer_error_reaches_queue_item_stream(
1199 mass_minimal: MusicAssistant,
1200) -> None:
1201 """A real AudioBuffer producer error is surfaced instead of a truncated stream."""
1202
1203 async def _failing_source() -> AsyncGenerator[bytes]:
1204 yield ONE_SECOND_CHUNK
1205 raise RuntimeError("source failed")
1206
1207 streamdetails = _make_stream_details(MediaType.SOUND_EFFECT, duration=90, allow_seek=True)
1208 streamdetails.loudness = -10.0
1209 queue_item = QueueItem(
1210 queue_id="player_a",
1211 queue_item_id="current",
1212 name="Current",
1213 duration=90,
1214 streamdetails=streamdetails,
1215 )
1216 cast("Any", mass_minimal.player_queues).get = MagicMock(return_value=None)
1217 cast("Any", mass_minimal.streams.audio).get_media_stream = MagicMock(
1218 return_value=_failing_source()
1219 )
1220 controller = StreamsAudio(mass_minimal)
1221
1222 chunks: list[bytes] = []
1223 async for chunk in controller.get_queue_item_stream(
1224 queue_item, TEST_PCM_FORMAT, raise_on_error=False
1225 ):
1226 chunks.append(chunk)
1227
1228 # the stream waits for the buffer to become playable, so a producer failure is
1229 # reported before any audio is served rather than truncating it mid-stream
1230 assert chunks == []
1231 assert streamdetails.stream_error is True
1232 assert queue_item.available
1233
1234
1235@pytest.mark.asyncio
1236async def test_stale_stream_error_reset_on_stream_start(mass_minimal: MusicAssistant) -> None:
1237 """A stream_error left on reused streamdetails is cleared when a new stream starts."""
1238 streamdetails = _make_stream_details(MediaType.TRACK, duration=90, allow_seek=True)
1239 streamdetails.loudness = -10.0 # skip the audio-analysis hydration call
1240 streamdetails.stream_error = True # left over from a previously failed attempt
1241 queue_item = QueueItem(
1242 queue_id="player_a",
1243 queue_item_id="current",
1244 name="Current",
1245 duration=90,
1246 streamdetails=streamdetails,
1247 )
1248 controller = StreamsAudio(mass_minimal)
1249
1250 with patch.object(audio_mod, "AudioBuffer", _FakeAudioBuffer):
1251 async for _chunk in controller.get_queue_item_stream(queue_item, TEST_PCM_FORMAT):
1252 pass
1253
1254 assert streamdetails.stream_error is False
1255
1256
1257@pytest.mark.asyncio
1258async def test_audio_source_stream_error_reset_on_retry(mass_minimal: MusicAssistant) -> None:
1259 """A cached AudioSource stream clears a prior error before retrying."""
1260 streamdetails = _make_stream_details(MediaType.AUDIO_SOURCE, duration=None, allow_seek=False)
1261 streamdetails.stream_error = True
1262 queue_item = QueueItem(
1263 queue_id="player_a",
1264 queue_item_id="source",
1265 name="Source",
1266 duration=0,
1267 streamdetails=streamdetails,
1268 )
1269 controller = StreamsAudio(mass_minimal)
1270
1271 async def _source(
1272 _streamdetails: StreamDetails, _pcm_format: AudioFormat
1273 ) -> AsyncGenerator[bytes]:
1274 yield ONE_SECOND_CHUNK
1275
1276 with patch.object(controller, "_iter_audio_source_pcm", _source):
1277 chunks = [
1278 chunk async for chunk in controller.get_queue_item_stream(queue_item, TEST_PCM_FORMAT)
1279 ]
1280
1281 assert chunks == [ONE_SECOND_CHUNK]
1282 assert streamdetails.stream_error is False
1283