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.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# builtin providers that must not be auto-set-up during a fixture boot: local_audio
105# bridges the host machine's sound devices (built-in speakers, bluetooth, ...) as
106# sendspin players, which would leak real hardware into the player registry
107SUPPRESSED_BUILTIN_PROVIDERS = {"local_audio"}
108
109_orig_create_builtin_provider_config = ProviderConfigMixin.create_builtin_provider_config
110
111# the address a fixture's web and stream servers bind to, so a test run never listens
112# on the host's real interfaces
113LOOPBACK_IP = "127.0.0.1"
114
115
116@contextlib.contextmanager
117def use_ephemeral_server_ports() -> Iterator[None]:
118    """
119    Bind a full-server test fixture's web and stream servers to a free loopback port.
120
121    Port 0 has the kernel pick the port during the bind itself, so nothing else can
122    claim it in the meantime.
123
124    Binding loopback keeps a test run off the host's other interfaces and gives each
125    server a single socket, so it has one assigned port: asyncio binds a wildcard
126    address once per address family, each with its own port.
127    """
128    with (
129        patch("music_assistant.controllers.webserver.controller.DEFAULT_SERVER_PORT", 0),
130        patch("music_assistant.controllers.streams.controller.DEFAULT_PORT", 0),
131        patch("music_assistant.controllers.webserver.controller.DEFAULT_HOST", LOOPBACK_IP),
132        patch("music_assistant.controllers.streams.controller.DEFAULT_HOST", LOOPBACK_IP),
133        # keep address detection off the host's real interfaces
134        patch(
135            "music_assistant.controllers.streams.controller.get_ip_addresses",
136            AsyncMock(return_value=(LOOPBACK_IP,)),
137        ),
138        patch(
139            "music_assistant.controllers.streams.controller.get_publish_ip_candidates",
140            AsyncMock(return_value=(LOOPBACK_IP,)),
141        ),
142        patch(
143            "music_assistant.controllers.webserver.controller.get_ip_addresses",
144            AsyncMock(return_value=(LOOPBACK_IP,)),
145        ),
146        patch(
147            "music_assistant.controllers.webserver.controller.get_publish_ip_candidates",
148            AsyncMock(return_value=(LOOPBACK_IP,)),
149        ),
150    ):
151        yield
152
153
154@contextlib.contextmanager
155def suppress_auto_loaded_providers() -> Iterator[None]:
156    """
157    Stop a fixture boot from auto-setting-up providers that reach into the host.
158
159    Keeps a booted test instance isolated from the developer's machine: the default
160    device providers (airplay/chromecast/dlna/...) are not auto-configured, and neither
161    is the builtin local_audio provider, which would otherwise bridge the host's sound
162    devices (built-in speakers, bluetooth, ...) into the player registry.
163    """
164    with (
165        patch("music_assistant.mass.DEFAULT_PROVIDERS", ()),
166        patch.object(
167            ProviderConfigMixin,
168            "create_builtin_provider_config",
169            _create_builtin_provider_config_hermetic,
170        ),
171    ):
172        yield
173
174
175async def _create_builtin_provider_config_hermetic(
176    self: ProviderConfigMixin, provider_domain: str
177) -> None:
178    """Create builtin provider configs, skipping providers that discover host hardware."""
179    if provider_domain in SUPPRESSED_BUILTIN_PROVIDERS:
180        return
181    await _orig_create_builtin_provider_config(self, provider_domain)
182
183
184# Mock classes for testing
185
186
187def use_real_create_task(mass: MagicMock | MusicAssistant) -> None:
188    """
189    Give a mocked MusicAssistant the real create_task implementation.
190
191    Needed for any test that lets a `@use_cache` decorated method run, since the
192    decorator awaits the task it gets back to share one fetch between callers.
193
194    :param mass: The mock standing in for the MusicAssistant instance.
195    """
196    mass._tracked_tasks = {}
197    # on an AsyncMock this call would hand back a coroutine that nobody awaits
198    mass.verify_event_loop_thread = MagicMock()  # type: ignore[method-assign]
199    real_create_task = MethodType(MusicAssistant.create_task, mass)
200
201    def _create_task(target: Any, *args: Any, **kwargs: Any) -> Any:
202        if not (inspect.iscoroutine(target) or inspect.iscoroutinefunction(target)):
203            # tests hand this mocked methods too, which the real one refuses
204            return MagicMock()
205        # resolved per call so this also works from a synchronous fixture
206        mass.loop = asyncio.get_running_loop()
207        return real_create_task(target, *args, **kwargs)
208
209    # kept a mock so tests can still assert on the calls it received
210    mass.create_task = MagicMock(side_effect=_create_task)  # type: ignore[method-assign]
211
212
213def create_mock_config(name: str) -> MagicMock:
214    """Create a mock player config with the given name."""
215    config = MagicMock()
216    config.name = None  # No custom name, use default
217    config.default_name = name
218    config.get_value = MagicMock(return_value="none")  # Default to no power control
219    return config
220
221
222class MockProvider:
223    """Mock player provider for testing."""
224
225    def __init__(
226        self, domain: str, instance_id: str = "test_instance", mass: MagicMock | None = None
227    ) -> None:
228        """Initialize the mock provider."""
229        self.domain = domain
230        self.instance_id = instance_id
231        self.name = f"Mock {domain.title()}"
232        self.manifest = MagicMock()
233        self.manifest.name = f"Mock {domain} Provider"
234        self.mass = mass or MagicMock()
235        self.dashboards = MagicMock()
236        self.logger = logging.getLogger(f"test.{domain}")
237        self.unloading = False
238        # tests that let their players signal state updates fill this with the
239        # players of this provider, the way a real provider reports them
240        self.players: list[Player] = []
241
242
243class MockPlayer(Player):
244    """Mock player for testing."""
245
246    def __init__(
247        self,
248        provider: MockProvider,
249        player_id: str,
250        name: str,
251        player_type: PlayerType = PlayerType.PLAYER,
252        identifiers: dict[IdentifierType, str] | None = None,
253    ) -> None:
254        """Initialize the mock player."""
255        # Set up the mock config before calling super().__init__
256        # because the parent __init__ accesses config
257        provider.mass.config.get_base_player_config.return_value = create_mock_config(name)
258
259        super().__init__(provider, player_id)  # type: ignore[arg-type]
260        self._attr_name = name
261        # Set type as instance attribute (overrides class attribute)
262        self._attr_type = player_type
263        self._attr_available = True
264        self._attr_powered = True
265        self._attr_supported_features = {PlayerFeature.VOLUME_SET}
266        self._attr_can_group_with = set()
267        self._attr_group_members = []
268
269        # Set up device info with identifiers
270        self._attr_device_info = DeviceInfo(
271            model="Test Model",
272            manufacturer="Test Manufacturer",
273        )
274        if identifiers:
275            for conn_type, value in identifiers.items():
276                self._attr_device_info.add_identifier(conn_type, value)
277
278        # Clear cached properties after modifying attributes
279        self._cache.clear()
280
281    async def set_members(
282        self,
283        player_ids_to_add: list[str] | None = None,
284        player_ids_to_remove: list[str] | None = None,
285    ) -> None:
286        """Mock implementation of set_members."""
287        current_members = set(self._attr_group_members)
288
289        if player_ids_to_add:
290            current_members.update(player_ids_to_add)
291
292        if player_ids_to_remove:
293            current_members.difference_update(player_ids_to_remove)
294
295        # Always include self as first member if there are members
296        if current_members:
297            self._attr_group_members = [self.player_id] + [
298                pid for pid in current_members if pid != self.player_id
299            ]
300        else:
301            self._attr_group_members = []
302
303        # Clear cache to reflect changes
304        self._cache.clear()
305
306    async def stop(self) -> None:
307        """Stop playback - required abstract method."""
308
309
310class MockMass:
311    """Type hint for mocked MusicAssistant instance."""
312