/
/
/
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 async with full_mass_context(tmp_path) as mass_instance:
121 yield mass_instance
122
123
124@asynccontextmanager
125async def full_mass_context(tmp_path: pathlib.Path) -> AsyncGenerator[MusicAssistant]:
126 """
127 Boot a full server on the given temporary directory.
128
129 Exposed next to the ``mass`` fixture so a test that needs to seed ``data/settings.json``
130 (or the cache) before the boot can prepare the directory itself and boot it here.
131
132 :param tmp_path: Temporary directory for test data.
133 """
134 storage_path = tmp_path / "data"
135 cache_path = tmp_path / "cache"
136 storage_path.mkdir(parents=True, exist_ok=True)
137 cache_path.mkdir(parents=True, exist_ok=True)
138
139 logging.getLogger("aiosqlite").level = logging.INFO
140
141 mass_instance = MusicAssistant(str(storage_path), str(cache_path))
142
143 # Mock zeroconf to prevent real network I/O during tests
144 mock_zc = _create_mock_zeroconf()
145 mock_browser = NonCallableMagicMock() # Use NonCallable to avoid api_cmd issues
146
147 with (
148 use_ephemeral_server_ports(),
149 patch(
150 "music_assistant.controllers.discovery.controller.AsyncZeroconf",
151 return_value=mock_zc,
152 ),
153 patch(
154 "music_assistant.controllers.discovery.controller.AsyncServiceBrowser",
155 return_value=mock_browser,
156 ),
157 # Booting the server runs an ffmpeg presence check; mock it so tests can boot
158 # without the binary. Tests that actually spawn ffmpeg still use the real one.
159 patch(
160 "music_assistant.controllers.streams.controller.check_ffmpeg_version",
161 new=AsyncMock(),
162 ),
163 # keep the fixture isolated from the developer's machine: no auto-loaded
164 # device providers
165 suppress_auto_loaded_providers(),
166 # keep the booted instance quiet: no library sync firing into a running test
167 suppress_initial_library_sync(),
168 ):
169 try:
170 await mass_instance.start()
171 await wait_for_boot_to_settle(mass_instance)
172 yield mass_instance
173 finally:
174 # also stop after a failed boot: pytest holds on to the setup traceback,
175 # which keeps the half-started server (and the non-daemon threads of its
176 # open database connections) alive until the interpreter exits, where
177 # joining those threads then hangs the whole test process
178 await mass_instance.stop()
179
180
181@pytest.fixture
182async def mass_minimal(tmp_path: pathlib.Path) -> AsyncGenerator[MusicAssistant]:
183 """
184 Create a minimal Music Assistant instance without starting the full server.
185
186 Only initializes the event loop and config controller.
187 Useful for testing individual controllers without the overhead of the webserver.
188
189 :param tmp_path: Temporary directory for test data.
190 """
191 async with _minimal_mass_context(tmp_path) as mass_instance:
192 yield mass_instance
193
194
195@pytest.fixture(scope="class")
196async def music_mass_class(
197 tmp_path_factory: pytest.TempPathFactory,
198) -> AsyncGenerator[MusicAssistant]:
199 """Create a class-scoped Music Assistant instance with only library storage."""
200 async with _music_mass_context(tmp_path_factory.mktemp("music_class")) as mass_instance:
201 yield mass_instance
202
203
204@pytest.fixture(scope="module")
205async def music_mass_module(
206 tmp_path_factory: pytest.TempPathFactory,
207) -> AsyncGenerator[MusicAssistant]:
208 """Create a module-scoped Music Assistant instance with only library storage."""
209 async with _music_mass_context(tmp_path_factory.mktemp("music_module")) as mass_instance:
210 yield mass_instance
211
212
213@asynccontextmanager
214async def _minimal_mass_context(
215 tmp_path: pathlib.Path,
216) -> AsyncGenerator[MusicAssistant]:
217 """Create a minimal Music Assistant instance for a fixture."""
218 storage_path = tmp_path / "data"
219 cache_path = tmp_path / "cache"
220 storage_path.mkdir(parents=True)
221 cache_path.mkdir(parents=True)
222
223 logging.getLogger("aiosqlite").level = logging.INFO
224
225 mass_instance = MusicAssistant(str(storage_path), str(cache_path))
226 mass_instance.loop = asyncio.get_running_loop()
227 mass_instance.loop_thread_id = threading.get_ident()
228 mass_instance.config = ConfigController(mass_instance)
229 await mass_instance.config.setup()
230 mass_instance.discovery = DiscoveryController(mass_instance)
231 mass_instance.cache = CacheController(mass_instance)
232
233 try:
234 yield mass_instance
235 finally:
236 await mass_instance.cache.close()
237 await mass_instance.config.close()
238
239
240@asynccontextmanager
241async def _music_mass_context(
242 tmp_path: pathlib.Path,
243) -> AsyncGenerator[MusicAssistant]:
244 """Create a minimal Music Assistant instance with a real library database."""
245 async with _minimal_mass_context(tmp_path) as mass_instance:
246 mass_instance.tasks = TasksController(mass_instance)
247 tasks_config = await mass_instance.config.get_core_config(mass_instance.tasks.domain)
248 await mass_instance.tasks.setup(tasks_config)
249
250 mass_instance.metadata = MagicMock()
251 mass_instance.metadata.schedule_update_metadata = MagicMock()
252 mass_instance.metadata.invalidate_image_cache = AsyncMock()
253 mass_instance.webserver = MagicMock()
254 mass_instance.webserver.auth.list_users = AsyncMock(return_value=[])
255
256 mass_instance.music = MusicController(mass_instance)
257 music_config = await mass_instance.config.get_core_config(mass_instance.music.domain)
258 await mass_instance.music.setup(music_config)
259 await mass_instance.music.post_setup()
260 try:
261 yield mass_instance
262 finally:
263 await mass_instance.tasks.close()
264 await mass_instance.music.close()
265