music-assistant-server

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_reports_completion_only_on_a_clean_eof() -> None:
308    """on_complete fires when the source delivered everything, never on a failure."""
309    completed: list[str] = []
310
311    buf = AudioBuffer(TEST_PCM_FORMAT, buffer_size=BufferSize.MINIMAL)
312    buf.fill(_make_source(2), source_name="test", on_complete=lambda: completed.append("clean"))
313    await asyncio.sleep(0.1)
314    assert buf._eof_received
315    assert completed == ["clean"]
316
317    async def _failing_source() -> AsyncGenerator[bytes]:
318        yield ONE_SECOND_CHUNK
319        msg = "test error"
320        raise RuntimeError(msg)
321
322    failing = AudioBuffer(TEST_PCM_FORMAT, buffer_size=BufferSize.MINIMAL)
323    failing.fill(
324        _failing_source(), source_name="test", on_complete=lambda: completed.append("failed")
325    )
326    await asyncio.sleep(0.1)
327    assert completed == ["clean"]
328
329
330@pytest.mark.asyncio
331async def test_fill_does_not_report_completion_on_cancellation() -> None:
332    """A cancelled fill never claims its source delivered everything."""
333    completed: list[str] = []
334
335    async def _endless_source() -> AsyncGenerator[bytes]:
336        while True:
337            yield ONE_SECOND_CHUNK
338
339    buf = AudioBuffer(TEST_PCM_FORMAT, buffer_size=BufferSize.MINIMAL)
340    buf.fill(_endless_source(), source_name="test", on_complete=lambda: completed.append("x"))
341    await asyncio.sleep(0.05)
342    await buf.clear()
343    assert completed == []
344
345
346@pytest.mark.asyncio
347async def test_fill_error_propagation() -> None:
348    """When the source errors after producing data, valid chunks are still delivered."""
349
350    async def _failing_source() -> AsyncGenerator[bytes]:
351        yield ONE_SECOND_CHUNK
352        msg = "test error"
353        raise RuntimeError(msg)
354
355    buf = AudioBuffer(TEST_PCM_FORMAT, buffer_size=BufferSize.MINIMAL)
356    buf.fill(_failing_source(), source_name="test")
357    await asyncio.sleep(0.1)
358
359    assert buf.has_error
360
361    # consumer should receive the valid chunk before the source error surfaces.
362    result: list[bytes] = []
363
364    async def _consume() -> None:
365        async for chunk in buf.get_raw_stream():
366            result.append(chunk)
367
368    with pytest.raises(RuntimeError, match="test error"):
369        await _consume()
370    assert result == [ONE_SECOND_CHUNK]
371
372
373@pytest.mark.asyncio
374async def test_fill_error_surfaces_to_analysis_reader() -> None:
375    """An aborted source raises its error to the analysis reader instead of a clean EOF."""
376
377    async def _failing_source() -> AsyncGenerator[bytes]:
378        yield ONE_SECOND_CHUNK
379        msg = "test error"
380        raise RuntimeError(msg)
381
382    buf = AudioBuffer(TEST_PCM_FORMAT, buffer_size=BufferSize.MINIMAL)
383    buf.fill(_failing_source(), source_name="test")
384
385    # buffered chunks are still delivered before the error surfaces
386    assert await buf.read_chunk_for_analysis(0) == ONE_SECOND_CHUNK
387    with pytest.raises(RuntimeError, match="test error"):
388        await buf.read_chunk_for_analysis(1)
389
390
391@pytest.mark.asyncio
392async def test_fill_error_no_data() -> None:
393    """When the source errors without producing any data, the error propagates."""
394
395    async def _failing_source() -> AsyncGenerator[bytes]:
396        msg = "test error"
397        raise RuntimeError(msg)
398        yield  # type: ignore[unreachable]
399
400    buf = AudioBuffer(TEST_PCM_FORMAT, buffer_size=BufferSize.MINIMAL)
401    buf.fill(_failing_source(), source_name="test")
402    await asyncio.sleep(0.1)
403
404    assert buf.has_error
405
406    async def _consume() -> list[bytes]:
407        result = []
408        async for chunk in buf.get_raw_stream():
409            result.append(chunk)
410        return result
411
412    with pytest.raises(RuntimeError, match="test error"):
413        await _consume()
414
415
416async def _silent_source() -> AsyncGenerator[bytes]:
417    """Create an async generator that never delivers audio."""
418    await asyncio.sleep(10)
419    yield ONE_SECOND_CHUNK
420
421
422@pytest.mark.asyncio
423async def test_logs_time_to_first_playable_audio(caplog: pytest.LogCaptureFixture) -> None:
424    """A buffer reports how long its first playable audio took."""
425    buf = AudioBuffer(TEST_PCM_FORMAT, buffer_size=BufferSize.MINIMAL)
426    with caplog.at_level(logging.DEBUG, logger="music_assistant.audio_buffer"):
427        buf.fill(_make_source(2), source_name="test://item")
428        await buf.ready.wait()
429
430    assert "test://item became ready after" in caplog.text
431
432
433@pytest.mark.asyncio
434async def test_logs_time_to_ready_for_a_stream_below_its_threshold(
435    caplog: pytest.LogCaptureFixture,
436) -> None:
437    """A stream that ends before reaching its ready threshold still reports its startup time."""
438    buf = AudioBuffer(TEST_PCM_FORMAT, buffer_size=BufferSize.MINIMAL, ready_threshold=8)
439    with caplog.at_level(logging.DEBUG, logger="music_assistant.audio_buffer"):
440        buf.fill(_make_source(2), source_name="test://short")
441        await buf.ready.wait()
442
443    assert "test://short became ready after" in caplog.text
444
445
446@pytest.mark.asyncio
447async def test_empty_stream_is_not_reported_as_ready(caplog: pytest.LogCaptureFixture) -> None:
448    """A source that ends without delivering audio never became playable."""
449    buf = AudioBuffer(TEST_PCM_FORMAT, buffer_size=BufferSize.MINIMAL)
450    with caplog.at_level(logging.DEBUG, logger="music_assistant.audio_buffer"):
451        buf.fill(_make_source(0), source_name="test://empty")
452        await buf.ready.wait()
453
454    assert "became ready after" not in caplog.text
455
456
457@pytest.mark.asyncio
458async def test_failed_stream_below_threshold_is_not_reported_as_ready(
459    caplog: pytest.LogCaptureFixture,
460) -> None:
461    """A source that fails before reaching its threshold never became playable."""
462
463    async def _failing_source() -> AsyncGenerator[bytes]:
464        yield ONE_SECOND_CHUNK
465        msg = "test error"
466        raise RuntimeError(msg)
467
468    buf = AudioBuffer(TEST_PCM_FORMAT, buffer_size=BufferSize.MINIMAL, ready_threshold=8)
469    with caplog.at_level(logging.DEBUG, logger="music_assistant.audio_buffer"):
470        buf.fill(_failing_source(), source_name="test://failing")
471        await buf.ready.wait()
472
473    assert "became ready after" not in caplog.text
474
475
476@pytest.mark.asyncio
477async def test_ready_timeout_reports_the_wait(caplog: pytest.LogCaptureFixture) -> None:
478    """The readiness deadline reports the caller, provider and how much audio arrived."""
479    streamdetails = _make_stream_details(MediaType.TRACK, duration=100, allow_seek=True)
480    buf = AudioBuffer(TEST_PCM_FORMAT, buffer_size=BufferSize.MINIMAL)
481    streamdetails.buffer = buf
482    buf.fill(_silent_source(), source_name=streamdetails.uri)
483
484    with (
485        caplog.at_level(logging.WARNING),
486        pytest.raises(AudioError, match="Timeout waiting for audio data"),
487    ):
488        await buf._wait_until_ready(streamdetails, 0.05, "get_buffer[prepare]")
489
490    assert "get_buffer[prepare]: Gave up on builtin" in caplog.text
491    assert ", 0s buffered" in caplog.text
492    assert streamdetails.buffer is None
493
494
495@pytest.mark.asyncio
496async def test_ready_timeout_stays_quiet_for_a_released_buffer(
497    caplog: pytest.LogCaptureFixture,
498) -> None:
499    """A buffer released to free a stream slot is not reported as a provider stall."""
500    streamdetails = _make_stream_details(MediaType.TRACK, duration=100, allow_seek=True)
501    buf = AudioBuffer(TEST_PCM_FORMAT, buffer_size=BufferSize.MINIMAL)
502    streamdetails.buffer = buf
503    buf.fill(_silent_source(), source_name=streamdetails.uri)
504
505    with caplog.at_level(logging.WARNING):
506        waiter = asyncio.create_task(
507            buf._wait_until_ready(streamdetails, 0.05, "get_buffer[prepare_next]")
508        )
509        await asyncio.sleep(0)
510        await buf.clear()
511        with pytest.raises(AudioError, match="Timeout waiting for audio data"):
512            await waiter
513
514    assert "Gave up on" not in caplog.text
515
516
517@pytest.mark.asyncio
518async def test_ready_timeout_stays_quiet_while_the_source_is_finalizing(
519    caplog: pytest.LogCaptureFixture,
520) -> None:
521    """A release is not reported as a stall while the source is still cleaning up."""
522    finalized = asyncio.Event()
523
524    async def _slow_to_finalize_source() -> AsyncGenerator[bytes]:
525        try:
526            await asyncio.sleep(10)
527            yield ONE_SECOND_CHUNK
528        finally:
529            await finalized.wait()
530
531    streamdetails = _make_stream_details(MediaType.TRACK, duration=100, allow_seek=True)
532    buf = AudioBuffer(TEST_PCM_FORMAT, buffer_size=BufferSize.MINIMAL)
533    streamdetails.buffer = buf
534    buf.fill(_slow_to_finalize_source(), source_name=streamdetails.uri)
535
536    with caplog.at_level(logging.WARNING):
537        waiter = asyncio.create_task(
538            buf._wait_until_ready(streamdetails, 0.05, "get_buffer[prepare_next]")
539        )
540        await asyncio.sleep(0)
541        assert buf._producer_task is not None
542        buf._producer_task.cancel()
543        # the deadline expires while the source is still inside its cleanup
544        await asyncio.sleep(0.1)
545        finalized.set()
546        with pytest.raises(AudioError, match="Timeout waiting for audio data"):
547            await waiter
548
549    assert "Gave up on" not in caplog.text
550
551
552@pytest.mark.asyncio
553@pytest.mark.parametrize("media_type", [MediaType.SOUND_EFFECT, MediaType.AUDIO_SOURCE])
554async def test_get_buffer_skips_analysis_for_non_analyzed_types(media_type: MediaType) -> None:
555    """get_buffer skips audio analysis for sound effects and audio sources."""
556    mass, start_analysis, scheduled_tasks = _make_mass_for_get_buffer()
557    streamdetails = _make_stream_details(
558        media_type,
559        duration=30 if media_type == MediaType.SOUND_EFFECT else None,
560        allow_seek=media_type == MediaType.SOUND_EFFECT,
561    )
562
563    buffer = await AudioBuffer.get_buffer(mass, streamdetails, reason="test")
564
565    assert scheduled_tasks == []
566    start_analysis.assert_not_called()
567    await buffer.clear()
568
569
570@pytest.mark.asyncio
571async def test_get_buffer_fill_completion_prepares_the_next_item() -> None:
572    """A realtime track's finished fill frees its slot and starts the next item's fetch."""
573    mass, _start_analysis, scheduled_tasks = _make_mass_for_get_buffer()
574    streamdetails = _make_stream_details(
575        MediaType.TRACK, duration=180, allow_seek=True, queue_id="queue_a"
576    )
577    streamdetails.is_realtime = True
578
579    buffer = await AudioBuffer.get_buffer(mass, streamdetails, reason="test")
580    await asyncio.sleep(0.1)
581
582    mass.player_queues.prepare_next_audio_buffer.assert_called_once_with("queue_a")
583    await asyncio.gather(*scheduled_tasks)
584    await buffer.clear()
585
586
587@pytest.mark.asyncio
588async def test_get_buffer_fill_completion_is_ignored_for_non_realtime_sources() -> None:
589    """A source that delivers faster than playback frees no slot worth chaining on."""
590    mass, _start_analysis, scheduled_tasks = _make_mass_for_get_buffer()
591    streamdetails = _make_stream_details(
592        MediaType.TRACK, duration=180, allow_seek=True, queue_id="queue_a"
593    )
594
595    buffer = await AudioBuffer.get_buffer(mass, streamdetails, reason="test")
596    await asyncio.sleep(0.1)
597
598    mass.player_queues.prepare_next_audio_buffer.assert_not_called()
599    await asyncio.gather(*scheduled_tasks)
600    await buffer.clear()
601
602
603@pytest.mark.asyncio
604async def test_get_buffer_still_starts_analysis_for_track() -> None:
605    """get_buffer still schedules audio analysis for tracks."""
606    mass, start_analysis, scheduled_tasks = _make_mass_for_get_buffer()
607    streamdetails = _make_stream_details(MediaType.TRACK, duration=180, allow_seek=True)
608
609    buffer = await AudioBuffer.get_buffer(mass, streamdetails, reason="test")
610
611    assert len(scheduled_tasks) == 1
612    await asyncio.gather(*scheduled_tasks)
613    start_analysis.assert_awaited_once()
614    await buffer.clear()
615
616
617@pytest.mark.asyncio
618async def test_get_buffer_sound_effect_uses_default_ready_threshold_without_crossfade() -> None:
619    """Sound effects should not use the larger crossfade buffering threshold."""
620    mass, start_analysis, scheduled_tasks = _make_mass_for_get_buffer(
621        queue=SimpleNamespace(crossfade_enabled=True)
622    )
623    streamdetails = _make_stream_details(
624        MediaType.SOUND_EFFECT,
625        duration=30,
626        allow_seek=True,
627        queue_id="queue-1",
628    )
629
630    buffer = await AudioBuffer.get_buffer(mass, streamdetails, reason="test")
631
632    assert buffer._ready_at_chunk == 2
633    assert scheduled_tasks == []
634    start_analysis.assert_not_called()
635    await buffer.clear()
636
637
638@pytest.mark.parametrize(
639    ("max_concurrent_streams", "has_free_slot", "expect_released"),
640    [(1, False, True), (1, True, False), (None, True, False)],
641    ids=["slot_limited_saturated", "slot_limited_with_free_slot", "unlimited"],
642)
643@pytest.mark.asyncio
644async def test_get_buffer_releases_a_slot_limited_producer_before_replacing_it(
645    max_concurrent_streams: int | None, has_free_slot: bool, expect_released: bool
646) -> None:
647    """The superseded producer only gives up its slot when the provider has none to spare."""
648    mass, _start_analysis, _scheduled_tasks = _make_mass_for_get_buffer()
649    provider = MagicMock(spec=MusicProvider)
650    provider.max_concurrent_streams = max_concurrent_streams
651    provider.has_available_stream_slot = has_free_slot
652    mass.get_provider.return_value = provider
653    streamdetails = _make_stream_details(MediaType.TRACK, duration=600, allow_seek=True)
654    blocked = asyncio.Event()
655
656    async def _never_ending_source() -> AsyncGenerator[bytes]:
657        yield _make_chunk(0)
658        await blocked.wait()
659
660    stale_buffer = AudioBuffer(TEST_PCM_FORMAT)
661    stale_buffer.fill(_never_ending_source())
662    await asyncio.sleep(0)
663    await asyncio.sleep(0)
664    streamdetails.buffer = stale_buffer
665    # the producer is still charging a source slot and the consumer is active right now,
666    # so the 30s inactivity heuristic must not be what decides this
667    assert stale_buffer.is_buffering
668    assert time.time() - stale_buffer._last_access_time < 30
669
670    # a forward seek far past the buffered window can not be served by this buffer
671    assert not stale_buffer.is_valid((SEEK_WAIT_THRESHOLD + 60) * 1000)
672    replacement = await AudioBuffer.get_buffer(
673        mass,
674        streamdetails,
675        seek_position_ms=(SEEK_WAIT_THRESHOLD + 60) * 1000,
676        reason="test",
677    )
678
679    assert replacement is not stale_buffer
680    assert stale_buffer.cancelled is expect_released
681    assert stale_buffer.is_buffering is not expect_released
682    blocked.set()
683    await stale_buffer.clear()
684    await replacement.clear()
685
686
687@pytest.mark.asyncio
688async def test_get_buffer_replaces_a_live_buffer_on_a_short_forward_seek() -> None:
689    """A live source is restarted at the position rather than waited out."""
690    mass, _start_analysis, _scheduled_tasks = _make_mass_for_get_buffer()
691    provider = MagicMock(spec=MusicProvider)
692    provider.max_concurrent_streams = None
693    provider.has_available_stream_slot = True
694    mass.get_provider.return_value = provider
695    streamdetails = _make_stream_details(MediaType.TRACK, duration=600, allow_seek=True)
696    streamdetails.is_realtime = True
697    blocked = asyncio.Event()
698
699    async def _never_ending_source() -> AsyncGenerator[bytes]:
700        yield _make_chunk(0)
701        await blocked.wait()
702
703    live_buffer = AudioBuffer(TEST_PCM_FORMAT, is_realtime=True)
704    live_buffer.fill(_never_ending_source())
705    await asyncio.sleep(0)
706    await asyncio.sleep(0)
707    streamdetails.buffer = live_buffer
708
709    # well inside the span a recorded source would have been waited out for
710    replacement = await AudioBuffer.get_buffer(
711        mass, streamdetails, seek_position_ms=5000, reason="test"
712    )
713
714    assert replacement is not live_buffer
715    # the replacement carries the flag on, and starts at the position asked for
716    assert replacement.is_realtime
717    assert replacement._discarded_chunks == 5
718    blocked.set()
719    await live_buffer.clear()
720    await replacement.clear()
721
722
723@pytest.mark.asyncio
724async def test_fill_closes_source_on_cancel() -> None:
725    """The source generator is finalized immediately when the fill task is cancelled."""
726    source_closed = asyncio.Event()
727
728    async def _endless_source() -> AsyncGenerator[bytes]:
729        try:
730            while True:
731                yield ONE_SECOND_CHUNK
732        finally:
733            source_closed.set()
734
735    buf = AudioBuffer(TEST_PCM_FORMAT, buffer_size=BufferSize.MINIMAL)
736    buf.fill(_endless_source(), source_name="test")
737    # let the fill task run until it blocks on the full buffer
738    await asyncio.sleep(0.1)
739
740    # clear() cancels the fill task, which must close the source generator
741    await buf.clear()
742    await asyncio.wait_for(source_closed.wait(), timeout=1)
743
744
745# -- Seek and is_valid --
746
747
748@pytest.mark.asyncio
749async def test_is_valid_basic() -> None:
750    """is_valid returns True for buffered positions."""
751    buf = AudioBuffer(TEST_PCM_FORMAT, buffer_size=BufferSize.MINIMAL)
752    for _ in range(10):
753        await buf._put(ONE_SECOND_CHUNK)
754
755    assert buf.is_valid(seek_position_ms=0)
756    assert buf.is_valid(seek_position_ms=5000)
757    assert buf.is_valid(seek_position_ms=9000)
758
759
760@pytest.mark.asyncio
761async def test_is_valid_cancelled() -> None:
762    """is_valid returns False for cancelled buffer."""
763    buf = AudioBuffer(TEST_PCM_FORMAT)
764    await buf._put(ONE_SECOND_CHUNK)
765    await buf.clear()
766    assert not buf.is_valid()
767
768
769@pytest.mark.asyncio
770async def test_is_valid_seek_ahead_within_threshold() -> None:
771    """is_valid returns True when seek is within SEEK_WAIT_THRESHOLD of buffered data."""
772    buf = AudioBuffer(TEST_PCM_FORMAT, buffer_size=BufferSize.MINIMAL)
773    for _ in range(10):
774        await buf._put(ONE_SECOND_CHUNK)
775
776    # 10 chunks buffered, seek to 10+SEEK_WAIT_THRESHOLD seconds should be valid
777    seek_ms = (10 + SEEK_WAIT_THRESHOLD) * 1000
778    assert buf.is_valid(seek_position_ms=seek_ms)
779
780
781@pytest.mark.asyncio
782async def test_is_valid_seek_ahead_beyond_threshold() -> None:
783    """is_valid returns False when seek is beyond SEEK_WAIT_THRESHOLD."""
784    buf = AudioBuffer(TEST_PCM_FORMAT, buffer_size=BufferSize.MINIMAL)
785    for _ in range(10):
786        await buf._put(ONE_SECOND_CHUNK)
787
788    seek_ms = (10 + SEEK_WAIT_THRESHOLD + 1) * 1000
789    assert not buf.is_valid(seek_position_ms=seek_ms)
790
791
792@pytest.mark.asyncio
793async def test_is_valid_with_eof() -> None:
794    """is_valid returns True for any position when EOF is received."""
795    buf = AudioBuffer(TEST_PCM_FORMAT, buffer_size=BufferSize.MINIMAL)
796    for _ in range(5):
797        await buf._put(ONE_SECOND_CHUNK)
798    await buf._set_eof()
799
800    # even beyond buffered data, is_valid returns True with EOF
801    assert buf.is_valid(seek_position_ms=100_000)
802
803
804@pytest.mark.asyncio
805async def test_seek_in_raw_stream() -> None:
806    """get_raw_stream with seek_position_ms skips to correct chunk."""
807    buf = AudioBuffer(TEST_PCM_FORMAT, buffer_size=BufferSize.MINIMAL)
808    buf.fill(_make_source(10), source_name="test")
809    await asyncio.sleep(0.1)
810
811    chunks = []
812    async for chunk in buf.get_raw_stream(seek_position_ms=5000):
813        chunks.append(chunk)
814
815    assert len(chunks) == 5
816    # first chunk should be chunk #5
817    assert chunks[0] == _make_chunk(5)
818
819
820async def test_exact_raw_seek_preserves_millisecond_position() -> None:
821    """Crossfade continuation does not round its media-time resume backward."""
822    audio_buffer = AudioBuffer(TEST_PCM_FORMAT)
823    await audio_buffer._put(ONE_SECOND_CHUNK)
824    await audio_buffer._set_eof()
825
826    regular_stream = audio_buffer.get_raw_stream(seek_position_ms=250)
827    exact_stream = audio_buffer.get_raw_stream(seek_position_ms=250, exact_seek=True)
828    regular_chunk = await anext(regular_stream)
829    exact_chunk = await anext(exact_stream)
830    await regular_stream.aclose()
831    await exact_stream.aclose()
832
833    assert len(regular_chunk) == int(len(ONE_SECOND_CHUNK) * 0.8)
834    assert len(exact_chunk) == int(len(ONE_SECOND_CHUNK) * 0.75)
835
836
837# -- Analysis reader (read_chunk_for_analysis) --
838
839
840@pytest.mark.asyncio
841async def test_read_chunk_for_analysis_returns_buffered_chunk() -> None:
842    """A passive reader gets a retained chunk without discarding it."""
843    buf = AudioBuffer(TEST_PCM_FORMAT, buffer_size=BufferSize.MINIMAL)
844    await buf._put(_make_chunk(0))
845    await buf._put(_make_chunk(1))
846
847    assert await buf.read_chunk_for_analysis(0) == _make_chunk(0)
848    assert await buf.read_chunk_for_analysis(1) == _make_chunk(1)
849    # Reading must not have discarded anything — both chunks are still buffered.
850    assert buf.seconds_available == 2
851    assert buf.first_buffered_chunk == 0
852
853
854@pytest.mark.asyncio
855async def test_read_chunk_for_analysis_waits_then_returns() -> None:
856    """A reader ahead of the filled position waits until the chunk is produced."""
857    buf = AudioBuffer(TEST_PCM_FORMAT, buffer_size=BufferSize.MINIMAL)
858    reader = asyncio.ensure_future(buf.read_chunk_for_analysis(0))
859    await asyncio.sleep(0.05)
860    assert not reader.done()  # nothing buffered yet
861
862    await buf._put(_make_chunk(0))
863    assert await asyncio.wait_for(reader, timeout=1.0) == _make_chunk(0)
864
865
866@pytest.mark.asyncio
867async def test_read_chunk_for_analysis_raises_eof_past_end() -> None:
868    """Reading past the last chunk of an ended stream raises AudioBufferEOF."""
869    buf = AudioBuffer(TEST_PCM_FORMAT, buffer_size=BufferSize.MINIMAL)
870    await buf._put(_make_chunk(0))
871    await buf._set_eof()
872
873    assert await buf.read_chunk_for_analysis(0) == _make_chunk(0)
874    with pytest.raises(AudioBufferEOF):
875        await buf.read_chunk_for_analysis(1)
876
877
878@pytest.mark.asyncio
879async def test_read_chunk_for_analysis_raises_discarded_when_evicted() -> None:
880    """Requesting a chunk that has been evicted from the window raises AudioBufferDiscarded."""
881    buf = AudioBuffer(TEST_PCM_FORMAT, buffer_size=BufferSize.MINIMAL)
882    await buf._put(_make_chunk(0))
883    # Simulate the playback consumer sliding the window past chunk 0.
884    buf._chunks.popleft()
885    buf._discarded_chunks += 1
886    assert buf.first_buffered_chunk == 1
887
888    with pytest.raises(AudioBufferDiscarded):
889        await buf.read_chunk_for_analysis(0)
890
891
892@pytest.mark.asyncio
893async def test_read_chunk_for_analysis_raises_discarded_on_clear() -> None:
894    """A reader blocked on a torn-down buffer is released with AudioBufferDiscarded."""
895    buf = AudioBuffer(TEST_PCM_FORMAT, buffer_size=BufferSize.MINIMAL)
896    reader = asyncio.ensure_future(buf.read_chunk_for_analysis(0))
897    await asyncio.sleep(0.05)
898    await buf.clear()
899    with pytest.raises(AudioBufferDiscarded):
900        await asyncio.wait_for(reader, timeout=1.0)
901
902
903# -- Buffer size limits --
904
905
906@pytest.mark.asyncio
907async def test_rolling_buffer_fifo() -> None:
908    """ROLLING mode works as a FIFO — get pops the oldest chunk."""
909    buf = AudioBuffer(TEST_PCM_FORMAT, mode=BufferMode.ROLLING)
910
911    for i in range(5):
912        await buf._put(_make_chunk(i))
913
914    assert buf.size_seconds == 5
915
916    # get pops the oldest chunk and frees space
917    result = await buf._get(chunk_number=0)
918    assert result == _make_chunk(0)
919    assert buf.size_seconds == 4
920    assert buf._discarded_chunks == 1
921
922    # next get returns the next chunk
923    result = await buf._get(chunk_number=1)
924    assert result == _make_chunk(1)
925    assert buf.size_seconds == 3
926    assert buf._discarded_chunks == 2
927
928
929@pytest.mark.asyncio
930async def test_rolling_buffer_drained_surfaces_producer_error() -> None:
931    """A drained rolling buffer raises the producer error instead of a clean EOF."""
932
933    async def _failing_source() -> AsyncGenerator[bytes]:
934        yield ONE_SECOND_CHUNK
935        msg = "test error"
936        raise RuntimeError(msg)
937
938    buf = AudioBuffer(TEST_PCM_FORMAT, mode=BufferMode.ROLLING)
939    buf.fill(_failing_source(), source_name="test")
940    while not buf.has_error:
941        await asyncio.sleep(0.01)
942
943    # the buffered chunk is still delivered before the error surfaces
944    assert await buf._get() == ONE_SECOND_CHUNK
945    with pytest.raises(RuntimeError, match="test error"):
946        await buf._get()
947
948
949@pytest.mark.asyncio
950async def test_seekable_buffer_backpressure() -> None:
951    """SEEKABLE mode waits on put when full, consumer frees space on get."""
952    buf = AudioBuffer(TEST_PCM_FORMAT, buffer_size=BufferSize.MINIMAL)
953    max_size = buf.max_size_seconds
954
955    # use fill() so there's an active producer task (eviction only happens
956    # when the producer is running and needs space)
957    buf.fill(_make_source(max_size + 5), source_name="test")
958    async with asyncio.timeout(5):
959        async with buf._data_available:
960            await buf._data_available.wait_for(lambda: buf.size_seconds == max_size)
961
962    assert buf.size_seconds == max_size
963
964    # reading from a full buffer frees space for the producer
965    chunk = await buf._get(chunk_number=0)
966    assert chunk == _make_chunk(0)
967    assert buf._discarded_chunks == 1
968
969
970@pytest.mark.asyncio
971async def test_seekable_no_eviction_after_eof() -> None:
972    """After EOF, reads from a full buffer do not evict chunks."""
973    buf = AudioBuffer(TEST_PCM_FORMAT, buffer_size=BufferSize.MINIMAL)
974    max_size = buf.max_size_seconds
975
976    buf.fill(_make_source(max_size), source_name="test")
977    await asyncio.sleep(0.1)
978
979    assert buf._eof_received
980    assert buf.size_seconds == max_size
981
982    # read should NOT evict since producer is done
983    chunk = await buf._get(chunk_number=0)
984    assert chunk == _make_chunk(0)
985    assert buf._discarded_chunks == 0
986    assert buf.size_seconds == max_size
987
988
989# -- get_stream passthrough --
990
991
992@pytest.mark.asyncio
993async def test_get_stream_no_filters() -> None:
994    """get_stream without filters passes through raw data."""
995    buf = AudioBuffer(TEST_PCM_FORMAT, buffer_size=BufferSize.MINIMAL)
996    buf.fill(_make_source(3), source_name="test")
997    await asyncio.sleep(0.1)
998
999    chunks = []
1000    async for chunk in buf.get_stream(output_format=TEST_PCM_FORMAT):
1001        chunks.append(chunk)
1002
1003    assert len(chunks) == 3
1004    assert chunks[0] == _make_chunk(0)
1005
1006
1007# -- Rolling mode --
1008
1009
1010@pytest.mark.asyncio
1011async def test_rolling_mode_max_size() -> None:
1012    """ROLLING mode uses RADIO_BUFFER_SIZE."""
1013    buf = AudioBuffer(TEST_PCM_FORMAT, mode=BufferMode.ROLLING)
1014    assert buf.max_size_seconds == RADIO_BUFFER_SIZE
1015
1016
1017# -- Ready threshold with seek offset --
1018
1019
1020@pytest.mark.asyncio
1021async def test_ready_accounts_for_seek_offset() -> None:
1022    """Ready fires only after enough data past the seek point is buffered."""
1023    buf = AudioBuffer(TEST_PCM_FORMAT, ready_threshold=3)
1024    # simulate get_buffer setting the offset for a seek to 100s
1025    buf._discarded_chunks = 100
1026    buf._ready_at_chunk = 100 + 3  # seek_chunk + threshold
1027
1028    await buf._put(ONE_SECOND_CHUNK)  # chunk 100
1029    assert not buf.ready.is_set()
1030    await buf._put(ONE_SECOND_CHUNK)  # chunk 101
1031    assert not buf.ready.is_set()
1032    await buf._put(ONE_SECOND_CHUNK)  # chunk 102
1033    assert buf.ready.is_set()
1034
1035
1036@pytest.mark.asyncio
1037async def test_chunk_numbering_with_seek_offset() -> None:
1038    """Chunks are numbered correctly when buffer starts at a seek offset."""
1039    buf = AudioBuffer(TEST_PCM_FORMAT, buffer_size=BufferSize.MINIMAL)
1040    # simulate a buffer created for a seek to 300s
1041    buf._discarded_chunks = 300
1042
1043    for i in range(5):
1044        await buf._put(_make_chunk(i))
1045
1046    # chunk 300 should be the first chunk (value 0)
1047    result = await buf._get(chunk_number=300)
1048    assert result == _make_chunk(0)
1049    # chunk 304 should be the fifth chunk (value 4)
1050    result = await buf._get(chunk_number=304)
1051    assert result == _make_chunk(4)
1052
1053
1054@pytest.mark.asyncio
1055async def test_is_valid_with_seek_offset() -> None:
1056    """is_valid works correctly with a seek offset."""
1057    buf = AudioBuffer(TEST_PCM_FORMAT, buffer_size=BufferSize.MINIMAL)
1058    buf._discarded_chunks = 300
1059
1060    for _ in range(10):
1061        await buf._put(ONE_SECOND_CHUNK)
1062
1063    # positions before the offset are invalid (discarded)
1064    assert not buf.is_valid(seek_position_ms=299_000)
1065    # positions within the buffer are valid
1066    assert buf.is_valid(seek_position_ms=300_000)
1067    assert buf.is_valid(seek_position_ms=305_000)
1068
1069
1070@pytest.mark.asyncio
1071async def test_raw_stream_with_seek_offset() -> None:
1072    """get_raw_stream works correctly when buffer has a seek offset."""
1073    buf = AudioBuffer(TEST_PCM_FORMAT, buffer_size=BufferSize.MINIMAL)
1074    buf._discarded_chunks = 300
1075
1076    for i in range(5):
1077        await buf._put(_make_chunk(i))
1078    await buf._set_eof()
1079
1080    chunks = []
1081    async for chunk in buf.get_raw_stream(seek_position_ms=300_000):
1082        chunks.append(chunk)
1083
1084    assert len(chunks) == 5
1085    assert chunks[0] == _make_chunk(0)
1086    assert chunks[4] == _make_chunk(4)
1087
1088
1089# -- Callback error isolation --
1090
1091
1092@pytest.mark.asyncio
1093async def test_clear_fires_cancel_callbacks() -> None:
1094    """clear() fires registered cancel callbacks before removing them."""
1095    cancel_called = False
1096
1097    def _cancel_callback() -> None:
1098        nonlocal cancel_called
1099        cancel_called = True
1100
1101    buf = AudioBuffer(TEST_PCM_FORMAT, buffer_size=BufferSize.MINIMAL)
1102    buf.register_cancel_callback(_cancel_callback)
1103    await buf._put(ONE_SECOND_CHUNK)
1104
1105    await buf.clear()
1106    assert cancel_called is True
1107    assert len(buf._cancel_callbacks) == 0
1108
1109
1110# -- Inactivity monitor --
1111
1112
1113@pytest.mark.asyncio
1114async def test_inactivity_monitor_releases_drained_buffer() -> None:
1115    """
1116    A buffer that has drained to empty is still released by the inactivity monitor.
1117
1118    Regression test: the monitor previously only cleared when chunks remained, so an
1119    abandoned rolling buffer that drained to zero chunks looped forever and leaked it
1120    (and its producer/ffmpeg) until the process exited.
1121    """
1122    buf = AudioBuffer(TEST_PCM_FORMAT, mode=BufferMode.ROLLING)
1123    # no chunks buffered and last access long ago -> the buffer is inactive
1124    assert buf.size_seconds == 0
1125    buf._last_access_time = time.time() - 10_000
1126
1127    await buf._monitor_inactivity(inactivity_timeout=0.01, check_interval=0.01)
1128
1129    assert buf.cancelled is True
1130
1131
1132@pytest.mark.asyncio
1133async def test_inactivity_monitor_keeps_active_buffer() -> None:
1134    """A buffer that is still being accessed is not cleared by the inactivity monitor."""
1135    buf = AudioBuffer(TEST_PCM_FORMAT, mode=BufferMode.ROLLING)
1136    buf._last_access_time = time.time()
1137
1138    monitor = asyncio.create_task(
1139        buf._monitor_inactivity(inactivity_timeout=5, check_interval=0.01)
1140    )
1141    await asyncio.sleep(0.05)
1142
1143    assert not monitor.done()
1144    assert buf.cancelled is False
1145
1146    monitor.cancel()
1147    with suppress(asyncio.CancelledError):
1148        await monitor
1149
1150
1151# -- Pre-buffering of the next queue item --
1152
1153
1154@pytest.fixture
1155async def mass_minimal(mass_minimal: MusicAssistant) -> MusicAssistant:
1156    """Extend the base fixture with the player_queues/streams stand-ins get_queue_item_stream needs."""
1157    mass_minimal.player_queues = SimpleNamespace(  # type: ignore[assignment]
1158        get_active_queue=lambda _queue_id: None,
1159        prepare_next_audio_buffer=lambda _queue_id: None,
1160        queue_data_or_none=lambda _queue_id: None,
1161    )
1162    mass_minimal.streams = MagicMock()
1163    return mass_minimal
1164
1165
1166class _FakeAudioBuffer:
1167    """AudioBuffer test double that streams a fixed run of 1-second chunks."""
1168
1169    has_error = False
1170    pcm_format = TEST_PCM_FORMAT
1171
1172    @classmethod
1173    async def get_buffer(cls, **_kwargs: Any) -> _FakeAudioBuffer:
1174        return cls()
1175
1176    async def get_stream(self, **_kwargs: Any) -> AsyncGenerator[bytes]:
1177        async for chunk in _make_source(90):
1178            yield chunk
1179
1180
1181async def _stream_until_prebuffer_window(
1182    mass: MusicAssistant,
1183    *,
1184    next_item_media_type: MediaType,
1185    queue_id: str,
1186    is_realtime: bool = False,
1187) -> None:
1188    """
1189    Drive get_queue_item_stream for a 90s current TRACK item past the pre-buffer trigger point.
1190
1191    Sets up a queue whose next item has ``next_item_media_type`` and streams the current
1192    item to completion, so the pre-buffer trigger condition (evaluated once more than
1193    duration - 60 seconds of PCM has been yielded) gets a chance to fire.
1194
1195    :param is_realtime: Whether the current item's source hands over its audio
1196        just-in-time, which moves the trigger to the source itself.
1197    """
1198    streamdetails = _make_stream_details(MediaType.TRACK, duration=90, allow_seek=True)
1199    streamdetails.is_realtime = is_realtime
1200    streamdetails.loudness = -10.0  # skip the audio-analysis hydration call
1201    current_item = QueueItem(
1202        queue_id=queue_id,
1203        queue_item_id="current",
1204        name="Current",
1205        duration=90,
1206        streamdetails=streamdetails,
1207    )
1208    next_item = SimpleNamespace(queue_item_id="next", media_type=next_item_media_type)
1209    queue = SimpleNamespace(next_item=next_item)
1210    mass.player_queues.get_active_queue = lambda _player_id: queue  # type: ignore[method-assign, assignment, return-value]
1211
1212    controller = StreamsAudio(mass)
1213    with patch.object(audio_mod, "AudioBuffer", _FakeAudioBuffer):
1214        async for _chunk in controller.get_queue_item_stream(current_item, TEST_PCM_FORMAT):
1215            pass
1216
1217
1218@pytest.mark.asyncio
1219async def test_sound_effect_next_item_triggers_prebuffer(mass_minimal: MusicAssistant) -> None:
1220    """A SOUND_EFFECT next item is pre-buffered like a track."""
1221    calls: list[str] = []
1222    mass_minimal.player_queues.prepare_next_audio_buffer = (  # type: ignore[method-assign]
1223        lambda queue_id: calls.append(queue_id)
1224    )
1225
1226    await _stream_until_prebuffer_window(
1227        mass_minimal, next_item_media_type=MediaType.SOUND_EFFECT, queue_id="player_a"
1228    )
1229
1230    assert calls == ["player_a"]
1231
1232
1233@pytest.mark.asyncio
1234async def test_audio_source_next_item_is_not_prebuffered(mass_minimal: MusicAssistant) -> None:
1235    """A live AUDIO_SOURCE next item is still excluded from pre-buffering."""
1236    calls: list[str] = []
1237    mass_minimal.player_queues.prepare_next_audio_buffer = (  # type: ignore[method-assign]
1238        lambda queue_id: calls.append(queue_id)
1239    )
1240
1241    await _stream_until_prebuffer_window(
1242        mass_minimal, next_item_media_type=MediaType.AUDIO_SOURCE, queue_id="player_a"
1243    )
1244
1245    assert calls == []
1246
1247
1248@pytest.mark.asyncio
1249async def test_realtime_source_also_gets_the_fallback_prebuffer_trigger(
1250    mass_minimal: MusicAssistant,
1251) -> None:
1252    """
1253    A realtime track keeps the read-side trigger as its fallback.
1254
1255    Its slot usually frees (and prepares the next item) when an earlier fill
1256    completes, but once the lead spans a whole item that moment has no next
1257    item yet - this trigger is what starts it then.
1258    """
1259    calls: list[str] = []
1260    mass_minimal.player_queues.prepare_next_audio_buffer = (  # type: ignore[method-assign]
1261        lambda queue_id: calls.append(queue_id)
1262    )
1263
1264    await _stream_until_prebuffer_window(
1265        mass_minimal,
1266        next_item_media_type=MediaType.TRACK,
1267        queue_id="player_a",
1268        is_realtime=True,
1269    )
1270
1271    assert calls == ["player_a"]
1272
1273
1274@pytest.mark.asyncio
1275async def test_real_buffer_producer_error_reaches_queue_item_stream(
1276    mass_minimal: MusicAssistant,
1277) -> None:
1278    """A real AudioBuffer producer error is surfaced instead of a truncated stream."""
1279
1280    async def _failing_source() -> AsyncGenerator[bytes]:
1281        yield ONE_SECOND_CHUNK
1282        raise RuntimeError("source failed")
1283
1284    streamdetails = _make_stream_details(MediaType.SOUND_EFFECT, duration=90, allow_seek=True)
1285    streamdetails.loudness = -10.0
1286    queue_item = QueueItem(
1287        queue_id="player_a",
1288        queue_item_id="current",
1289        name="Current",
1290        duration=90,
1291        streamdetails=streamdetails,
1292    )
1293    cast("Any", mass_minimal.player_queues).get = MagicMock(return_value=None)
1294    cast("Any", mass_minimal.streams.audio).get_media_stream = MagicMock(
1295        return_value=_failing_source()
1296    )
1297    controller = StreamsAudio(mass_minimal)
1298
1299    chunks: list[bytes] = []
1300    async for chunk in controller.get_queue_item_stream(
1301        queue_item, TEST_PCM_FORMAT, raise_on_error=False
1302    ):
1303        chunks.append(chunk)
1304
1305    # the stream waits for the buffer to become playable, so a producer failure is
1306    # reported before any audio is served rather than truncating it mid-stream
1307    assert chunks == []
1308    assert streamdetails.stream_error is True
1309    assert queue_item.available
1310
1311
1312@pytest.mark.asyncio
1313async def test_stale_stream_error_reset_on_stream_start(mass_minimal: MusicAssistant) -> None:
1314    """A stream_error left on reused streamdetails is cleared when a new stream starts."""
1315    streamdetails = _make_stream_details(MediaType.TRACK, duration=90, allow_seek=True)
1316    streamdetails.loudness = -10.0  # skip the audio-analysis hydration call
1317    streamdetails.stream_error = True  # left over from a previously failed attempt
1318    queue_item = QueueItem(
1319        queue_id="player_a",
1320        queue_item_id="current",
1321        name="Current",
1322        duration=90,
1323        streamdetails=streamdetails,
1324    )
1325    controller = StreamsAudio(mass_minimal)
1326
1327    with patch.object(audio_mod, "AudioBuffer", _FakeAudioBuffer):
1328        async for _chunk in controller.get_queue_item_stream(queue_item, TEST_PCM_FORMAT):
1329            pass
1330
1331    assert streamdetails.stream_error is False
1332
1333
1334@pytest.mark.asyncio
1335async def test_audio_source_stream_error_reset_on_retry(mass_minimal: MusicAssistant) -> None:
1336    """A cached AudioSource stream clears a prior error before retrying."""
1337    streamdetails = _make_stream_details(MediaType.AUDIO_SOURCE, duration=None, allow_seek=False)
1338    streamdetails.stream_error = True
1339    queue_item = QueueItem(
1340        queue_id="player_a",
1341        queue_item_id="source",
1342        name="Source",
1343        duration=0,
1344        streamdetails=streamdetails,
1345    )
1346    controller = StreamsAudio(mass_minimal)
1347
1348    async def _source(
1349        _streamdetails: StreamDetails, _pcm_format: AudioFormat
1350    ) -> AsyncGenerator[bytes]:
1351        yield ONE_SECOND_CHUNK
1352
1353    with patch.object(controller, "_iter_audio_source_pcm", _source):
1354        chunks = [
1355            chunk async for chunk in controller.get_queue_item_stream(queue_item, TEST_PCM_FORMAT)
1356        ]
1357
1358    assert chunks == [ONE_SECOND_CHUNK]
1359    assert streamdetails.stream_error is False
1360
1361
1362# -- Provider-filled buffers --
1363