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