/
/
/
1"""Fixtures for testing Music Assistant."""
2
3import asyncio
4import logging
5import pathlib
6from collections.abc import AsyncGenerator
7
8import pytest
9
10from music_assistant.controllers.cache import CacheController
11from music_assistant.controllers.config import ConfigController
12from music_assistant.mass import MusicAssistant
13
14
15@pytest.fixture(name="caplog")
16def caplog_fixture(caplog: pytest.LogCaptureFixture) -> pytest.LogCaptureFixture:
17 """Set log level to debug for tests using the caplog fixture."""
18 caplog.set_level(logging.DEBUG)
19 return caplog
20
21
22@pytest.fixture
23async def mass(tmp_path: pathlib.Path) -> AsyncGenerator[MusicAssistant, None]:
24 """Start a Music Assistant in test mode.
25
26 :param tmp_path: Temporary directory for test data.
27 """
28 storage_path = tmp_path / "data"
29 cache_path = tmp_path / "cache"
30 storage_path.mkdir(parents=True)
31 cache_path.mkdir(parents=True)
32
33 logging.getLogger("aiosqlite").level = logging.INFO
34
35 mass_instance = MusicAssistant(str(storage_path), str(cache_path))
36
37 # TODO: Configure a random port to avoid conflicts when MA is already running
38 # The conftest was modified in PR #2738 to add port configuration but it doesn't
39 # work correctly - the settings.json file is created but the config isn't respected.
40 # For now, tests that use the `mass` fixture will fail if MA is running on port 8095.
41
42 await mass_instance.start()
43
44 try:
45 yield mass_instance
46 finally:
47 await mass_instance.stop()
48
49
50@pytest.fixture
51async def mass_minimal(tmp_path: pathlib.Path) -> AsyncGenerator[MusicAssistant, None]:
52 """Create a minimal Music Assistant instance without starting the full server.
53
54 Only initializes the event loop and config controller.
55 Useful for testing individual controllers without the overhead of the webserver.
56
57 :param tmp_path: Temporary directory for test data.
58 """
59 storage_path = tmp_path / "data"
60 cache_path = tmp_path / "cache"
61 storage_path.mkdir(parents=True)
62 cache_path.mkdir(parents=True)
63
64 logging.getLogger("aiosqlite").level = logging.INFO
65
66 mass_instance = MusicAssistant(str(storage_path), str(cache_path))
67
68 mass_instance.loop = asyncio.get_running_loop()
69 mass_instance.loop_thread_id = (
70 getattr(mass_instance.loop, "_thread_id", None)
71 if hasattr(mass_instance.loop, "_thread_id")
72 else id(mass_instance.loop)
73 )
74
75 mass_instance.config = ConfigController(mass_instance)
76 await mass_instance.config.setup()
77
78 mass_instance.cache = CacheController(mass_instance)
79
80 try:
81 yield mass_instance
82 finally:
83 if mass_instance.cache.database:
84 await mass_instance.cache.database.close()
85 await mass_instance.config.close()
86