/
/
/
1"""Common test helpers for Music Assistant tests."""
2
3import asyncio
4import contextlib
5import inspect
6import logging
7import pathlib
8from collections.abc import AsyncGenerator, Iterator
9from types import MethodType
10from typing import TYPE_CHECKING, Any
11from unittest.mock import AsyncMock, MagicMock, patch
12
13import aiofiles.os
14from music_assistant_models.enums import EventType, IdentifierType, PlayerFeature, PlayerType
15from music_assistant_models.player import DeviceInfo
16
17from music_assistant.controllers.tasks.constants import TASK_LIFECYCLE_UPDATE_DEBOUNCE
18from music_assistant.mass import MusicAssistant
19from music_assistant.models.player import Player
20
21if TYPE_CHECKING:
22 from music_assistant_models.event import MassEvent
23
24
25def utf8_safe(value: object) -> object:
26 """
27 Return ``value`` with any non-UTF-8-encodable strings made encodable.
28
29 Lone surrogates (e.g. from undecodable filesystem paths) are replaced with
30 their backslash escapes so the value survives strict-UTF-8 serialization.
31 """
32 if isinstance(value, str):
33 try:
34 value.encode()
35 except UnicodeEncodeError:
36 return value.encode("utf-8", "backslashreplace").decode()
37 return value
38 if isinstance(value, list):
39 return [utf8_safe(item) for item in value]
40 if isinstance(value, tuple):
41 return tuple(utf8_safe(item) for item in value)
42 if isinstance(value, dict):
43 return {utf8_safe(key): utf8_safe(item) for key, item in value.items()}
44 return value
45
46
47def _get_fixture_folder(provider: str | None = None) -> pathlib.Path:
48 tests_base = pathlib.Path(__file__).parent
49 if provider:
50 return tests_base / "providers" / provider / "fixtures"
51 return tests_base / "fixtures"
52
53
54async def get_fixtures_dir(
55 subdir: str, provider: str | None = None
56) -> AsyncGenerator[tuple[str, bytes]]:
57 """Yield the contents of every fixture in a fixtures folder."""
58 dir_path = _get_fixture_folder(provider) / subdir
59 for file in await aiofiles.os.listdir(dir_path):
60 async with aiofiles.open(dir_path / file, "rb") as fp:
61 yield (file, await fp.read())
62
63
64@contextlib.contextmanager
65def collect_loop_errors() -> Iterator[list[dict[str, Any]]]:
66 """
67 Capture everything the running loop reports to its exception handler.
68
69 Yields the (initially empty) list the captured contexts are appended to; the loop's
70 own handler is restored on exit. Use it to assert that an operation does not surface
71 an error the server itself already handles, which would otherwise reach the user as
72 an ERROR log entry with a traceback.
73 """
74 loop = asyncio.get_running_loop()
75 previous = loop.get_exception_handler()
76 reported: list[dict[str, Any]] = []
77 loop.set_exception_handler(lambda _loop, context: reported.append(context))
78 try:
79 yield reported
80 finally:
81 loop.set_exception_handler(previous)
82
83
84@contextlib.asynccontextmanager
85async def wait_for_sync_completion(mass: MusicAssistant) -> AsyncGenerator[None]:
86 """Wait for a sync to finish."""
87 flag = asyncio.Event()
88
89 def _event(_event: MassEvent) -> None:
90 flag.set()
91
92 release_cb = mass.subscribe(_event, EventType.MUSIC_SYNC_COMPLETED)
93
94 try:
95 yield
96 finally:
97 try:
98 if mass.music.active_sync_tasks:
99 await flag.wait()
100 finally:
101 release_cb()
102
103
104# the address a fixture's web and stream servers bind to, so a test run never listens
105# on the host's real interfaces
106LOOPBACK_IP = "127.0.0.1"
107
108
109@contextlib.contextmanager
110def use_ephemeral_server_ports() -> Iterator[None]:
111 """
112 Bind a full-server test fixture's web and stream servers to a free loopback port.
113
114 Port 0 has the kernel pick the port during the bind itself, so nothing else can
115 claim it in the meantime.
116
117 Binding loopback keeps a test run off the host's other interfaces and gives each
118 server a single socket, so it has one assigned port: asyncio binds a wildcard
119 address once per address family, each with its own port.
120 """
121 with (
122 patch("music_assistant.controllers.webserver.controller.DEFAULT_SERVER_PORT", 0),
123 patch("music_assistant.controllers.streams.controller.DEFAULT_PORT", 0),
124 patch("music_assistant.controllers.webserver.controller.DEFAULT_HOST", LOOPBACK_IP),
125 patch("music_assistant.controllers.streams.controller.DEFAULT_HOST", LOOPBACK_IP),
126 # keep address detection off the host's real interfaces
127 patch(
128 "music_assistant.controllers.streams.controller.get_ip_addresses",
129 AsyncMock(return_value=(LOOPBACK_IP,)),
130 ),
131 patch(
132 "music_assistant.controllers.streams.controller.get_publish_ip_candidates",
133 AsyncMock(return_value=(LOOPBACK_IP,)),
134 ),
135 patch(
136 "music_assistant.controllers.webserver.controller.get_ip_addresses",
137 AsyncMock(return_value=(LOOPBACK_IP,)),
138 ),
139 patch(
140 "music_assistant.controllers.webserver.controller.get_publish_ip_candidates",
141 AsyncMock(return_value=(LOOPBACK_IP,)),
142 ),
143 ):
144 yield
145
146
147@contextlib.contextmanager
148def suppress_auto_loaded_providers() -> Iterator[None]:
149 """
150 Stop a fixture boot from auto-setting-up providers that reach into the host.
151
152 Keeps a booted test instance isolated from the developer's machine: the default
153 device providers (airplay/chromecast/dlna/...) are not auto-configured.
154 """
155 with patch("music_assistant.mass.DEFAULT_PROVIDERS", ()):
156 yield
157
158
159async def wait_for_boot_to_settle(mass: MusicAssistant) -> None:
160 """
161 Wait out the events a fixture boot leaves in flight.
162
163 A provider finishes loading in a detached task that registers background tasks, and
164 registering one emits a debounced task list, so without this a test can start watching
165 for events in time to catch the tail of its own fixture's boot.
166
167 :param mass: The started instance to settle.
168 """
169 for provider in mass.providers:
170 await provider.initialized.wait()
171 # twice the window: the debounce a registration already armed, plus the tail of the
172 # post-load work that runs after a provider marks itself initialized
173 await asyncio.sleep(TASK_LIFECYCLE_UPDATE_DEBOUNCE * 2)
174
175
176@contextlib.contextmanager
177def suppress_initial_library_sync() -> Iterator[None]:
178 """
179 Hold a fixture boot's music providers to their recurring library sync only.
180
181 The first sync of a freshly loaded provider otherwise runs seconds into the boot, so on
182 a loaded machine it lands in the middle of whatever test is running by then, rewriting
183 the library under it. Tests that want a sync call ``start_sync()`` themselves.
184 """
185 with patch("music_assistant.controllers.music.controller.INITIAL_SYNC_DELAY", None):
186 yield
187
188
189# Mock classes for testing
190
191
192def use_real_create_task(mass: MagicMock | MusicAssistant) -> None:
193 """
194 Give a mocked MusicAssistant the real create_task implementation.
195
196 Needed for any test that lets a `@use_cache` decorated method run, since the
197 decorator awaits the task it gets back to share one fetch between callers.
198
199 :param mass: The mock standing in for the MusicAssistant instance.
200 """
201 mass._tracked_tasks = {}
202 # on an AsyncMock this call would hand back a coroutine that nobody awaits
203 mass.verify_event_loop_thread = MagicMock() # type: ignore[method-assign]
204 real_create_task = MethodType(MusicAssistant.create_task, mass)
205
206 def _create_task(target: Any, *args: Any, **kwargs: Any) -> Any:
207 if not (inspect.iscoroutine(target) or inspect.iscoroutinefunction(target)):
208 # tests hand this mocked methods too, which the real one refuses
209 return MagicMock()
210 # resolved per call so this also works from a synchronous fixture
211 mass.loop = asyncio.get_running_loop()
212 return real_create_task(target, *args, **kwargs)
213
214 # kept a mock so tests can still assert on the calls it received
215 mass.create_task = MagicMock(side_effect=_create_task) # type: ignore[method-assign]
216
217
218def create_mock_config(name: str) -> MagicMock:
219 """Create a mock player config with the given name."""
220 config = MagicMock()
221 config.name = None # No custom name, use default
222 config.default_name = name
223 config.get_value = MagicMock(return_value="none") # Default to no power control
224 return config
225
226
227class MockProvider:
228 """Mock player provider for testing."""
229
230 def __init__(
231 self, domain: str, instance_id: str = "test_instance", mass: MagicMock | None = None
232 ) -> None:
233 """Initialize the mock provider."""
234 self.domain = domain
235 self.instance_id = instance_id
236 self.name = f"Mock {domain.title()}"
237 self.manifest = MagicMock()
238 self.manifest.name = f"Mock {domain} Provider"
239 self.mass = mass or MagicMock()
240 self.dashboards = MagicMock()
241 self.logger = logging.getLogger(f"test.{domain}")
242 self.unloading = False
243 # tests that let their players signal state updates fill this with the
244 # players of this provider, the way a real provider reports them
245 self.players: list[Player] = []
246
247
248class MockPlayer(Player):
249 """Mock player for testing."""
250
251 def __init__(
252 self,
253 provider: MockProvider,
254 player_id: str,
255 name: str,
256 player_type: PlayerType = PlayerType.PLAYER,
257 identifiers: dict[IdentifierType, str] | None = None,
258 ) -> None:
259 """Initialize the mock player."""
260 # Set up the mock config before calling super().__init__
261 # because the parent __init__ accesses config
262 provider.mass.config.get_base_player_config.return_value = create_mock_config(name)
263
264 super().__init__(provider, player_id) # type: ignore[arg-type]
265 self._attr_name = name
266 # Set type as instance attribute (overrides class attribute)
267 self._attr_type = player_type
268 self._attr_available = True
269 self._attr_powered = True
270 self._attr_supported_features = {PlayerFeature.VOLUME_SET}
271 self._attr_can_group_with = set()
272 self._attr_group_members = []
273
274 # Set up device info with identifiers
275 self._attr_device_info = DeviceInfo(
276 model="Test Model",
277 manufacturer="Test Manufacturer",
278 )
279 if identifiers:
280 for conn_type, value in identifiers.items():
281 self._attr_device_info.add_identifier(conn_type, value)
282
283 # Clear cached properties after modifying attributes
284 self._cache.clear()
285
286 async def set_members(
287 self,
288 player_ids_to_add: list[str] | None = None,
289 player_ids_to_remove: list[str] | None = None,
290 ) -> None:
291 """Mock implementation of set_members."""
292 current_members = set(self._attr_group_members)
293
294 if player_ids_to_add:
295 current_members.update(player_ids_to_add)
296
297 if player_ids_to_remove:
298 current_members.difference_update(player_ids_to_remove)
299
300 # Always include self as first member if there are members
301 if current_members:
302 self._attr_group_members = [self.player_id] + [
303 pid for pid in current_members if pid != self.player_id
304 ]
305 else:
306 self._attr_group_members = []
307
308 # Clear cache to reflect changes
309 self._cache.clear()
310
311 async def stop(self) -> None:
312 """Stop playback - required abstract method."""
313
314
315class MockMass:
316 """Type hint for mocked MusicAssistant instance."""
317