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