/
/
/
1"""Tests for the squeezelite multi-client stream task."""
2
3from __future__ import annotations
4
5import asyncio
6from collections.abc import AsyncGenerator
7from typing import Any
8
9import pytest
10from music_assistant_models.enums import ContentType
11from music_assistant_models.media_items import AudioFormat
12
13from music_assistant.providers.squeezelite.multi_client_stream import MultiClientStream
14
15PCM_FORMAT = AudioFormat(
16 content_type=ContentType.PCM_S16LE,
17 sample_rate=44100,
18 bit_depth=16,
19 channels=2,
20)
21
22
23@pytest.fixture(name="instant_sleep")
24def instant_sleep_fixture(monkeypatch: pytest.MonkeyPatch) -> None:
25 """Make the runner's retry delays elapse immediately while still yielding control."""
26 real_sleep = asyncio.sleep
27
28 async def _sleep(_delay: float, *args: Any, **kwargs: Any) -> Any:
29 return await real_sleep(0, *args, **kwargs)
30
31 monkeypatch.setattr(asyncio, "sleep", _sleep)
32
33
34@pytest.mark.usefixtures("instant_sleep")
35async def test_runner_closes_the_source_when_no_client_connects() -> None:
36 """
37 Giving up on a stream nobody subscribed to tears the source down.
38
39 The source counts as active playback for as long as it is open, so leaving it
40 suspended here would hold audio analysis in its reduced-CPU mode indefinitely.
41 """
42 closed = asyncio.Event()
43
44 async def _source() -> AsyncGenerator[bytes]:
45 try:
46 while True:
47 yield b"chunk"
48 finally:
49 closed.set()
50
51 stream = MultiClientStream(
52 audio_source=_source(),
53 audio_format=PCM_FORMAT,
54 queue_id="queue-1",
55 session_id="session-1",
56 )
57 await stream.task
58
59 assert closed.is_set()
60