/
/
/
1"""Tests for SharedGroupStream."""
2
3from __future__ import annotations
4
5import asyncio
6from collections.abc import AsyncIterator
7from typing import TYPE_CHECKING
8from unittest.mock import AsyncMock, MagicMock, Mock, patch
9
10import pytest
11from music_assistant_models.enums import ContentType
12from music_assistant_models.media_items import AudioFormat
13
14from music_assistant.providers.msx_bridge.http_server import MSXHTTPServer
15from music_assistant.providers.msx_bridge.player import MSXPlayer
16from music_assistant.providers.msx_bridge.provider import SharedGroupStream
17
18if TYPE_CHECKING:
19 from music_assistant.providers.msx_bridge.provider import MSXBridgeProvider
20
21
22async def _chunks(*data: bytes) -> AsyncIterator[bytes]:
23 """Yield bytes chunks as an async iterator."""
24 for chunk in data:
25 yield chunk
26
27
28async def _collect(stream: SharedGroupStream, player_id: str) -> list[bytes]:
29 """Subscribe and collect all chunks into a list."""
30 result = []
31 async for chunk in stream.subscribe(player_id):
32 result.append(chunk)
33 return result
34
35
36async def test_subscribe_receives_all_chunks() -> None:
37 """Subscriber should receive every chunk produced."""
38 stream = SharedGroupStream("g1", "uri://test")
39 await stream.start(_chunks(b"a", b"b", b"c"))
40 result = await _collect(stream, "tv1")
41 assert result == [b"a", b"b", b"c"]
42
43
44async def test_late_joiner_after_finish_does_not_hang() -> None:
45 """
46 subscribe() called after producer has already finished must not block indefinitely.
47
48 Regression test for: https://github.com/music-assistant/server/pull/3123#discussion_r2842897555
49 """
50 stream = SharedGroupStream("g1", "uri://test")
51 await stream.start(_chunks(b"x", b"y"))
52
53 # Wait for the producer to fully finish before subscribing.
54 assert stream.producer_task is not None
55 await asyncio.wait_for(stream.producer_task, timeout=5.0)
56 assert stream.finished is True
57
58 # Late subscriber: must get catch-up data and exit cleanly (no hang).
59 result = await asyncio.wait_for(_collect(stream, "late"), timeout=5.0)
60 assert result == [b"x", b"y"]
61
62
63async def test_late_joiner_with_empty_stream_does_not_hang() -> None:
64 """Late joiner on a stream that produced zero chunks must also exit cleanly."""
65 stream = SharedGroupStream("g1", "uri://test")
66 await stream.start(_chunks()) # no chunks
67
68 assert stream.producer_task is not None
69 await asyncio.wait_for(stream.producer_task, timeout=5.0)
70 assert stream.finished is True
71
72 result = await asyncio.wait_for(_collect(stream, "late"), timeout=5.0)
73 assert result == []
74
75
76async def test_concurrent_subscribers_receive_live_chunks() -> None:
77 """Multiple subscribers joining before stream starts all receive all chunks."""
78 stream = SharedGroupStream("g1", "uri://test")
79
80 async def slow_source() -> AsyncIterator[bytes]:
81 for i in range(3):
82 await asyncio.sleep(0)
83 yield bytes([i])
84
85 await stream.start(slow_source())
86
87 results = await asyncio.gather(
88 _collect(stream, "tv1"),
89 _collect(stream, "tv2"),
90 )
91 assert results[0] == results[1]
92 assert len(results[0]) == 3
93
94
95async def test_concurrent_replace_yields_single_stream(provider: MSXBridgeProvider) -> None:
96 """
97 Concurrent replacing get_or_create_shared_stream calls must yield ONE stream.
98
99 Without serialization, both callers pass the "existing" check while the old
100 producer is being awaited, each creates its own stream, and the loser's
101 ffmpeg producer is orphaned â consuming audio with zero subscribers.
102 """
103
104 async def infinite_source() -> AsyncIterator[bytes]:
105 while True:
106 await asyncio.sleep(0.01)
107 yield b"chunk"
108
109 old = await provider.get_or_create_shared_stream("g1", "uri://old", infinite_source())
110 try:
111 results = await asyncio.gather(
112 provider.get_or_create_shared_stream("g1", "uri://new", infinite_source()),
113 provider.get_or_create_shared_stream("g1", "uri://new", infinite_source()),
114 )
115 assert results[0] is results[1]
116 assert provider._shared_streams["g1"] is results[0]
117 finally:
118 await old.stop()
119 for stream in {id(s): s for s in provider._shared_streams.values()}.values():
120 await stream.stop()
121 await asyncio.gather(
122 *(s.stop() for s in results),
123 return_exceptions=True,
124 )
125
126
127async def test_cancel_stops_subscription() -> None:
128 """Cancelling a subscriber's task cleans up the subscriber registry."""
129
130 async def infinite_source() -> AsyncIterator[bytes]:
131 while True:
132 await asyncio.sleep(0.01)
133 yield b"chunk"
134
135 stream = SharedGroupStream("g1", "uri://test")
136 await stream.start(infinite_source())
137
138 task = asyncio.create_task(_collect(stream, "tv1"))
139 await asyncio.sleep(0.05) # let subscriber register and receive some chunks
140 task.cancel()
141 with pytest.raises(asyncio.CancelledError):
142 await task
143
144 # After cancel, subscriber should be cleaned up
145 assert "tv1" not in stream.subscribers
146
147 # Cleanup
148 if stream.producer_task:
149 stream.producer_task.cancel()
150 with pytest.raises(asyncio.CancelledError):
151 await stream.producer_task
152
153
154async def test_shared_stream_paces_output(provider: MSXBridgeProvider, mass_mock: Mock) -> None:
155 """The shared group encoder carries the same pacing ceiling as the per-player one."""
156 server = MSXHTTPServer(provider, 0)
157 player = MagicMock(spec=MSXPlayer)
158 player.player_id = "msx_leader"
159 media = Mock(source_id=None, queue_item_id=None)
160
161 mass_mock.streams = Mock()
162 mass_mock.streams.get_stream = Mock(return_value=_chunks(b"pcm"))
163 mass_mock.streams.audio.get_player_output_plan = Mock(return_value=Mock(filter_params=[]))
164 provider.get_or_create_shared_stream = AsyncMock( # type: ignore[method-assign]
165 side_effect=RuntimeError("stop here")
166 )
167
168 pcm = AudioFormat(content_type=ContentType.PCM_S16LE)
169 out = AudioFormat(content_type=ContentType.MP3)
170 with (
171 patch(
172 "music_assistant.providers.msx_bridge.http_server.get_ffmpeg_stream",
173 return_value=_chunks(b"encoded"),
174 ) as ffmpeg_mock,
175 pytest.raises(RuntimeError),
176 ):
177 # leader path: player_id == group_id
178 await server._serve_shared_stream(Mock(), player, media, "msx_leader", pcm, out, {})
179
180 extra_args = ffmpeg_mock.call_args.kwargs["extra_input_args"]
181 assert "-readrate" in extra_args
182 assert "-readrate_initial_burst" in extra_args
183