/
/
/
1"""Tests for Sonos provider discovery."""
2
3import asyncio
4import logging
5from unittest.mock import MagicMock
6
7import pytest
8from zeroconf import ServiceStateChange
9
10from music_assistant.mass import MusicAssistant
11from music_assistant.providers.sonos.provider import SonosPlayerProvider
12
13PLAYER_ID = "sonos_player"
14SERVICE_NAME = "Toilet._sonos._tcp.local."
15
16
17def _bind_provider(mass: MusicAssistant | MagicMock) -> SonosPlayerProvider:
18 """Create a Sonos provider bound to the given MusicAssistant."""
19 provider = SonosPlayerProvider.__new__(SonosPlayerProvider)
20 provider.mass = mass
21 provider.logger = logging.getLogger("test.sonos.discovery")
22 provider._ignored_disabled_players = set()
23 provider._pending_setup_tasks = set()
24 provider._unloaded = False
25 return provider
26
27
28def _make_provider(enabled: bool = False) -> tuple[SonosPlayerProvider, MagicMock]:
29 """Create a Sonos provider with mocked discovery dependencies."""
30 mass = MagicMock()
31 mass.config.get_raw_player_config_value.return_value = enabled
32 mass.players.get_player.return_value = None
33 return _bind_provider(mass), mass
34
35
36def _bind_discovering_provider(mass: MusicAssistant) -> SonosPlayerProvider:
37 """Create a Sonos provider that discovers players on the real timer machinery."""
38 mass.config = MagicMock()
39 mass.config.get_raw_player_config_value.return_value = True
40 mass.players = MagicMock()
41 mass.players.get_player.return_value = None
42 mass.streams = MagicMock()
43 return _bind_provider(mass)
44
45
46def _make_discovery_info(player_id: str = PLAYER_ID) -> MagicMock:
47 """Create minimal Sonos mDNS discovery information."""
48 info = MagicMock()
49 info.decoded_properties = {"uuid": player_id}
50 return info
51
52
53@pytest.mark.asyncio
54async def test_disabled_discovery_is_not_scheduled_repeatedly(
55 caplog: pytest.LogCaptureFixture,
56) -> None:
57 """Test repeated announcements for a disabled player are ignored once."""
58 provider, mass = _make_provider()
59 info = _make_discovery_info()
60
61 with caplog.at_level(logging.DEBUG, logger=provider.logger.name):
62 await provider.on_mdns_service_state_change(SERVICE_NAME, ServiceStateChange.Added, info)
63 await provider.on_mdns_service_state_change(SERVICE_NAME, ServiceStateChange.Updated, info)
64
65 mass.call_later.assert_not_called()
66 ignored_records = [
67 record for record in caplog.records if "in discovery as it is disabled" in record.message
68 ]
69 assert len(ignored_records) == 1
70
71
72@pytest.mark.asyncio
73async def test_disabled_discovery_logs_again_after_reenable(
74 caplog: pytest.LogCaptureFixture,
75) -> None:
76 """Test a new disabled period produces a new discovery diagnostic."""
77 provider, mass = _make_provider()
78 info = _make_discovery_info()
79
80 with caplog.at_level(logging.DEBUG, logger=provider.logger.name):
81 await provider.on_mdns_service_state_change(SERVICE_NAME, ServiceStateChange.Added, info)
82 mass.config.get_raw_player_config_value.return_value = True
83 await provider.on_mdns_service_state_change(SERVICE_NAME, ServiceStateChange.Updated, info)
84 mass.config.get_raw_player_config_value.return_value = False
85 await provider.on_mdns_service_state_change(SERVICE_NAME, ServiceStateChange.Updated, info)
86
87 mass.call_later.assert_called_once()
88 ignored_records = [
89 record for record in caplog.records if "in discovery as it is disabled" in record.message
90 ]
91 assert len(ignored_records) == 2
92
93
94@pytest.mark.asyncio
95async def test_unload_cancels_a_pending_player_setup(timer_mass: MusicAssistant) -> None:
96 """Test a discovered player that is still waiting to be set up is dropped on unload."""
97 provider = _bind_discovering_provider(timer_mass)
98
99 await provider.on_mdns_service_state_change(
100 SERVICE_NAME, ServiceStateChange.Added, _make_discovery_info()
101 )
102 handle = timer_mass._tracked_timers[f"setup_sonos_{PLAYER_ID}"]
103
104 await provider.unload(is_removed=True)
105
106 assert handle.cancelled()
107 assert provider._pending_setup_tasks == set()
108
109
110@pytest.mark.asyncio
111async def test_unload_cancels_a_player_setup_that_already_started(
112 timer_mass: MusicAssistant,
113) -> None:
114 """Test a player setup that is already running is aborted when the provider unloads."""
115 provider = _bind_discovering_provider(timer_mass)
116 setup_started = asyncio.Event()
117
118 async def _setup_player(*_args: object) -> None:
119 setup_started.set()
120 await asyncio.sleep(5)
121
122 provider._setup_player = _setup_player # type: ignore[assignment]
123 await provider.on_mdns_service_state_change(
124 SERVICE_NAME, ServiceStateChange.Added, _make_discovery_info()
125 )
126 task_id = f"setup_sonos_{PLAYER_ID}"
127 # let the armed timer fire now instead of waiting out the discovery debounce
128 timer_mass.call_later(0, provider._setup_player, task_id=task_id)
129 await setup_started.wait()
130 task = timer_mass._tracked_tasks[task_id]
131
132 await provider.unload(is_removed=True)
133
134 with pytest.raises(asyncio.CancelledError):
135 await task
136
137
138@pytest.mark.asyncio
139async def test_an_announcement_arriving_after_the_unload_is_ignored(
140 timer_mass: MusicAssistant,
141) -> None:
142 """Test a discovery callback that lands after the unload does not arm a new setup."""
143 provider = _bind_discovering_provider(timer_mass)
144
145 await provider.unload(is_removed=True)
146 await provider.on_mdns_service_state_change(
147 SERVICE_NAME, ServiceStateChange.Added, _make_discovery_info()
148 )
149
150 assert timer_mass._tracked_timers == {}
151 assert provider._pending_setup_tasks == set()
152
153
154@pytest.mark.asyncio
155async def test_unload_sweeps_every_discovered_player(timer_mass: MusicAssistant) -> None:
156 """Test a burst of discovered players leaves no setup behind after unload."""
157 provider = _bind_discovering_provider(timer_mass)
158
159 for index in range(3):
160 await provider.on_mdns_service_state_change(
161 f"Speaker{index}._sonos._tcp.local.",
162 ServiceStateChange.Added,
163 _make_discovery_info(f"sonos_player_{index}"),
164 )
165 assert len(timer_mass._tracked_timers) == 3
166
167 await provider.unload(is_removed=True)
168
169 assert timer_mass._tracked_timers == {}
170 assert provider._pending_setup_tasks == set()
171