music-assistant-server

1"""Tests for the direct-PCM stream helper on the streams controller."""
2
3from __future__ import annotations
4
5from collections.abc import AsyncGenerator
6from typing import Any, cast
7from unittest.mock import MagicMock
8
9import pytest
10from music_assistant_models.enums import ContentType, MediaType, StreamType
11from music_assistant_models.media_items import AudioFormat
12from music_assistant_models.player import PlayerMedia
13from music_assistant_models.queue_item import QueueItem
14from music_assistant_models.streamdetails import StreamDetails
15
16from music_assistant.controllers.streams.controller import StreamsController
17
18PCM_FORMAT = AudioFormat(
19    content_type=ContentType.PCM_S16LE,
20    sample_rate=44100,
21    bit_depth=16,
22    channels=2,
23)
24
25QUEUE_ID = "player-1"
26QUEUE_ITEM_ID = "item-1"
27
28
29def _queue_item(seek_position: int) -> QueueItem:
30    """Build an audiobook queue item whose streamdetails carry the given seek position."""
31    return QueueItem(
32        queue_id=QUEUE_ID,
33        queue_item_id=QUEUE_ITEM_ID,
34        name="Some Audiobook",
35        duration=7200,
36        streamdetails=StreamDetails(
37            provider="builtin",
38            item_id="book-1",
39            audio_format=PCM_FORMAT,
40            media_type=MediaType.AUDIOBOOK,
41            stream_type=StreamType.HTTP,
42            path="http://example.com/book.mp3",
43            duration=7200,
44            can_seek=True,
45            allow_seek=True,
46            queue_id=QUEUE_ID,
47            seek_position=seek_position,
48        ),
49    )
50
51
52def _controller(queue_item: QueueItem) -> tuple[StreamsController, dict[str, Any]]:
53    """
54    Build a streams controller that records the kwargs of the single item stream call.
55
56    The queue itself is absent, which keeps the request on the non-flow (single item)
57    branch: crossfade and audio overlay both need a queue to force flow mode.
58
59    :param queue_item: The item the controller resolves the stream request to.
60    """
61    mass = MagicMock()
62    mass.config.get_raw_core_config_value.return_value = "GLOBAL"
63    mass.player_queues.get.return_value = None
64    mass.player_queues.get_item.return_value = queue_item
65    controller = StreamsController(mass)
66    call_kwargs: dict[str, Any] = {}
67
68    def _record(**kwargs: Any) -> object:
69        call_kwargs.update(kwargs)
70        return object()
71
72    controller.audio = MagicMock()
73    controller.audio.get_queue_item_stream.side_effect = _record
74    return controller, call_kwargs
75
76
77@pytest.mark.parametrize("seek_position", [1800, 0])
78def test_single_item_stream_forwards_the_seek_position(seek_position: int) -> None:
79    """A resumed (or seeked) item is streamed from its seek position, not from the start."""
80    queue_item = _queue_item(seek_position)
81    controller, call_kwargs = _controller(queue_item)
82
83    controller.get_stream(
84        PlayerMedia(
85            uri="library://audiobook/1",
86            media_type=MediaType.AUDIOBOOK,
87            source_id=QUEUE_ID,
88            queue_item_id=QUEUE_ITEM_ID,
89        ),
90        PCM_FORMAT,
91    )
92
93    assert call_kwargs["seek_position"] == seek_position
94
95
96async def _pcm_chunks() -> AsyncGenerator[bytes]:
97    """Yield two PCM chunks so the gauge can be observed mid-stream."""
98    yield b"chunk-1"
99    yield b"chunk-2"
100
101
102def _pcm_stream_controller() -> StreamsController:
103    """Build a controller whose single item stream yields real PCM chunks."""
104    controller, _ = _controller(_queue_item(0))
105    audio = cast("Any", controller.audio)
106    audio.get_queue_item_stream.side_effect = lambda **_kwargs: _pcm_chunks()
107    return controller
108
109
110@pytest.mark.asyncio
111async def test_direct_pcm_stream_counts_as_an_active_output_stream() -> None:
112    """A player consuming raw PCM registers as playing, so analysis yields CPU to it."""
113    controller = _pcm_stream_controller()
114
115    stream = controller.get_stream(
116        PlayerMedia(
117            uri="library://audiobook/1",
118            media_type=MediaType.AUDIOBOOK,
119            source_id=QUEUE_ID,
120            queue_item_id=QUEUE_ITEM_ID,
121        ),
122        PCM_FORMAT,
123    )
124    assert controller.output_stream_active() is False
125
126    chunks = []
127    async for chunk in stream:
128        chunks.append(chunk)
129        assert controller.output_stream_active() is True
130
131    assert chunks == [b"chunk-1", b"chunk-2"]
132    assert controller.output_stream_active() is False
133
134
135@pytest.mark.asyncio
136async def test_abandoned_direct_pcm_stream_releases_the_gauge() -> None:
137    """A consumer that stops mid-stream releases its count, so analysis regains its budget."""
138    controller = _pcm_stream_controller()
139
140    stream = controller.get_stream(
141        PlayerMedia(
142            uri="library://audiobook/1",
143            media_type=MediaType.AUDIOBOOK,
144            source_id=QUEUE_ID,
145            queue_item_id=QUEUE_ITEM_ID,
146        ),
147        PCM_FORMAT,
148    )
149    assert await anext(stream) == b"chunk-1"
150    assert controller.output_stream_active() is True
151
152    await stream.aclose()
153    assert controller.output_stream_active() is False
154