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