/
/
/
1"""Fixtures for Snapcast provider tests."""
2
3from __future__ import annotations
4
5from typing import Any
6from unittest.mock import AsyncMock, MagicMock
7
8import pytest
9
10from music_assistant.providers.snapcast.constants import DEFAULT_SNAPCAST_FORMAT
11
12
13class FakeSnapstream:
14 """Minimal Snapstream stand-in for testing ma_stream._register_tcp_server_source."""
15
16 def __init__(
17 self,
18 identifier: str,
19 name: str,
20 path: str = "",
21 sampleformat: str = "48000:16:2",
22 packed_s24le: bool = False,
23 ) -> None:
24 """Initialize with stream identifier, name, and optional path."""
25 self.identifier = identifier
26 self.name = name
27 self.path = path
28 query: dict[str, Any] = {"sampleformat": sampleformat}
29 if packed_s24le:
30 query["packed_s24le"] = "true"
31 self._stream: dict[str, Any] = {"uri": {"host": "127.0.0.1", "query": query}}
32 self._callback = None
33
34 def set_callback(self, cb: Any) -> None:
35 """Register a callback for stream events."""
36 self._callback = cb
37
38
39class FakeSnapserver:
40 """Configurable mock implementing the subset of SnapserverProto used by ma_stream."""
41
42 def __init__(self) -> None:
43 """Initialize with empty stream registry and response queues."""
44 # streams added via stream_add_stream are kept here keyed by id
45 self._streams_by_id: dict[str, FakeSnapstream] = {}
46 # responses to return for each call to stream_add_stream, in order
47 # Each entry is a dict that looks like the snapserver result OR error payload.
48 self._add_stream_responses: list[dict[str, Any]] = []
49 # Calls captured for assertions
50 self.add_stream_calls: list[str] = []
51
52 self.stream_add_stream = AsyncMock(side_effect=self._add_stream_impl)
53 self.stream_remove_stream = AsyncMock(side_effect=self._remove_stream_impl)
54 self.status = AsyncMock(side_effect=self._status_impl)
55 self._status_payload: dict[str, Any] | None = None
56
57 async def _remove_stream_impl(self, stream_id: str) -> None:
58 self._streams_by_id.pop(stream_id, None)
59
60 @property
61 def streams(self) -> list[FakeSnapstream]:
62 """Return all registered streams."""
63 return list(self._streams_by_id.values())
64
65 def stream(self, identifier: str) -> FakeSnapstream:
66 """Return the stream with the given identifier."""
67 return self._streams_by_id[identifier]
68
69 def synchronize(self, status: dict[str, Any]) -> None:
70 """Populate _streams_by_id from a status payload, simulating real Snapserver."""
71 streams = status.get("server", {}).get("streams", [])
72 for s in streams:
73 sid = s.get("id")
74 if not sid:
75 continue
76 if sid not in self._streams_by_id:
77 uri = s.get("uri", {}) if isinstance(s.get("uri"), dict) else {}
78 query = uri.get("query", {}) if isinstance(uri.get("query"), dict) else {}
79 self._streams_by_id[sid] = FakeSnapstream(
80 sid,
81 name=s.get("name", sid),
82 sampleformat=str(query.get("sampleformat", "48000:16:2")),
83 packed_s24le=str(query.get("packed_s24le", "")).lower() in {"1", "true", "yes"},
84 )
85
86 async def _status_impl(self) -> tuple[Any, Any]:
87 # status() in the real Snapserver is async and returns (result, error)
88 return (self._status_payload, None)
89
90 async def _add_stream_impl(self, stream_uri: str) -> dict[str, Any]:
91 self.add_stream_calls.append(stream_uri)
92 if self._add_stream_responses:
93 response = self._add_stream_responses.pop(0)
94 else:
95 response = {
96 "code": -32603,
97 "data": "Generic snapserver error",
98 "message": "Internal error",
99 }
100 # On a successful add, also register the stream in our internal store
101 if isinstance(response, dict) and "id" in response:
102 stream_id = response["id"]
103 # parse name= from uri for realism
104 name = stream_uri.split("&name=", 1)[1] if "&name=" in stream_uri else stream_id
105 self._streams_by_id[stream_id] = FakeSnapstream(stream_id, name, path=stream_uri)
106 return response
107
108 def queue_response(self, response: dict[str, Any]) -> None:
109 """Queue a response that the next stream_add_stream call will return."""
110 self._add_stream_responses.append(response)
111
112 def queue_success(self, stream_id: str = "stream-1") -> None:
113 """Queue a successful stream_add_stream response with the given stream id."""
114 self.queue_response({"id": stream_id})
115
116 def queue_name_collision(self) -> None:
117 """Queue a name-already-exists error response."""
118 self.queue_response(
119 {
120 "code": -32603,
121 "data": 'Stream with name "x" already exists',
122 "message": "Internal error",
123 }
124 )
125
126 def queue_other_error(self, data: str = "bind: Address already in use") -> None:
127 """Queue a generic (non-name-collision) error response."""
128 self.queue_response({"code": -32603, "data": data, "message": "Internal error"})
129
130 def cache_stream_directly(
131 self,
132 stream_id: str,
133 name: str,
134 *,
135 sampleformat: str = "48000:16:2",
136 packed_s24le: bool = False,
137 ) -> FakeSnapstream:
138 """
139 Pre-register a stream in the local cache (skipping the status round-trip).
140
141 Use this when the test wants the orphan to be visible WITHOUT requiring
142 adopt() to call status() + synchronize() first.
143 """
144 s = FakeSnapstream(
145 stream_id,
146 name,
147 sampleformat=sampleformat,
148 packed_s24le=packed_s24le,
149 )
150 self._streams_by_id[stream_id] = s
151 return s
152
153 def stage_orphan_stream(
154 self,
155 stream_id: str,
156 name: str,
157 *,
158 sampleformat: str = "48000:16:2",
159 packed_s24le: bool = False,
160 ) -> None:
161 """
162 Stage an orphan stream that only appears after .synchronize(.status()).
163
164 Use this when the test wants to verify that adopt() falls back to a
165 status-driven resync (the actual Bug B post-MA-restart scenario).
166 The stream is NOT in _streams_by_id until synchronize() is called.
167 """
168 query: dict[str, Any] = {"sampleformat": sampleformat}
169 if packed_s24le:
170 query["packed_s24le"] = "true"
171 self._status_payload = {
172 "server": {
173 "streams": [
174 {
175 "id": stream_id,
176 "name": name,
177 "uri": {"query": query},
178 }
179 ],
180 "groups": [],
181 }
182 }
183
184
185@pytest.fixture
186def fake_snapserver() -> FakeSnapserver:
187 """Provide a fresh FakeSnapserver for each test."""
188 return FakeSnapserver()
189
190
191@pytest.fixture
192def fake_provider(fake_snapserver: FakeSnapserver) -> MagicMock:
193 """Provide a minimal SnapCastProvider mock wired to fake_snapserver."""
194 provider = MagicMock()
195 provider._snapserver = fake_snapserver
196 provider._snapcast_stream_idle_threshold = 60000
197 provider._use_builtin_server = False
198 provider.stream_audio_format = DEFAULT_SNAPCAST_FORMAT
199 provider.logger = MagicMock()
200 return provider
201