/
/
/
1"""Tests for the discovery core controller."""
2
3import logging
4from typing import Any
5from unittest.mock import AsyncMock, MagicMock, patch
6
7import pytest
8from zeroconf import IPVersion
9from zeroconf.asyncio import AsyncZeroconf
10
11from music_assistant.constants import VERBOSE_LOG_LEVEL
12from music_assistant.mass import MusicAssistant
13
14
15class StubUpnpProvider:
16 """Minimal provider stub for SSDP discovery tests."""
17
18 def __init__(self) -> None:
19 """Initialize the provider stub."""
20 self.instance_id = "stub_upnp"
21 self.name = "Stub UPNP"
22 self.available = True
23 self.logger = MagicMock()
24 self.manifest = MagicMock(mdns_discovery=None, upnp_discovery=["roku:ecp"])
25 self.on_mdns_service_state_change = AsyncMock()
26 self.on_upnp_service_discovered = AsyncMock()
27
28
29@pytest.mark.parametrize(
30 ("controller_level", "library_level"),
31 [
32 (VERBOSE_LOG_LEVEL, logging.DEBUG),
33 (logging.DEBUG, logging.INFO),
34 (logging.INFO, logging.WARNING),
35 ],
36)
37def test_discovery_library_logger_levels(
38 mass_minimal: MusicAssistant, controller_level: int, library_level: int
39) -> None:
40 """Discovery library loggers should only expose debug output at VERBOSE."""
41 controller_logger = mass_minimal.discovery.logger
42 library_loggers = [
43 logging.getLogger("async_upnp_client"),
44 logging.getLogger("zeroconf"),
45 ]
46 previous_controller_level = controller_logger.level
47 previous_library_levels = [logger.level for logger in library_loggers]
48 try:
49 controller_logger.setLevel(controller_level)
50 mass_minimal.discovery._configure_library_loggers()
51
52 assert all(logger.level == library_level for logger in library_loggers)
53 finally:
54 controller_logger.setLevel(previous_controller_level)
55 for logger, previous_level in zip(library_loggers, previous_library_levels, strict=True):
56 logger.setLevel(previous_level)
57
58
59async def test_run_provider_discovery_dispatches_upnp_callbacks(mass: MusicAssistant) -> None:
60 """Provider-targeted discovery should fan SSDP results back into the provider callback."""
61 provider = StubUpnpProvider()
62 mass._providers[provider.instance_id] = provider # type: ignore[assignment]
63
64 async def fake_async_search(callback: Any, search_target: str, target: Any = None) -> None:
65 del target
66 await callback(
67 {
68 "st": search_target,
69 "usn": "uuid:roku-123::roku:ecp",
70 "_host": "192.168.1.25",
71 }
72 )
73
74 with patch(
75 "music_assistant.controllers.discovery.controller.async_upnp_search",
76 new=AsyncMock(side_effect=fake_async_search),
77 ) as mock_async_search:
78 await mass.discovery.run_provider_discovery(provider) # type: ignore[arg-type]
79
80 assert mock_async_search.await_count == 1
81 assert mock_async_search.await_args is not None
82 assert mock_async_search.await_args.kwargs["search_target"] == "roku:ecp"
83 provider.on_upnp_service_discovered.assert_awaited_once()
84 mass._providers.pop(provider.instance_id, None)
85 mass.discovery.on_provider_unload(provider.instance_id)
86
87
88async def test_discovery_controller_owns_async_zeroconf(mass_minimal: MusicAssistant) -> None:
89 """The shared AsyncZeroconf instance should be created by the discovery controller."""
90 mass_minimal.config.set(
91 "core/discovery",
92 {"domain": "discovery", "values": {"zeroconf_interfaces": "all"}},
93 )
94 mass_minimal.webserver = MagicMock(
95 publish_ip="127.0.0.1",
96 publish_addresses=["127.0.0.1"],
97 publish_port=8095,
98 base_url="http://127.0.0.1:8095",
99 )
100 mock_zc = MagicMock(spec=AsyncZeroconf)
101 mock_zc.async_register_service = AsyncMock()
102 mock_zc.async_update_service = AsyncMock()
103 mock_zc.async_unregister_service = AsyncMock()
104 mock_zc.async_close = AsyncMock()
105
106 # Stub the resolved zeroconf args so the assertion below does not depend on
107 # the host's network adapters or platform (both are covered by the util tests)
108 zc_args = {"ip_version": IPVersion.All, "interfaces": ["192.168.1.10", "fd00::1%2"]}
109
110 with (
111 patch(
112 "music_assistant.controllers.discovery.controller.AsyncZeroconf",
113 return_value=mock_zc,
114 ) as mock_async_zeroconf,
115 patch(
116 "music_assistant.controllers.discovery.controller.get_ip_pton",
117 new=AsyncMock(return_value=b"\x7f\x00\x00\x01"),
118 ),
119 patch(
120 "music_assistant.controllers.discovery.controller.get_zeroconf_args",
121 autospec=True,
122 return_value=zc_args,
123 ) as mock_get_zeroconf_args,
124 ):
125 await mass_minimal.discovery.setup(await mass_minimal.config.get_core_config("discovery"))
126 assert mass_minimal.discovery.aiozc is mock_zc
127
128 await mass_minimal.discovery.close()
129
130 mock_get_zeroconf_args.assert_called_once_with(True)
131 mock_async_zeroconf.assert_called_once_with(
132 ip_version=zc_args["ip_version"],
133 interfaces=zc_args["interfaces"],
134 )
135
136
137async def test_mass_service_advertises_webserver_publish_addresses(
138 mass_minimal: MusicAssistant,
139) -> None:
140 """The advertisement must contain exactly the webserver's publish addresses."""
141 mass_minimal.webserver = MagicMock(
142 publish_ip="192.168.1.10",
143 publish_addresses=["192.168.1.10", "fd00::10"],
144 publish_port=8095,
145 base_url="http://192.168.1.10:8095",
146 )
147 mock_zc = MagicMock(spec=AsyncZeroconf)
148 mock_zc.async_register_service = AsyncMock()
149 mock_zc.async_update_service = AsyncMock()
150 mass_minimal.discovery._aiozc = mock_zc
151
152 with (
153 patch(
154 "music_assistant.controllers.discovery.controller.AsyncServiceInfo"
155 ) as mock_service_info,
156 patch(
157 "music_assistant.controllers.discovery.controller.get_ip_pton",
158 new=AsyncMock(side_effect=lambda ip: ip.encode()),
159 ),
160 ):
161 await mass_minimal.discovery._register_mass_service()
162
163 assert mock_service_info.call_args.kwargs["addresses"] == [b"192.168.1.10", b"fd00::10"]
164
165
166async def test_mass_service_refreshes_on_core_state_update(
167 mass_minimal: MusicAssistant,
168) -> None:
169 """A server info change must refresh the already registered mdns record."""
170 mass_minimal.webserver = MagicMock(
171 publish_ip="192.168.1.10",
172 publish_addresses=["192.168.1.10"],
173 publish_port=8095,
174 base_url="http://192.168.1.10:8095",
175 )
176 mock_zc = MagicMock(spec=AsyncZeroconf)
177 mock_zc.async_register_service = AsyncMock()
178 mock_zc.async_update_service = AsyncMock()
179 mass_minimal.discovery._aiozc = mock_zc
180 mass_minimal.discovery._mass_service_info = MagicMock()
181
182 with (
183 patch("music_assistant.controllers.discovery.controller.AsyncServiceInfo"),
184 patch(
185 "music_assistant.controllers.discovery.controller.get_ip_pton",
186 new=AsyncMock(side_effect=lambda ip: ip.encode()),
187 ),
188 ):
189 await mass_minimal.discovery._on_core_state_updated(MagicMock())
190
191 mock_zc.async_update_service.assert_awaited_once()
192 mock_zc.async_register_service.assert_not_awaited()
193
194
195async def test_async_find_mdns_service_matches_exact_device_name(mass: MusicAssistant) -> None:
196 """A device name must not cross-match another device whose name contains it."""
197 mass.discovery.aiozc.zeroconf.cache.cache = {
198 "aabbccddeeff@kelder._raop._tcp.local.": {},
199 "atv kelder._raop._tcp.local.": {},
200 }
201 mock_info = MagicMock()
202 mock_info.async_request = AsyncMock(return_value=True)
203 with patch(
204 "music_assistant.controllers.discovery.controller.AsyncServiceInfo",
205 return_value=mock_info,
206 ) as mock_service_info:
207 result = await mass.discovery.async_find_mdns_service(
208 "_raop._tcp.local.", "Kelder", timeout=1.0
209 )
210
211 assert result is mock_info
212 # "Kelder" must resolve to its own RAOP entry, never the "ATV Kelder" one.
213 mock_service_info.assert_called_once_with(
214 "_raop._tcp.local.", "aabbccddeeff@kelder._raop._tcp.local."
215 )
216
217
218async def test_async_find_mdns_service_no_substring_match(mass: MusicAssistant) -> None:
219 """Looking up a name that is only a substring of a discovered name should not match."""
220 mass.discovery.aiozc.zeroconf.cache.cache = {
221 "atv kelder._raop._tcp.local.": {},
222 }
223 with patch(
224 "music_assistant.controllers.discovery.controller.AsyncServiceInfo",
225 ) as mock_service_info:
226 result = await mass.discovery.async_find_mdns_service(
227 "_raop._tcp.local.", "Kelder", timeout=0.1
228 )
229
230 assert result is None
231 mock_service_info.assert_not_called()
232
233
234async def test_async_find_mdns_service_preserves_at_sign_in_name(mass: MusicAssistant) -> None:
235 """Only a RAOP MAC prefix is stripped, so device names containing '@' still match."""
236 mass.discovery.aiozc.zeroconf.cache.cache = {
237 "living@home._airplay._tcp.local.": {},
238 }
239 mock_info = MagicMock()
240 mock_info.async_request = AsyncMock(return_value=True)
241 with patch(
242 "music_assistant.controllers.discovery.controller.AsyncServiceInfo",
243 return_value=mock_info,
244 ) as mock_service_info:
245 result = await mass.discovery.async_find_mdns_service(
246 "_airplay._tcp.local.", "Living@Home", timeout=1.0
247 )
248
249 assert result is mock_info
250 mock_service_info.assert_called_once_with(
251 "_airplay._tcp.local.", "living@home._airplay._tcp.local."
252 )
253