/
/
/
1"""Fixtures for testing Music Assistant."""
2
3import asyncio
4import logging
5import pathlib
6from collections.abc import AsyncGenerator
7from unittest.mock import AsyncMock, MagicMock, NonCallableMagicMock, patch
8
9import pytest
10from zeroconf.asyncio import AsyncZeroconf
11
12from music_assistant.controllers.cache import CacheController
13from music_assistant.controllers.config import ConfigController
14from music_assistant.mass import MusicAssistant
15
16
17@pytest.fixture(name="caplog")
18def caplog_fixture(caplog: pytest.LogCaptureFixture) -> pytest.LogCaptureFixture:
19 """Set log level to debug for tests using the caplog fixture."""
20 caplog.set_level(logging.DEBUG)
21 return caplog
22
23
24def _create_mock_zeroconf() -> MagicMock:
25 """Create a mock AsyncZeroconf that prevents real network I/O.
26
27 Uses spec=AsyncZeroconf to ensure the mock only has valid attributes,
28 preventing it from being mistakenly registered as an API handler.
29 """
30 mock_zc = MagicMock(spec=AsyncZeroconf)
31 # Set up nested zeroconf object with proper spec
32 mock_inner_zc = NonCallableMagicMock()
33 mock_inner_zc.cache = NonCallableMagicMock()
34 mock_inner_zc.cache.cache = {} # Empty cache - no discovered services
35 mock_zc.zeroconf = mock_inner_zc
36 # Set up async methods
37 mock_zc.async_register_service = AsyncMock()
38 mock_zc.async_update_service = AsyncMock()
39 mock_zc.async_unregister_service = AsyncMock()
40 mock_zc.async_close = AsyncMock()
41 return mock_zc
42
43
44@pytest.fixture
45async def mass(tmp_path: pathlib.Path) -> AsyncGenerator[MusicAssistant, None]:
46 """Start a Music Assistant in test mode.
47
48 :param tmp_path: Temporary directory for test data.
49 """
50 storage_path = tmp_path / "data"
51 cache_path = tmp_path / "cache"
52 storage_path.mkdir(parents=True)
53 cache_path.mkdir(parents=True)
54
55 logging.getLogger("aiosqlite").level = logging.INFO
56
57 mass_instance = MusicAssistant(str(storage_path), str(cache_path))
58
59 # TODO: Configure a random port to avoid conflicts when MA is already running
60 # The conftest was modified in PR #2738 to add port configuration but it doesn't
61 # work correctly - the settings.json file is created but the config isn't respected.
62 # For now, tests that use the `mass` fixture will fail if MA is running on port 8095.
63
64 # Mock zeroconf to prevent real network I/O during tests
65 mock_zc = _create_mock_zeroconf()
66 mock_browser = NonCallableMagicMock() # Use NonCallable to avoid api_cmd issues
67
68 with (
69 patch("music_assistant.mass.AsyncZeroconf", return_value=mock_zc),
70 patch("music_assistant.mass.AsyncServiceBrowser", return_value=mock_browser),
71 ):
72 await mass_instance.start()
73
74 try:
75 yield mass_instance
76 finally:
77 await mass_instance.stop()
78
79
80@pytest.fixture
81async def mass_minimal(tmp_path: pathlib.Path) -> AsyncGenerator[MusicAssistant, None]:
82 """Create a minimal Music Assistant instance without starting the full server.
83
84 Only initializes the event loop and config controller.
85 Useful for testing individual controllers without the overhead of the webserver.
86
87 :param tmp_path: Temporary directory for test data.
88 """
89 storage_path = tmp_path / "data"
90 cache_path = tmp_path / "cache"
91 storage_path.mkdir(parents=True)
92 cache_path.mkdir(parents=True)
93
94 logging.getLogger("aiosqlite").level = logging.INFO
95
96 mass_instance = MusicAssistant(str(storage_path), str(cache_path))
97
98 mass_instance.loop = asyncio.get_running_loop()
99 mass_instance.loop_thread_id = (
100 getattr(mass_instance.loop, "_thread_id", None)
101 if hasattr(mass_instance.loop, "_thread_id")
102 else id(mass_instance.loop)
103 )
104
105 mass_instance.config = ConfigController(mass_instance)
106 await mass_instance.config.setup()
107
108 mass_instance.cache = CacheController(mass_instance)
109
110 try:
111 yield mass_instance
112 finally:
113 if mass_instance.cache.database:
114 await mass_instance.cache.database.close()
115 await mass_instance.config.close()
116