music-assistant-server

1"""Unit tests for the Sonos S1 player provider."""
2
3from __future__ import annotations
4
5import asyncio
6import logging
7import threading
8from dataclasses import dataclass
9from typing import TYPE_CHECKING, Any
10from unittest.mock import AsyncMock, MagicMock, patch
11
12import pytest
13from music_assistant_models.enums import CoreState
14
15from music_assistant.mass import MusicAssistant
16from music_assistant.providers.sonos_s1.constants import DISCOVERY_INTERVAL
17from music_assistant.providers.sonos_s1.provider import SonosPlayerProvider
18
19if TYPE_CHECKING:
20    from collections.abc import AsyncGenerator, Callable
21
22
23@dataclass
24class DiscoveryHarness:
25    """A bare MusicAssistant together with the timers armed on its event loop."""
26
27    mass: MusicAssistant
28    scheduled: list[asyncio.TimerHandle]
29
30    @property
31    def armed_reschedules(self) -> list[asyncio.TimerHandle]:
32        """
33        Return the discovery reschedules still armed, however they were scheduled.
34
35        The reschedule is the only timer these tests arm for the long haul, so the
36        delay tells it apart from the short-lived handles asyncio arms internally.
37        """
38        now = self.mass.loop.time()
39        return [
40            handle
41            for handle in self.scheduled
42            if not handle.cancelled() and handle.when() - now > DISCOVERY_INTERVAL / 2
43        ]
44
45    def make_provider(self) -> SonosPlayerProvider:
46        """Create a SonosPlayerProvider bound to this MusicAssistant."""
47        provider = object.__new__(SonosPlayerProvider)
48        provider.mass = self.mass
49        provider.logger = MagicMock()
50        provider.config = MagicMock()
51        provider.config.instance_id = "sonos_s1--abc"
52        provider.config.get_value.side_effect = lambda key, *_args, **_kwargs: (
53            [] if "ip" in key else False
54        )
55        provider._discovery_task_id = "sonos_s1_discovery_test"
56        provider._unloaded = False
57        return provider
58
59
60@pytest.fixture
61async def harness() -> AsyncGenerator[DiscoveryHarness]:
62    """Create a bare MusicAssistant exposing the real task/timer machinery."""
63    mass = object.__new__(MusicAssistant)
64    loop = asyncio.get_running_loop()
65    mass.loop = loop
66    mass.loop_thread_id = threading.get_ident()
67    mass._tracked_timers = {}
68    mass._tracked_tasks = {}
69    mass._state = CoreState.RUNNING
70    scheduled: list[asyncio.TimerHandle] = []
71    original_call_later = loop.call_later
72
73    # record every handle armed on the loop, so a reschedule that bypasses the
74    # tracked timers is still visible to the assertions
75    def _recording_call_later(
76        delay: float, callback: Callable[..., object], *args: Any, **kwargs: Any
77    ) -> asyncio.TimerHandle:
78        handle = original_call_later(delay, callback, *args, **kwargs)
79        scheduled.append(handle)
80        return handle
81
82    loop.call_later = _recording_call_later  # type: ignore[assignment,method-assign]
83    harness = DiscoveryHarness(mass, scheduled)
84    yield harness
85    loop.call_later = original_call_later  # type: ignore[method-assign]
86    # only the reschedules are ours to disarm: the rest of the recorded handles
87    # belong to asyncio itself
88    for handle in harness.armed_reschedules:
89        handle.cancel()
90    for task in mass._tracked_tasks.values():
91        task.cancel()
92
93
94async def test_repeated_discovery_arms_a_single_reschedule(harness: DiscoveryHarness) -> None:
95    """Every load runs discovery twice, which must not leave a second reschedule armed."""
96    provider = harness.make_provider()
97
98    with patch("music_assistant.providers.sonos_s1.provider.discover", return_value=set()):
99        await provider.discover_players()
100        await provider.discover_players()
101
102    assert len(harness.armed_reschedules) == 1
103
104
105async def test_unload_disarms_the_reschedule(harness: DiscoveryHarness) -> None:
106    """An unloaded provider must not keep a discovery reschedule armed."""
107    provider = harness.make_provider()
108
109    with patch("music_assistant.providers.sonos_s1.provider.discover", return_value=set()):
110        await provider.discover_players()
111        await provider.discover_players()
112        with patch("music_assistant.providers.sonos_s1.provider.events_asyncio") as events:
113            events.event_listener = None
114            await provider.unload()
115
116    assert harness.armed_reschedules == []
117
118
119async def test_unload_aborts_a_reschedule_that_already_fired(harness: DiscoveryHarness) -> None:
120    """A rescheduled discovery that already started must be cancelled by the unload."""
121    provider = harness.make_provider()
122    loop = asyncio.get_running_loop()
123    scanning = asyncio.Event()
124    unload_reached = threading.Event()
125
126    def _blocking_discover(*_args: object, **_kwargs: object) -> set[object]:
127        loop.call_soon_threadsafe(scanning.set)
128        # hold the worker thread so the scan is unmistakably in flight during the unload
129        assert unload_reached.wait(timeout=5)
130        return set()
131
132    with patch("music_assistant.providers.sonos_s1.provider.discover", _blocking_discover):
133        harness.mass.call_later(0, provider.discover_players, task_id=provider._discovery_task_id)
134        await scanning.wait()
135        task = harness.mass._tracked_tasks[provider._discovery_task_id]
136
137        with patch("music_assistant.providers.sonos_s1.provider.events_asyncio") as events:
138            events.event_listener = None
139            unload_task = asyncio.create_task(provider.unload())
140            # the unload waits out the scan already running in its worker thread
141            await asyncio.sleep(0)
142            assert not unload_task.done()
143            unload_reached.set()
144            async with asyncio.timeout(5):
145                await unload_task
146
147    assert task.cancelled()
148    assert harness.armed_reschedules == []
149
150
151async def test_unload_stops_an_untracked_discovery_from_rescheduling(
152    harness: DiscoveryHarness,
153) -> None:
154    """A discovery the provider does not own must not reschedule itself after the unload."""
155    provider = harness.make_provider()
156    loop = asyncio.get_running_loop()
157    scanning = asyncio.Event()
158    unload_reached = threading.Event()
159
160    def _blocking_discover(*_args: object, **_kwargs: object) -> set[object]:
161        loop.call_soon_threadsafe(scanning.set)
162        assert unload_reached.wait(timeout=5)
163        return set()
164
165    with patch("music_assistant.providers.sonos_s1.provider.discover", _blocking_discover):
166        # the discovery run after a provider load is awaited directly, so it is not
167        # tracked under the provider's task id and the unload cannot cancel it
168        post_load = asyncio.create_task(provider.discover_players())
169        await scanning.wait()
170
171        with patch("music_assistant.providers.sonos_s1.provider.events_asyncio") as events:
172            events.event_listener = None
173            unload_task = asyncio.create_task(provider.unload())
174            await asyncio.sleep(0)
175            unload_reached.set()
176            async with asyncio.timeout(5):
177                await unload_task
178        async with asyncio.timeout(5):
179            await post_load
180
181    assert harness.armed_reschedules == []
182
183
184def _make_provider() -> tuple[SonosPlayerProvider, MagicMock]:
185    """Create a provider bound to a mocked server, and return both."""
186    mass = MagicMock()
187    mass.players.get_player.return_value = None
188    mass.config.get_raw_player_config_value.return_value = True
189    provider = object.__new__(SonosPlayerProvider)
190    provider.mass = mass
191    provider.logger = logging.getLogger("test.sonos_s1.provider")
192    return provider, mass
193
194
195def _make_soco(ip_address: str = "127.0.0.1") -> MagicMock:
196    """Create a mocked soco device as discovery hands it over."""
197    soco = MagicMock()
198    soco.uid = "RINCON_000E58AAAAAA01400"
199    soco.ip_address = ip_address
200    soco.is_visible = True
201    soco.fixed_volume = False
202    # a freshly discovered device has not been asked for its details yet
203    soco.speaker_info = {}
204    return soco
205
206
207async def test_disabled_speaker_is_not_interrogated() -> None:
208    """A speaker the user disabled must not be queried over the network at all."""
209    provider, mass = _make_provider()
210    mass.config.get_raw_player_config_value.return_value = False
211    soco = _make_soco()
212
213    with patch("music_assistant.providers.sonos_s1.provider.SonosPlayer") as player_cls:
214        await provider._setup_player(soco)
215
216    soco.get_speaker_info.assert_not_called()
217    player_cls.assert_not_called()
218
219
220async def test_invisible_speaker_is_not_interrogated() -> None:
221    """A bridge or stereo-pair follower is skipped before it is queried any further."""
222    provider, _ = _make_provider()
223    soco = _make_soco()
224    soco.is_visible = False
225
226    with patch("music_assistant.providers.sonos_s1.provider.SonosPlayer") as player_cls:
227        await provider._setup_player(soco)
228
229    soco.get_speaker_info.assert_not_called()
230    player_cls.assert_not_called()
231
232
233@pytest.mark.parametrize("fixed_volume", [False, True])
234async def test_new_speaker_is_registered(fixed_volume: bool) -> None:
235    """A newly discovered speaker is built with the volume mode it was interrogated for."""
236    provider, _ = _make_provider()
237    soco = _make_soco()
238    soco.fixed_volume = fixed_volume
239
240    with patch("music_assistant.providers.sonos_s1.provider.SonosPlayer") as player_cls:
241        player_cls.return_value.setup = AsyncMock()
242        await provider._setup_player(soco)
243
244    soco.get_speaker_info.assert_called_once()
245    player_cls.assert_called_once_with(provider, soco, fixed_volume=fixed_volume)
246    player_cls.return_value.setup.assert_awaited_once()
247
248
249async def test_speaker_found_at_a_new_address_is_handed_the_rediscovered_device() -> None:
250    """A known speaker seen at another address adopts the device discovery just built."""
251    provider, mass = _make_provider()
252    existing = MagicMock()
253    existing.soco.ip_address = "127.0.0.1"
254    existing.update_ip = AsyncMock()
255    mass.players.get_player.return_value = existing
256    soco = _make_soco("127.0.0.2")
257
258    await provider._setup_player(soco)
259
260    existing.update_ip.assert_awaited_once_with(soco)
261
262
263async def test_known_speaker_at_the_same_address_is_left_alone() -> None:
264    """Rediscovering a speaker at the address it already uses changes nothing."""
265    provider, mass = _make_provider()
266    existing = MagicMock()
267    existing.soco.ip_address = "127.0.0.1"
268    existing.update_ip = AsyncMock()
269    mass.players.get_player.return_value = existing
270
271    await provider._setup_player(_make_soco("127.0.0.1"))
272
273    existing.update_ip.assert_not_awaited()
274