music-assistant-server

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