/
/
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_async_find_mdns_service_matches_exact_device_name(mass: MusicAssistant) -> None:
167 """A device name must not cross-match another device whose name contains it."""
168 mass.discovery.aiozc.zeroconf.cache.cache = {
169 "aabbccddeeff@kelder._raop._tcp.local.": {},
170 "atv kelder._raop._tcp.local.": {},
171 }
172 mock_info = MagicMock()
173 mock_info.async_request = AsyncMock(return_value=True)
174 with patch(
175 "music_assistant.controllers.discovery.controller.AsyncServiceInfo",
176 return_value=mock_info,
177 ) as mock_service_info:
178 result = await mass.discovery.async_find_mdns_service(
179 "_raop._tcp.local.", "Kelder", timeout=1.0
180 )
181
182 assert result is mock_info
183 # "Kelder" must resolve to its own RAOP entry, never the "ATV Kelder" one.
184 mock_service_info.assert_called_once_with(
185 "_raop._tcp.local.", "aabbccddeeff@kelder._raop._tcp.local."
186 )
187
188
189async def test_async_find_mdns_service_no_substring_match(mass: MusicAssistant) -> None:
190 """Looking up a name that is only a substring of a discovered name should not match."""
191 mass.discovery.aiozc.zeroconf.cache.cache = {
192 "atv kelder._raop._tcp.local.": {},
193 }
194 with patch(
195 "music_assistant.controllers.discovery.controller.AsyncServiceInfo",
196 ) as mock_service_info:
197 result = await mass.discovery.async_find_mdns_service(
198 "_raop._tcp.local.", "Kelder", timeout=0.1
199 )
200
201 assert result is None
202 mock_service_info.assert_not_called()
203
204
205async def test_async_find_mdns_service_preserves_at_sign_in_name(mass: MusicAssistant) -> None:
206 """Only a RAOP MAC prefix is stripped, so device names containing '@' still match."""
207 mass.discovery.aiozc.zeroconf.cache.cache = {
208 "living@home._airplay._tcp.local.": {},
209 }
210 mock_info = MagicMock()
211 mock_info.async_request = AsyncMock(return_value=True)
212 with patch(
213 "music_assistant.controllers.discovery.controller.AsyncServiceInfo",
214 return_value=mock_info,
215 ) as mock_service_info:
216 result = await mass.discovery.async_find_mdns_service(
217 "_airplay._tcp.local.", "Living@Home", timeout=1.0
218 )
219
220 assert result is mock_info
221 mock_service_info.assert_called_once_with(
222 "_airplay._tcp.local.", "living@home._airplay._tcp.local."
223 )
224