/
/
1"""Tests for the direct-PCM stream helper on the streams controller."""
2
3from __future__ import annotations
4
5from typing import Any
6from unittest.mock import MagicMock
7
8import pytest
9from music_assistant_models.enums import ContentType, MediaType, StreamType
10from music_assistant_models.media_items import AudioFormat
11from music_assistant_models.player import PlayerMedia
12from music_assistant_models.queue_item import QueueItem
13from music_assistant_models.streamdetails import StreamDetails
14
15from music_assistant.controllers.streams.controller import StreamsController
16
17PCM_FORMAT = AudioFormat(
18 content_type=ContentType.PCM_S16LE,
19 sample_rate=44100,
20 bit_depth=16,
21 channels=2,
22)
23
24QUEUE_ID = "player-1"
25QUEUE_ITEM_ID = "item-1"
26
27
28def _queue_item(seek_position: int) -> QueueItem:
29 """Build an audiobook queue item whose streamdetails carry the given seek position."""
30 return QueueItem(
31 queue_id=QUEUE_ID,
32 queue_item_id=QUEUE_ITEM_ID,
33 name="Some Audiobook",
34 duration=7200,
35 streamdetails=StreamDetails(
36 provider="builtin",
37 item_id="book-1",
38 audio_format=PCM_FORMAT,
39 media_type=MediaType.AUDIOBOOK,
40 stream_type=StreamType.HTTP,
41 path="http://example.com/book.mp3",
42 duration=7200,
43 can_seek=True,
44 allow_seek=True,
45 queue_id=QUEUE_ID,
46 seek_position=seek_position,
47 ),
48 )
49
50
51def _controller(queue_item: QueueItem) -> tuple[StreamsController, dict[str, Any]]:
52 """
53 Build a streams controller that records the kwargs of the single item stream call.
54
55 The queue itself is absent, which keeps the request on the non-flow (single item)
56 branch: crossfade and audio overlay both need a queue to force flow mode.
57
58 :param queue_item: The item the controller resolves the stream request to.
59 """
60 mass = MagicMock()
61 mass.config.get_raw_core_config_value.return_value = "GLOBAL"
62 mass.player_queues.get.return_value = None
63 mass.player_queues.get_item.return_value = queue_item
64 controller = StreamsController(mass)
65 call_kwargs: dict[str, Any] = {}
66
67 def _record(**kwargs: Any) -> object:
68 call_kwargs.update(kwargs)
69 return object()
70
71 controller.audio = MagicMock()
72 controller.audio.get_queue_item_stream.side_effect = _record
73 return controller, call_kwargs
74
75
76@pytest.mark.parametrize("seek_position", [1800, 0])
77def test_single_item_stream_forwards_the_seek_position(seek_position: int) -> None:
78 """A resumed (or seeked) item is streamed from its seek position, not from the start."""
79 queue_item = _queue_item(seek_position)
80 controller, call_kwargs = _controller(queue_item)
81
82 controller.get_stream(
83 PlayerMedia(
84 uri="library://audiobook/1",
85 media_type=MediaType.AUDIOBOOK,
86 source_id=QUEUE_ID,
87 queue_item_id=QUEUE_ITEM_ID,
88 ),
89 PCM_FORMAT,
90 )
91
92 assert call_kwargs["seek_position"] == seek_position
93