music-assistant-server

1"""Fixtures for testing Music Assistant."""
2
3import asyncio
4import logging
5import os
6import pathlib
7import tempfile
8import threading
9from collections.abc import AsyncGenerator, Generator
10from contextlib import asynccontextmanager
11from unittest.mock import AsyncMock, MagicMock, NonCallableMagicMock, patch
12
13import pytest
14from music_assistant_models import helpers as models_helpers
15from zeroconf.asyncio import AsyncZeroconf
16
17from music_assistant.controllers.cache import CacheController
18from music_assistant.controllers.config import ConfigController
19from music_assistant.controllers.discovery import DiscoveryController
20from music_assistant.controllers.music import MusicController
21from music_assistant.controllers.tasks import TasksController
22from music_assistant.mass import MusicAssistant
23from tests.common import (
24    suppress_auto_loaded_providers,
25    suppress_initial_library_sync,
26    use_ephemeral_server_ports,
27    utf8_safe,
28    wait_for_boot_to_settle,
29)
30
31NUMBA_CACHE_DIR = pytest.StashKey[tempfile.TemporaryDirectory[str]]()
32
33
34def pytest_configure(config: pytest.Config) -> None:
35    """
36    Give this test process its own numba kernel cache.
37
38    librosa compiles its numba kernels with ``cache=True`` into a directory shared by
39    every process on the machine. numba updates that cache's index non-atomically, so
40    xdist workers filling a cold cache can leave an entry pointing at another
41    signature's machine code — calling it then segfaults the worker.
42    See https://github.com/numba/numba/issues/10128.
43    """
44    cache_dir = tempfile.TemporaryDirectory(prefix="ma-numba-cache-")
45    config.stash[NUMBA_CACHE_DIR] = cache_dir
46    # numba reads this once, when it is imported; nothing here imports it that early.
47    os.environ["NUMBA_CACHE_DIR"] = cache_dir.name
48
49
50def pytest_unconfigure(config: pytest.Config) -> None:
51    """Drop this test process's numba kernel cache."""
52    if (cache_dir := config.stash.get(NUMBA_CACHE_DIR, None)) is not None:
53        cache_dir.cleanup()
54
55
56@pytest.hookimpl(wrapper=True)
57def pytest_report_to_serializable() -> Generator[None, object, object]:
58    """
59    Make serialized test reports strict-UTF-8 safe for pytest-xdist.
60
61    Lone surrogates in a captured report (e.g. undecodable filesystem paths) kill
62    the execnet worker channel. Covers test reports only; other xdist payloads
63    (warnings, log-start nodeids) are serialized outside this hook.
64    """
65    data = yield
66    return utf8_safe(data)
67
68
69@pytest.fixture(autouse=True)
70def isolate_models_global_cache() -> Generator[None]:
71    """
72    Reset the models package's process-global cache between tests.
73
74    A full server boot populates module-level globals in music_assistant_models
75    (e.g. ``available_providers``, which drives ``MediaItem.available``). Left in
76    place, they leak into later tests in the same pytest process and change item
77    availability depending on test ordering. Note that an empty cache falls back
78    to permissive defaults, so tests sharing a broader-scoped server instance are
79    unaffected by the per-test clear.
80    """
81    yield
82    models_helpers._global_cache.clear()
83
84
85@pytest.fixture(name="caplog")
86def caplog_fixture(caplog: pytest.LogCaptureFixture) -> pytest.LogCaptureFixture:
87    """Set log level to debug for tests using the caplog fixture."""
88    caplog.set_level(logging.DEBUG)
89    return caplog
90
91
92def _create_mock_zeroconf() -> MagicMock:
93    """
94    Create a mock AsyncZeroconf that prevents real network I/O.
95
96    Uses spec=AsyncZeroconf to ensure the mock only has valid attributes,
97    preventing it from being mistakenly registered as an API handler.
98    """
99    mock_zc = MagicMock(spec=AsyncZeroconf)
100    # Set up nested zeroconf object with proper spec
101    mock_inner_zc = NonCallableMagicMock()
102    mock_inner_zc.cache = NonCallableMagicMock()
103    mock_inner_zc.cache.cache = {}  # Empty cache - no discovered services
104    mock_zc.zeroconf = mock_inner_zc
105    # Set up async methods
106    mock_zc.async_register_service = AsyncMock()
107    mock_zc.async_update_service = AsyncMock()
108    mock_zc.async_unregister_service = AsyncMock()
109    mock_zc.async_close = AsyncMock()
110    return mock_zc
111
112
113@pytest.fixture
114async def mass(tmp_path: pathlib.Path) -> AsyncGenerator[MusicAssistant]:
115    """
116    Start a Music Assistant in test mode.
117
118    :param tmp_path: Temporary directory for test data.
119    """
120    storage_path = tmp_path / "data"
121    cache_path = tmp_path / "cache"
122    storage_path.mkdir(parents=True)
123    cache_path.mkdir(parents=True)
124
125    logging.getLogger("aiosqlite").level = logging.INFO
126
127    mass_instance = MusicAssistant(str(storage_path), str(cache_path))
128
129    # Mock zeroconf to prevent real network I/O during tests
130    mock_zc = _create_mock_zeroconf()
131    mock_browser = NonCallableMagicMock()  # Use NonCallable to avoid api_cmd issues
132
133    with (
134        use_ephemeral_server_ports(),
135        patch(
136            "music_assistant.controllers.discovery.controller.AsyncZeroconf",
137            return_value=mock_zc,
138        ),
139        patch(
140            "music_assistant.controllers.discovery.controller.AsyncServiceBrowser",
141            return_value=mock_browser,
142        ),
143        # Booting the server runs an ffmpeg presence check; mock it so tests can boot
144        # without the binary. Tests that actually spawn ffmpeg still use the real one.
145        patch(
146            "music_assistant.controllers.streams.controller.check_ffmpeg_version",
147            new=AsyncMock(),
148        ),
149        # keep the fixture isolated from the developer's machine: no auto-loaded
150        # device providers
151        suppress_auto_loaded_providers(),
152        # keep the booted instance quiet: no library sync firing into a running test
153        suppress_initial_library_sync(),
154    ):
155        try:
156            await mass_instance.start()
157            await wait_for_boot_to_settle(mass_instance)
158            yield mass_instance
159        finally:
160            # also stop after a failed boot: pytest holds on to the setup traceback,
161            # which keeps the half-started server (and the non-daemon threads of its
162            # open database connections) alive until the interpreter exits, where
163            # joining those threads then hangs the whole test process
164            await mass_instance.stop()
165
166
167@pytest.fixture
168async def mass_minimal(tmp_path: pathlib.Path) -> AsyncGenerator[MusicAssistant]:
169    """
170    Create a minimal Music Assistant instance without starting the full server.
171
172    Only initializes the event loop and config controller.
173    Useful for testing individual controllers without the overhead of the webserver.
174
175    :param tmp_path: Temporary directory for test data.
176    """
177    async with _minimal_mass_context(tmp_path) as mass_instance:
178        yield mass_instance
179
180
181@pytest.fixture(scope="class")
182async def music_mass_class(
183    tmp_path_factory: pytest.TempPathFactory,
184) -> AsyncGenerator[MusicAssistant]:
185    """Create a class-scoped Music Assistant instance with only library storage."""
186    async with _music_mass_context(tmp_path_factory.mktemp("music_class")) as mass_instance:
187        yield mass_instance
188
189
190@pytest.fixture(scope="module")
191async def music_mass_module(
192    tmp_path_factory: pytest.TempPathFactory,
193) -> AsyncGenerator[MusicAssistant]:
194    """Create a module-scoped Music Assistant instance with only library storage."""
195    async with _music_mass_context(tmp_path_factory.mktemp("music_module")) as mass_instance:
196        yield mass_instance
197
198
199@asynccontextmanager
200async def _minimal_mass_context(
201    tmp_path: pathlib.Path,
202) -> AsyncGenerator[MusicAssistant]:
203    """Create a minimal Music Assistant instance for a fixture."""
204    storage_path = tmp_path / "data"
205    cache_path = tmp_path / "cache"
206    storage_path.mkdir(parents=True)
207    cache_path.mkdir(parents=True)
208
209    logging.getLogger("aiosqlite").level = logging.INFO
210
211    mass_instance = MusicAssistant(str(storage_path), str(cache_path))
212    mass_instance.loop = asyncio.get_running_loop()
213    mass_instance.loop_thread_id = threading.get_ident()
214    mass_instance.config = ConfigController(mass_instance)
215    await mass_instance.config.setup()
216    mass_instance.discovery = DiscoveryController(mass_instance)
217    mass_instance.cache = CacheController(mass_instance)
218
219    try:
220        yield mass_instance
221    finally:
222        await mass_instance.cache.close()
223        await mass_instance.config.close()
224
225
226@asynccontextmanager
227async def _music_mass_context(
228    tmp_path: pathlib.Path,
229) -> AsyncGenerator[MusicAssistant]:
230    """Create a minimal Music Assistant instance with a real library database."""
231    async with _minimal_mass_context(tmp_path) as mass_instance:
232        mass_instance.tasks = TasksController(mass_instance)
233        tasks_config = await mass_instance.config.get_core_config(mass_instance.tasks.domain)
234        await mass_instance.tasks.setup(tasks_config)
235
236        mass_instance.metadata = MagicMock()
237        mass_instance.metadata.schedule_update_metadata = MagicMock()
238        mass_instance.metadata.invalidate_image_cache = AsyncMock()
239        mass_instance.webserver = MagicMock()
240        mass_instance.webserver.auth.list_users = AsyncMock(return_value=[])
241
242        mass_instance.music = MusicController(mass_instance)
243        music_config = await mass_instance.config.get_core_config(mass_instance.music.domain)
244        await mass_instance.music.setup(music_config)
245        await mass_instance.music.post_setup()
246        try:
247            yield mass_instance
248        finally:
249            await mass_instance.tasks.close()
250            await mass_instance.music.close()
251