music-assistant-server

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