music-assistant-server

1"""Tests for the ffmpeg input arguments StreamsAudio.get_media_stream builds."""
2
3from __future__ import annotations
4
5import asyncio
6from collections.abc import AsyncGenerator
7from contextlib import suppress
8from types import SimpleNamespace
9from typing import Any, cast
10from unittest.mock import MagicMock
11
12import pytest
13from music_assistant_models.enums import ContentType, MediaType, ProviderType, StreamType
14from music_assistant_models.errors import AudioError, ProviderUnavailableError
15from music_assistant_models.media_items import AudioFormat
16from music_assistant_models.streamdetails import MultiPartPath, StreamDetails
17
18import music_assistant.controllers.streams.audio as audio_mod
19from music_assistant.controllers.streams.audio import StreamsAudio
20from music_assistant.controllers.streams.audio_buffer import AudioBuffer
21from music_assistant.models.music_provider import MusicProvider, ProviderStreamLimitError
22
23# input args a provider may attach to its StreamDetails (podcastfeed does exactly this).
24# Kept as a tuple so the tests below can never assert against a mutated expectation.
25_PROVIDER_INPUT_ARGS = ("-user_agent", "Test/1.0")
26
27
28class _FakeFFMpeg:
29    """FFMpeg test double that records the arguments it was constructed with."""
30
31    last_instance: _FakeFFMpeg | None = None
32    # exit code the process is reaped with; None while it is still running
33    exit_code: int = 0
34
35    def __init__(
36        self,
37        *,
38        audio_input: object,
39        input_format: AudioFormat,
40        extra_input_args: list[str] | None = None,
41        **_kwargs: Any,
42    ) -> None:
43        self.audio_input = audio_input
44        self.extra_input_args = extra_input_args
45        # Mirror the real FFMpeg, which mutates this object's codec_type after probe.
46        # Tests inspect the original `input_format` AudioFormat passed in to confirm
47        # which one the controller picked.
48        self.input_format = input_format
49        self._probed_codec_type = ContentType.FLAC  # arbitrary, distinct from PCM/OGG
50        self.parsed_duration: int | None = None
51        # mirrors the real FFMpeg: unset while the process runs, filled in on reap
52        self.returncode: int | None = None
53        self.log_history: list[str] = []
54        self.proc = MagicMock(pid=1234)
55        self.stdin_feeder_exception: Exception | None = None
56        # records which teardown the stream picked: a clean end drains, anything
57        # else kills outright
58        self.torn_down_via: str | None = None
59        type(self).last_instance = self
60
61    async def start(self) -> None:
62        # Simulate ffmpeg's post-probe codec detection: real FFMpeg mutates
63        # self.input_format.codec_type once it reads the input header.
64        self.input_format.codec_type = self._probed_codec_type
65
66    async def iter_chunked(self, _chunk_size: int) -> AsyncGenerator[bytes]:
67        yield b"\x00\x01" * 256
68
69    async def wait_with_timeout(self, _timeout: float) -> None:
70        self.returncode = self.exit_code
71
72    async def close(self) -> None:
73        self.torn_down_via = "close"
74        self.returncode = self.exit_code
75
76    async def kill(self) -> None:
77        self.torn_down_via = "kill"
78        self.returncode = -9
79
80
81@pytest.fixture
82def patch_ffmpeg(monkeypatch: pytest.MonkeyPatch) -> type[_FakeFFMpeg]:
83    """Swap the real FFMpeg in the streams.audio module for the fake."""
84    _FakeFFMpeg.last_instance = None
85    monkeypatch.setattr(audio_mod, "FFMpeg", _FakeFFMpeg)
86    return _FakeFFMpeg
87
88
89@pytest.fixture
90def patch_two_minute_ffmpeg(monkeypatch: pytest.MonkeyPatch) -> type[_TwoMinuteFFMpeg]:
91    """Swap the real FFMpeg for the fake that emits a fixed amount of audio."""
92    _TwoMinuteFFMpeg.last_instance = None
93    monkeypatch.setattr(audio_mod, "FFMpeg", _TwoMinuteFFMpeg)
94    return _TwoMinuteFFMpeg
95
96
97def _make_audio_controller() -> StreamsAudio:
98    """Build a StreamsAudio with just enough mass scaffolding to run get_media_stream."""
99    audio = StreamsAudio(MagicMock())
100    audio.mass.loop = MagicMock()
101    audio.mass.loop.time = MagicMock(return_value=0.0)
102    return audio
103
104
105def _make_pcm_format() -> AudioFormat:
106    return AudioFormat(
107        content_type=ContentType.PCM_S16LE,
108        codec_type=ContentType.PCM_S16LE,
109        sample_rate=44100,
110        bit_depth=16,
111        channels=2,
112    )
113
114
115_PCM_SAMPLE_SIZE = _make_pcm_format().pcm_sample_size
116
117
118def _make_streamdetails(
119    *,
120    audio_format: AudioFormat,
121    decoded_audio_format: AudioFormat | None = None,
122    extra_input_args: list[str] | None = None,
123) -> StreamDetails:
124    return StreamDetails(
125        provider="test_provider",
126        item_id="main",
127        audio_format=audio_format,
128        decoded_audio_format=decoded_audio_format,
129        media_type=MediaType.AUDIO_SOURCE,
130        stream_type=StreamType.NAMED_PIPE,
131        path="/tmp/fake-fifo",  # noqa: S108
132        extra_input_args=extra_input_args or [],
133    )
134
135
136def _seekable_streamdetails() -> StreamDetails:
137    """Build seekable StreamDetails carrying provider-supplied ffmpeg input args."""
138    return StreamDetails(
139        provider="test_provider",
140        item_id="episode-1",
141        audio_format=AudioFormat(content_type=ContentType.MP3),
142        media_type=MediaType.PODCAST_EPISODE,
143        stream_type=StreamType.HTTP,
144        path="http://test.invalid/episode-1.mp3",
145        duration=3600,
146        can_seek=True,
147        allow_seek=True,
148        extra_input_args=[*_PROVIDER_INPUT_ARGS],
149    )
150
151
152async def _drain(gen: AsyncGenerator[bytes]) -> None:
153    async for _ in gen:
154        pass
155
156
157def _recording_multi_file_stream() -> tuple[Any, list[int]]:
158    """
159    Build a stand-in for the concat stream plus the list of seek positions it received.
160
161    Avoids a real ffmpeg process and temp file while still proving the seek was
162    handed off to the source rather than applied through the -ss argument.
163    """
164    received_seeks: list[int] = []
165
166    async def _empty_stream() -> AsyncGenerator[bytes]:
167        yield b""
168
169    # record on call rather than on first iteration: the FFMpeg double never
170    # consumes the generator it is handed, so its body would never run
171    def _fake_stream(
172        _streamdetails: StreamDetails, seek_position: int = 0
173    ) -> AsyncGenerator[bytes]:
174        received_seeks.append(seek_position)
175        return _empty_stream()
176
177    return _fake_stream, received_seeks
178
179
180class _StallingFFMpeg(_FakeFFMpeg):
181    """FFMpeg double whose read never produces a chunk (frozen source)."""
182
183    async def iter_chunked(self, _chunk_size: int) -> AsyncGenerator[bytes]:
184        await asyncio.Event().wait()  # blocks until the watchdog cancels the read
185        yield b""  # unreachable
186
187
188class _StallingFeederErrorFFMpeg(_StallingFFMpeg):
189    """FFMpeg double that stalls after its input feeder fails."""
190
191    error: Exception
192
193    def __init__(self, **kwargs: Any) -> None:
194        super().__init__(**kwargs)
195        self.stdin_feeder_exception = self.error
196
197
198class _SlowConsumerFFMpeg(_FakeFFMpeg):
199    """FFMpeg double that hands over chunks instantly when asked."""
200
201    async def iter_chunked(self, _chunk_size: int) -> AsyncGenerator[bytes]:
202        for _ in range(3):
203            yield b"\x00\x01" * 256
204
205
206class _TwoMinuteFFMpeg(_FakeFFMpeg):
207    """FFMpeg double that emits exactly two minutes of PCM at the format below."""
208
209    seconds_emitted = 120
210
211    async def iter_chunked(self, _chunk_size: int) -> AsyncGenerator[bytes]:
212        for _ in range(self.seconds_emitted):
213            yield b"\x00" * _PCM_SAMPLE_SIZE
214
215
216class _AlreadyExitedFFMpeg(_FakeFFMpeg):
217    """FFMpeg double whose process exited with an error code before teardown."""
218
219    exit_code = 1
220
221
222class _FailingStartFFMpeg(_FakeFFMpeg):
223    """FFMpeg double that fails while opening its source."""
224
225    async def start(self) -> None:
226        """Fail source startup."""
227        raise RuntimeError("source failed")
228
229
230class _FeederErrorFFMpeg(_FakeFFMpeg):
231    """FFMpeg double whose input feeder failed before producing PCM."""
232
233    error: Exception
234
235    def __init__(self, **kwargs: Any) -> None:
236        super().__init__(**kwargs)
237        self.stdin_feeder_exception = self.error
238
239    async def iter_chunked(self, _chunk_size: int) -> AsyncGenerator[bytes]:
240        """Yield no PCM after the feeder failure."""
241        if _chunk_size < 0:
242            yield b""
243
244
245class _SourceConsumingFFMpeg(_FakeFFMpeg):
246    """FFMpeg double that consumes its generator input before ending stdout."""
247
248    async def iter_chunked(self, _chunk_size: int) -> AsyncGenerator[bytes]:
249        """Yield source bytes and record a source failure like the real feeder."""
250        assert isinstance(self.audio_input, AsyncGenerator)
251        try:
252            async for chunk in self.audio_input:
253                yield chunk
254        except Exception as err:
255            self.stdin_feeder_exception = err
256
257
258class _LimitedProvider(MusicProvider):
259    """Music provider with one source-stream slot."""
260
261    @property
262    def max_concurrent_streams(self) -> int:
263        """Return one source-stream slot."""
264        return 1
265
266    def get_audio_stream(
267        self, _streamdetails: StreamDetails, seek_position: int = 0
268    ) -> AsyncGenerator[bytes]:
269        """Yield a custom source chunk."""
270        del seek_position
271
272        async def _source() -> AsyncGenerator[bytes]:
273            yield b"source"
274
275        return _source()
276
277
278def _limited_provider() -> _LimitedProvider:
279    """Build a one-slot music provider."""
280    mass = MagicMock()
281    manifest = MagicMock()
282    manifest.type = ProviderType.MUSIC
283    manifest.domain = "limited"
284    manifest.name = "Limited"
285    config = MagicMock()
286    config.name = "Limited"
287    config.instance_id = "limited--1"
288    config.get_value.return_value = "GLOBAL"
289    provider = _LimitedProvider(mass, manifest, config)
290    # a provider that can serve a stream is a loaded one
291    provider.available = True
292    return provider
293
294
295def _provider_http_streamdetails(provider: MusicProvider) -> StreamDetails:
296    """Build HTTP stream details owned by the given provider."""
297    return StreamDetails(
298        provider=provider.instance_id,
299        item_id="track-1",
300        audio_format=AudioFormat(content_type=ContentType.MP3),
301        media_type=MediaType.TRACK,
302        stream_type=StreamType.HTTP,
303        path="http://test.invalid/track.mp3",
304    )
305
306
307def _multi_part_streamdetails() -> StreamDetails:
308    """Build StreamDetails for a multi-file audiobook of two 30 minute parts."""
309    return StreamDetails(
310        provider="test_provider",
311        item_id="audiobook-1",
312        audio_format=AudioFormat(content_type=ContentType.MP3),
313        media_type=MediaType.AUDIOBOOK,
314        stream_type=StreamType.HTTP,
315        path=[
316            MultiPartPath(path="http://test.invalid/part-1.mp3", duration=1800),
317            MultiPartPath(path="http://test.invalid/part-2.mp3", duration=1800),
318        ],
319        duration=3600,
320        can_seek=True,
321        allow_seek=True,
322    )
323
324
325def _flac_streamdetails(extra_input_args: list[str] | None = None) -> StreamDetails:
326    return _make_streamdetails(
327        audio_format=AudioFormat(
328            content_type=ContentType.FLAC,
329            codec_type=ContentType.FLAC,
330            sample_rate=44100,
331            bit_depth=16,
332            channels=2,
333        ),
334        extra_input_args=extra_input_args,
335    )
336
337
338@pytest.mark.asyncio
339async def test_get_media_stream_raises_when_source_stalls(
340    monkeypatch: pytest.MonkeyPatch,
341) -> None:
342    """A source that stops producing audio is surfaced as an AudioError."""
343    monkeypatch.setattr(audio_mod, "FFMpeg", _StallingFFMpeg)
344    monkeypatch.setattr(audio_mod, "STREAM_START_TIMEOUT", 0.1)
345    monkeypatch.setattr(audio_mod, "STREAM_STALL_TIMEOUT", 0.1)
346
347    audio = _make_audio_controller()
348    with pytest.raises(AudioError):
349        await _drain(audio.get_media_stream(_flac_streamdetails(), _make_pcm_format()))
350
351
352@pytest.mark.asyncio
353async def test_get_media_stream_kills_ffmpeg_when_source_stalls(
354    monkeypatch: pytest.MonkeyPatch,
355) -> None:
356    """A stalled source kills ffmpeg outright instead of waiting out the pipe drains."""
357    monkeypatch.setattr(audio_mod, "FFMpeg", _StallingFFMpeg)
358    monkeypatch.setattr(audio_mod, "STREAM_START_TIMEOUT", 0.1)
359    monkeypatch.setattr(audio_mod, "STREAM_STALL_TIMEOUT", 0.1)
360    monkeypatch.setattr(_StallingFFMpeg, "last_instance", None)
361
362    audio = _make_audio_controller()
363    with pytest.raises(AudioError):
364        await _drain(audio.get_media_stream(_flac_streamdetails(), _make_pcm_format()))
365
366    assert _StallingFFMpeg.last_instance is not None
367    assert _StallingFFMpeg.last_instance.torn_down_via == "kill"
368
369
370@pytest.mark.asyncio
371async def test_get_media_stream_kills_ffmpeg_when_cancelled(
372    patch_ffmpeg: type[_FakeFFMpeg],
373) -> None:
374    """A consumer that walks away mid-stream kills ffmpeg rather than draining it."""
375    audio = _make_audio_controller()
376    stream = audio.get_media_stream(_flac_streamdetails(), _make_pcm_format())
377    await anext(stream)
378    await stream.aclose()
379
380    assert patch_ffmpeg.last_instance is not None
381    assert patch_ffmpeg.last_instance.torn_down_via == "kill"
382
383
384@pytest.mark.asyncio
385async def test_get_media_stream_closes_ffmpeg_that_already_exited(
386    monkeypatch: pytest.MonkeyPatch,
387) -> None:
388    """An ffmpeg that already exited is closed, so its stdin feeder is still cancelled."""
389    monkeypatch.setattr(audio_mod, "FFMpeg", _AlreadyExitedFFMpeg)
390    monkeypatch.setattr(_AlreadyExitedFFMpeg, "last_instance", None)
391
392    audio = _make_audio_controller()
393    with pytest.raises(AudioError):
394        await _drain(audio.get_media_stream(_flac_streamdetails(), _make_pcm_format()))
395
396    assert _AlreadyExitedFFMpeg.last_instance is not None
397    assert _AlreadyExitedFFMpeg.last_instance.torn_down_via == "close"
398
399
400@pytest.mark.asyncio
401async def test_get_media_stream_closes_ffmpeg_on_clean_end(
402    patch_ffmpeg: type[_FakeFFMpeg],
403) -> None:
404    """A stream that reaches its end still drains ffmpeg, so no trailing audio is lost."""
405    audio = _make_audio_controller()
406    await _drain(audio.get_media_stream(_flac_streamdetails(), _make_pcm_format()))
407
408    assert patch_ffmpeg.last_instance is not None
409    assert patch_ffmpeg.last_instance.torn_down_via == "close"
410
411
412@pytest.mark.asyncio
413async def test_get_media_stream_does_not_stall_on_slow_consumer(
414    monkeypatch: pytest.MonkeyPatch,
415) -> None:
416    """A consumer slower than the stall timeout must not trip the watchdog."""
417    monkeypatch.setattr(audio_mod, "FFMpeg", _SlowConsumerFFMpeg)
418    monkeypatch.setattr(audio_mod, "STREAM_START_TIMEOUT", 0.1)
419    monkeypatch.setattr(audio_mod, "STREAM_STALL_TIMEOUT", 0.1)
420
421    audio = _make_audio_controller()
422    chunks = 0
423    async for _ in audio.get_media_stream(_flac_streamdetails(), _make_pcm_format()):
424        chunks += 1
425        await asyncio.sleep(0.3)  # downstream waits far longer than the stall timeout
426    assert chunks == 3
427
428
429@pytest.mark.asyncio
430async def test_get_media_stream_prefers_decoded_audio_format(
431    patch_ffmpeg: type[_FakeFFMpeg],
432) -> None:
433    """When decoded_audio_format is set, ffmpeg receives that as input_format."""
434    source_format = AudioFormat(
435        content_type=ContentType.OGG,
436        codec_type=ContentType.VORBIS,
437        sample_rate=44100,
438        bit_depth=16,
439        channels=2,
440        bit_rate=320,
441    )
442    decoded_format = AudioFormat(
443        content_type=ContentType.PCM_S16LE,
444        codec_type=ContentType.PCM_S16LE,
445        sample_rate=44100,
446        bit_depth=16,
447        channels=2,
448    )
449    streamdetails = _make_streamdetails(
450        audio_format=source_format, decoded_audio_format=decoded_format
451    )
452
453    audio = _make_audio_controller()
454    await _drain(audio.get_media_stream(streamdetails, _make_pcm_format()))
455
456    assert patch_ffmpeg.last_instance is not None
457    assert patch_ffmpeg.last_instance.input_format is decoded_format
458
459
460@pytest.mark.asyncio
461async def test_get_media_stream_falls_back_to_audio_format(
462    patch_ffmpeg: type[_FakeFFMpeg],
463) -> None:
464    """When decoded_audio_format is not set, ffmpeg receives audio_format as input_format."""
465    source_format = AudioFormat(
466        content_type=ContentType.FLAC,
467        codec_type=ContentType.FLAC,
468        sample_rate=44100,
469        bit_depth=16,
470        channels=2,
471    )
472    streamdetails = _make_streamdetails(audio_format=source_format)
473
474    audio = _make_audio_controller()
475    await _drain(audio.get_media_stream(streamdetails, _make_pcm_format()))
476
477    assert patch_ffmpeg.last_instance is not None
478    assert patch_ffmpeg.last_instance.input_format is source_format
479
480
481@pytest.mark.asyncio
482@pytest.mark.usefixtures("patch_ffmpeg")
483async def test_get_media_stream_does_not_overwrite_source_codec_when_decoded_format_set() -> None:
484    """audio_format.codec_type stays authoritative when decoded_audio_format is set."""
485    source_format = AudioFormat(
486        content_type=ContentType.OGG,
487        codec_type=ContentType.VORBIS,
488        sample_rate=44100,
489        bit_depth=16,
490        channels=2,
491        bit_rate=320,
492    )
493    decoded_format = AudioFormat(
494        content_type=ContentType.PCM_S16LE,
495        codec_type=ContentType.PCM_S16LE,
496        sample_rate=44100,
497        bit_depth=16,
498        channels=2,
499    )
500    streamdetails = _make_streamdetails(
501        audio_format=source_format, decoded_audio_format=decoded_format
502    )
503
504    audio = _make_audio_controller()
505    await _drain(audio.get_media_stream(streamdetails, _make_pcm_format()))
506
507    assert streamdetails.audio_format.codec_type is ContentType.VORBIS
508
509
510@pytest.mark.asyncio
511@pytest.mark.usefixtures("patch_ffmpeg")
512async def test_get_media_stream_writes_back_codec_when_no_decoded_format() -> None:
513    """Without decoded_audio_format, ffmpeg's probed codec_type is written back."""
514    source_format = AudioFormat(
515        content_type=ContentType.FLAC,
516        # Start with UNKNOWN so we can see the post-probe writeback take effect.
517        codec_type=ContentType.UNKNOWN,
518        sample_rate=44100,
519        bit_depth=16,
520        channels=2,
521    )
522    streamdetails = _make_streamdetails(audio_format=source_format)
523
524    audio = _make_audio_controller()
525    await _drain(audio.get_media_stream(streamdetails, _make_pcm_format()))
526
527    # _FakeFFMpeg's start() mutates input_format.codec_type to FLAC; with no
528    # decoded format that AudioFormat is the same object as streamdetails.audio_format,
529    # so the controller's writeback path is exercised end-to-end.
530    assert streamdetails.audio_format.codec_type is ContentType.FLAC
531
532
533@pytest.mark.asyncio
534async def test_get_media_stream_stores_measured_duration_for_full_playthrough(
535    monkeypatch: pytest.MonkeyPatch,
536    patch_two_minute_ffmpeg: type[_TwoMinuteFFMpeg],
537) -> None:
538    """A multi-file item streamed from the start gets its measured duration stored."""
539    streamdetails = _multi_part_streamdetails()
540    audio = _make_audio_controller()
541    fake_stream, _ = _recording_multi_file_stream()
542    monkeypatch.setattr(audio, "get_multi_file_stream", fake_stream)
543
544    await _drain(audio.get_media_stream(streamdetails, _make_pcm_format()))
545
546    assert streamdetails.duration == patch_two_minute_ffmpeg.seconds_emitted
547
548
549@pytest.mark.asyncio
550async def test_get_media_stream_keeps_duration_when_multi_file_seek_is_delegated(
551    monkeypatch: pytest.MonkeyPatch,
552    patch_two_minute_ffmpeg: type[_TwoMinuteFFMpeg],
553) -> None:
554    """Resuming a multi-file audiobook must not shrink its duration to the remainder."""
555    streamdetails = _multi_part_streamdetails()
556    audio = _make_audio_controller()
557    fake_stream, received_seeks = _recording_multi_file_stream()
558    monkeypatch.setattr(audio, "get_multi_file_stream", fake_stream)
559
560    await _drain(audio.get_media_stream(streamdetails, _make_pcm_format(), seek_position=1800))
561
562    # the concat stream consumes the seek itself, which clears the local seek
563    # position before the duration writeback runs at the end of the stream
564    assert received_seeks == [1800]
565    assert patch_two_minute_ffmpeg.last_instance is not None
566    assert "-ss" not in (patch_two_minute_ffmpeg.last_instance.extra_input_args or [])
567    assert streamdetails.duration == 3600
568
569
570@pytest.mark.asyncio
571async def test_get_media_stream_keeps_duration_when_provider_seek_is_delegated(
572    patch_two_minute_ffmpeg: type[_TwoMinuteFFMpeg],
573) -> None:
574    """A seekable provider stream must not shrink its duration to the remainder either."""
575    streamdetails = StreamDetails(
576        provider="test_provider",
577        item_id="track-1",
578        audio_format=AudioFormat(content_type=ContentType.OGG),
579        media_type=MediaType.TRACK,
580        stream_type=StreamType.CUSTOM,
581        duration=240,
582        can_seek=True,
583        allow_seek=True,
584    )
585    audio = _make_audio_controller()
586    provider = cast("MagicMock", audio.mass).get_provider.return_value
587
588    await _drain(audio.get_media_stream(streamdetails, _make_pcm_format(), seek_position=90))
589
590    # a provider that can seek receives the position and the local one is cleared,
591    # so only the remaining audio reaches ffmpeg
592    provider.get_audio_stream.assert_called_once_with(streamdetails, seek_position=90)
593    assert patch_two_minute_ffmpeg.last_instance is not None
594    assert "-ss" not in (patch_two_minute_ffmpeg.last_instance.extra_input_args or [])
595    assert streamdetails.duration == 240
596
597
598@pytest.mark.asyncio
599async def test_get_media_stream_keeps_caller_extra_input_args_intact(
600    patch_ffmpeg: type[_FakeFFMpeg],
601) -> None:
602    """Per-call input args must not leak back onto the caller's StreamDetails."""
603    streamdetails = _seekable_streamdetails()
604    audio = _make_audio_controller()
605
606    await _drain(audio.get_media_stream(streamdetails, _make_pcm_format(), seek_position=30))
607
608    assert patch_ffmpeg.last_instance is not None
609    assert patch_ffmpeg.last_instance.extra_input_args == [*_PROVIDER_INPUT_ARGS, "-ss", "30"]
610    assert streamdetails.extra_input_args == [*_PROVIDER_INPUT_ARGS]
611
612    # StreamDetails are cached on the queue item and reach this method again on a
613    # retry, another seek or from the background analyzer: every call must build its
614    # args from the provider's list alone instead of stacking onto the previous call's.
615    await _drain(audio.get_media_stream(streamdetails, _make_pcm_format(), seek_position=600))
616
617    assert patch_ffmpeg.last_instance.extra_input_args == [*_PROVIDER_INPUT_ARGS, "-ss", "600"]
618    assert streamdetails.extra_input_args == [*_PROVIDER_INPUT_ARGS]
619
620
621@pytest.mark.asyncio
622@pytest.mark.parametrize("stream_type", [StreamType.HTTP, StreamType.CUSTOM])
623@pytest.mark.usefixtures("patch_ffmpeg")
624async def test_music_provider_slot_covers_http_and_custom_until_eof(
625    stream_type: StreamType,
626) -> None:
627    """HTTP and CUSTOM sources hold one provider slot until their source reaches EOF."""
628    provider = _limited_provider()
629    audio = _make_audio_controller()
630    cast("MagicMock", audio.mass).get_provider.return_value = provider
631    streamdetails = StreamDetails(
632        provider=provider.instance_id,
633        item_id="track-1",
634        audio_format=AudioFormat(content_type=ContentType.MP3),
635        media_type=MediaType.TRACK,
636        stream_type=stream_type,
637        path="http://test.invalid/track.mp3" if stream_type == StreamType.HTTP else None,
638    )
639    stream = audio.get_media_stream(streamdetails, _make_pcm_format())
640
641    await anext(stream)
642    assert not provider.has_available_stream_slot
643    await _drain(stream)
644
645    cast("MagicMock", audio.mass).get_provider.assert_any_call(
646        provider.instance_id, return_unavailable=True
647    )
648    assert provider.has_available_stream_slot
649
650
651@pytest.mark.asyncio
652@pytest.mark.usefixtures("patch_ffmpeg")
653async def test_music_provider_slot_is_acquired_before_hls_resolution(
654    monkeypatch: pytest.MonkeyPatch,
655) -> None:
656    """HLS playlist resolution runs inside the provider source lease."""
657    provider = _limited_provider()
658    audio = _make_audio_controller()
659    cast("MagicMock", audio.mass).get_provider.return_value = provider
660
661    async def _get_hls_substream(_url: str) -> SimpleNamespace:
662        assert not provider.has_available_stream_slot
663        return SimpleNamespace(path="http://test.invalid/media.m3u8")
664
665    monkeypatch.setattr(audio, "get_hls_substream", _get_hls_substream)
666    streamdetails = StreamDetails(
667        provider=provider.instance_id,
668        item_id="track-1",
669        audio_format=AudioFormat(content_type=ContentType.AAC),
670        media_type=MediaType.TRACK,
671        stream_type=StreamType.HLS,
672        path="http://test.invalid/master.m3u8",
673    )
674
675    await _drain(audio.get_media_stream(streamdetails, _make_pcm_format()))
676
677    assert provider.has_available_stream_slot
678
679
680@pytest.mark.asyncio
681@pytest.mark.usefixtures("patch_ffmpeg")
682async def test_music_provider_without_free_slot_reports_stream_limit() -> None:
683    """A second source on a one-slot provider fails with a typed capacity error."""
684    provider = _limited_provider()
685    audio = _make_audio_controller()
686    cast("MagicMock", audio.mass).get_provider.return_value = provider
687    streamdetails = _provider_http_streamdetails(provider)
688    active_stream = audio.get_media_stream(streamdetails, _make_pcm_format())
689    await anext(active_stream)
690
691    with pytest.raises(ProviderStreamLimitError):
692        await _drain(
693            audio.get_media_stream(streamdetails, _make_pcm_format(), source_wait_timeout=0)
694        )
695
696    await active_stream.aclose()
697    assert provider.has_available_stream_slot
698
699
700def _unavailable_owner_with_sibling() -> tuple[_LimitedProvider, MagicMock, MagicMock]:
701    """Return an unavailable owner, a same-domain sibling, and a real get_provider double."""
702    owner = _limited_provider()
703    owner.available = False
704    sibling = MagicMock()
705    sibling.instance_id = "limited--2"
706
707    def _get_provider(_instance: str, return_unavailable: bool = False, **_kwargs: Any) -> Any:
708        # mirrors mass.get_provider: an unavailable streaming instance falls back to its domain
709        return owner if return_unavailable else sibling
710
711    lookup = MagicMock(side_effect=_get_provider)
712    return owner, sibling, lookup
713
714
715@pytest.mark.asyncio
716@pytest.mark.usefixtures("patch_ffmpeg")
717async def test_custom_source_never_streams_from_a_sibling_of_the_charged_instance() -> None:
718    """The slot is charged to the instance that issued the details, so it must serve them too."""
719    owner, sibling, lookup = _unavailable_owner_with_sibling()
720    audio = _make_audio_controller()
721    cast("MagicMock", audio.mass).get_provider = lookup
722    streamdetails = StreamDetails(
723        provider=owner.instance_id,
724        item_id="track-1",
725        audio_format=AudioFormat(content_type=ContentType.MP3),
726        media_type=MediaType.TRACK,
727        stream_type=StreamType.CUSTOM,
728    )
729
730    with pytest.raises(ProviderUnavailableError):
731        await _drain(audio.get_media_stream(streamdetails, _make_pcm_format()))
732
733    sibling.get_audio_stream.assert_not_called()
734    assert owner.has_available_stream_slot
735
736
737def test_audio_source_generator_never_opens_a_sibling_of_the_charged_instance() -> None:
738    """The AudioSource entry point pins the same instance as the regular source path."""
739    owner, sibling, lookup = _unavailable_owner_with_sibling()
740    audio = _make_audio_controller()
741    cast("MagicMock", audio.mass).get_provider = lookup
742    streamdetails = StreamDetails(
743        provider=owner.instance_id,
744        item_id="source-1",
745        audio_format=AudioFormat(content_type=ContentType.MP3),
746        media_type=MediaType.AUDIO_SOURCE,
747        stream_type=StreamType.CUSTOM,
748    )
749
750    with pytest.raises(ProviderUnavailableError):
751        audio._open_audio_source_generator(streamdetails)
752
753    sibling.get_audio_stream.assert_not_called()
754
755
756@pytest.mark.asyncio
757@pytest.mark.usefixtures("patch_ffmpeg")
758async def test_non_music_provider_source_takes_no_slot() -> None:
759    """Sources owned by a plugin provider stream without any capacity handling."""
760    plugin_provider = MagicMock()
761    audio = _make_audio_controller()
762    cast("MagicMock", audio.mass).get_provider.return_value = plugin_provider
763
764    await _drain(
765        audio.get_media_stream(
766            _provider_http_streamdetails(_limited_provider()), _make_pcm_format()
767        )
768    )
769
770    plugin_provider.acquire_stream_slot.assert_not_called()
771
772
773@pytest.mark.asyncio
774async def test_music_provider_slot_releases_on_source_error(
775    monkeypatch: pytest.MonkeyPatch,
776) -> None:
777    """A source startup error releases the provider slot."""
778    monkeypatch.setattr(audio_mod, "FFMpeg", _FailingStartFFMpeg)
779    provider = _limited_provider()
780    audio = _make_audio_controller()
781    cast("MagicMock", audio.mass).get_provider.return_value = provider
782
783    with pytest.raises(AudioError):
784        await _drain(
785            audio.get_media_stream(_provider_http_streamdetails(provider), _make_pcm_format())
786        )
787
788    assert provider.has_available_stream_slot
789
790
791@pytest.mark.asyncio
792async def test_music_provider_slot_releases_on_cancellation(
793    monkeypatch: pytest.MonkeyPatch,
794) -> None:
795    """Cancelling a stalled source closes the provider lease."""
796    monkeypatch.setattr(audio_mod, "FFMpeg", _StallingFFMpeg)
797    provider = _limited_provider()
798    audio = _make_audio_controller()
799    cast("MagicMock", audio.mass).get_provider.return_value = provider
800    stream = audio.get_media_stream(_provider_http_streamdetails(provider), _make_pcm_format())
801    read_task = asyncio.create_task(anext(stream))
802    await asyncio.sleep(0)
803    assert not provider.has_available_stream_slot
804
805    read_task.cancel()
806    with suppress(asyncio.CancelledError):
807        await read_task
808
809    assert provider.has_available_stream_slot
810
811
812@pytest.mark.asyncio
813async def test_audio_buffer_clear_closes_provider_slot(
814    monkeypatch: pytest.MonkeyPatch,
815) -> None:
816    """AudioBuffer cancellation closes the source generator and releases its provider slot."""
817    monkeypatch.setattr(audio_mod, "FFMpeg", _StallingFFMpeg)
818    provider = _limited_provider()
819    audio = _make_audio_controller()
820    cast("MagicMock", audio.mass).get_provider.return_value = provider
821    audio_buffer = AudioBuffer(_make_pcm_format())
822    audio_buffer.fill(
823        audio.get_media_stream(_provider_http_streamdetails(provider), _make_pcm_format()),
824        source_name="limited",
825    )
826    await asyncio.sleep(0)
827    assert not provider.has_available_stream_slot
828
829    await audio_buffer.clear()
830
831    assert provider.has_available_stream_slot
832
833
834@pytest.mark.asyncio
835async def test_provider_capacity_error_from_ffmpeg_feeder_remains_typed(
836    monkeypatch: pytest.MonkeyPatch,
837) -> None:
838    """A capacity error raised by a nested source survives the ffmpeg stage."""
839    provider = _limited_provider()
840    _FeederErrorFFMpeg.error = ProviderStreamLimitError(provider, 5)
841    monkeypatch.setattr(audio_mod, "FFMpeg", _FeederErrorFFMpeg)
842    audio = _make_audio_controller()
843    cast("MagicMock", audio.mass).get_provider.return_value = MagicMock()
844
845    with pytest.raises(ProviderStreamLimitError):
846        await _drain(
847            audio.get_media_stream(
848                _provider_http_streamdetails(provider),
849                _make_pcm_format(),
850            )
851        )
852
853
854@pytest.mark.asyncio
855async def test_custom_audio_source_failure_survives_ffmpeg_path(
856    monkeypatch: pytest.MonkeyPatch,
857) -> None:
858    """A CUSTOM AudioSource failure after PCM output reaches the consumer."""
859    monkeypatch.setattr(audio_mod, "FFMpeg", _SourceConsumingFFMpeg)
860    audio = _make_audio_controller()
861
862    async def _source() -> AsyncGenerator[bytes]:
863        yield b"\x00\x01" * 256
864        raise RuntimeError("source failed")
865
866    provider = MagicMock(available=True)
867    provider.get_audio_stream.return_value = _source()
868    cast("MagicMock", audio.mass).get_provider.return_value = provider
869    streamdetails = _flac_streamdetails()
870    streamdetails.stream_type = StreamType.CUSTOM
871    streamdetails.decoded_audio_format = _make_pcm_format()
872
873    with pytest.raises(AudioError, match="source failed") as err:
874        await _drain(audio.get_audio_source_stream(streamdetails, _make_pcm_format()))
875
876    assert isinstance(err.value.__cause__, RuntimeError)
877
878
879@pytest.mark.asyncio
880async def test_ffmpeg_error_path_prefers_source_failure(
881    monkeypatch: pytest.MonkeyPatch,
882) -> None:
883    """A feeder failure remains the cause when ffmpeg also stalls."""
884    _StallingFeederErrorFFMpeg.error = RuntimeError("source failed")
885    monkeypatch.setattr(audio_mod, "FFMpeg", _StallingFeederErrorFFMpeg)
886    monkeypatch.setattr(audio_mod, "STREAM_START_TIMEOUT", 0.1)
887    audio = _make_audio_controller()
888
889    with pytest.raises(AudioError, match="source failed") as err:
890        await _drain(audio.get_media_stream(_flac_streamdetails(), _make_pcm_format()))
891
892    assert isinstance(err.value.__cause__, RuntimeError)
893
894
895@pytest.mark.asyncio
896async def test_get_media_stream_adds_realtime_pacing_for_audio_source(
897    patch_ffmpeg: type[_FakeFFMpeg],
898) -> None:
899    """A live AudioSource gets realtime pacing with a small initial burst of headroom."""
900    audio = _make_audio_controller()
901    await _drain(audio.get_media_stream(_flac_streamdetails(), _make_pcm_format()))
902
903    assert patch_ffmpeg.last_instance is not None
904    assert patch_ffmpeg.last_instance.extra_input_args == [
905        "-readrate",
906        "1",
907        "-readrate_initial_burst",
908        "0.5",
909    ]
910
911
912@pytest.mark.asyncio
913@pytest.mark.parametrize(
914    "provider_pacing_args",
915    [["-readrate", "1.0", "-readrate_initial_burst", "2"], ["-re"]],
916    ids=["readrate", "re"],
917)
918async def test_get_media_stream_respects_provider_pacing_args(
919    patch_ffmpeg: type[_FakeFFMpeg],
920    provider_pacing_args: list[str],
921) -> None:
922    """Provider-supplied -re/-readrate args suppress the automatic AudioSource pacing."""
923    streamdetails = _flac_streamdetails(extra_input_args=list(provider_pacing_args))
924    audio = _make_audio_controller()
925    await _drain(audio.get_media_stream(streamdetails, _make_pcm_format()))
926
927    assert patch_ffmpeg.last_instance is not None
928    assert patch_ffmpeg.last_instance.extra_input_args == provider_pacing_args
929