/
/
/
1"""
2Tests that an audio buffer records the playback session that asked for it.
3
4A queue stop releases only the buffers of the session it is tearing down, so the claim has
5to be made where the buffer is attached. Stream details outlive a stop and are reused as
6they are, so the session they were originally resolved for says nothing about who is
7filling them now.
8"""
9
10from __future__ import annotations
11
12from typing import TYPE_CHECKING, Any
13from unittest.mock import AsyncMock, MagicMock
14
15from music_assistant_models.enums import ContentType, MediaType, StreamType
16from music_assistant_models.media_items import AudioFormat, ProviderMapping, SoundEffect
17from music_assistant_models.queue_item import QueueItem
18from music_assistant_models.streamdetails import StreamDetails
19
20from music_assistant.controllers.player_queues.state import PlayerQueueData
21from music_assistant.controllers.streams.audio import StreamsAudio
22from music_assistant.controllers.streams.audio_buffer import AudioBuffer
23
24if TYPE_CHECKING:
25 import pytest
26
27QUEUE_ID = "queue-1"
28INSTANCE = "service--1"
29ITEM_ID = "item-1"
30
31
32def _queue_item(stamped_with: str | None) -> QueueItem:
33 """
34 Build a queue item that already carries stream details, so none are resolved.
35
36 :param stamped_with: Session recorded on those details, as an earlier session would
37 have left them, or None for details that never backed a buffer.
38 """
39 media_item = SoundEffect(
40 item_id=ITEM_ID,
41 provider=INSTANCE,
42 name="Effect",
43 provider_mappings={
44 ProviderMapping(
45 item_id=ITEM_ID,
46 provider_domain=INSTANCE.split("--", maxsplit=1)[0],
47 provider_instance=INSTANCE,
48 audio_format=AudioFormat(content_type=ContentType.MP3),
49 )
50 },
51 )
52 queue_item = QueueItem(
53 queue_id=QUEUE_ID,
54 queue_item_id="queue-item-1",
55 name="Effect",
56 duration=30,
57 media_item=media_item,
58 )
59 queue_item.streamdetails = StreamDetails(
60 provider=INSTANCE,
61 item_id=ITEM_ID,
62 audio_format=AudioFormat(content_type=ContentType.MP3),
63 media_type=MediaType.SOUND_EFFECT,
64 stream_type=StreamType.HTTP,
65 path="http://test.invalid/item.mp3",
66 duration=30,
67 queue_id=QUEUE_ID,
68 )
69 queue_item.streamdetails.queue_session_id = stamped_with
70 return queue_item
71
72
73def _audio(session_id: str | None) -> StreamsAudio:
74 """Build a streams-audio controller whose queue is in the given playback session."""
75 mass = MagicMock()
76 mass.player_queues.queue_data_or_none.return_value = (
77 PlayerQueueData(queue=MagicMock(), session_id=session_id)
78 if session_id is not None
79 else None
80 )
81 mass.get_provider.return_value = MagicMock()
82 return StreamsAudio(mass)
83
84
85async def test_the_session_asking_for_a_buffer_claims_it(
86 monkeypatch: pytest.MonkeyPatch,
87) -> None:
88 """Details left stamped by an earlier session move to the session filling them now."""
89 queue_item = _queue_item(stamped_with="sess-1")
90 monkeypatch.setattr(
91 AudioBuffer, "get_buffer", AsyncMock(return_value=MagicMock(spec=AudioBuffer))
92 )
93
94 await _audio("sess-2").get_audio_buffer(queue_item, reason="streaming")
95
96 assert queue_item.streamdetails is not None
97 assert queue_item.streamdetails.queue_session_id == "sess-2"
98
99
100async def test_a_superseded_request_cannot_take_a_live_buffer(
101 monkeypatch: pytest.MonkeyPatch,
102) -> None:
103 """
104 Audio the playing session is filling stays its own, whoever asks for it next.
105
106 The single-item stream route serves a request for a session that is no longer the
107 queue's without rejecting it, and reusing a live buffer must not hand that session
108 the power to release it out from under the one still playing.
109 """
110 queue_item = _queue_item(stamped_with="sess-2")
111 live_buffer = MagicMock(spec=AudioBuffer)
112 queue_item.streamdetails.buffer = live_buffer # type: ignore[union-attr]
113 monkeypatch.setattr(AudioBuffer, "get_buffer", AsyncMock(return_value=live_buffer))
114
115 await _audio("sess-2").get_audio_buffer(queue_item, reason="streaming")
116
117 assert queue_item.streamdetails is not None
118 assert queue_item.streamdetails.queue_session_id == "sess-2"
119
120
121async def test_an_unregistered_queue_claims_nothing(monkeypatch: pytest.MonkeyPatch) -> None:
122 """
123 A queue the controller no longer holds a record for leaves no claim behind.
124
125 An unclaimed buffer is released by any stop, which is what has to happen: leaving it
126 would keep its producer - and the provider's stream slot - alive.
127 """
128 queue_item = _queue_item(stamped_with="sess-1")
129 monkeypatch.setattr(
130 AudioBuffer, "get_buffer", AsyncMock(return_value=MagicMock(spec=AudioBuffer))
131 )
132
133 await _audio(None).get_audio_buffer(queue_item, reason="streaming")
134
135 assert queue_item.streamdetails is not None
136 assert queue_item.streamdetails.queue_session_id is None
137
138
139async def test_the_claim_is_made_before_the_buffer_is_filled(
140 monkeypatch: pytest.MonkeyPatch,
141) -> None:
142 """A producer that starts during the fill is already covered by its session's stop."""
143 queue_item = _queue_item(stamped_with=None)
144 seen: list[str | None] = []
145
146 async def _record_the_claim(**kwargs: Any) -> MagicMock:
147 streamdetails: StreamDetails = kwargs["streamdetails"]
148 seen.append(streamdetails.queue_session_id)
149 return MagicMock(spec=AudioBuffer)
150
151 monkeypatch.setattr(AudioBuffer, "get_buffer", _record_the_claim)
152
153 await _audio("sess-1").get_audio_buffer(queue_item, reason="streaming")
154
155 assert seen == ["sess-1"]
156