music-assistant-server

1"""Shared fakes for the Sendspin Source provider tests."""
2
3from __future__ import annotations
4
5import asyncio
6from types import SimpleNamespace
7from typing import TYPE_CHECKING, Any, cast
8
9import pytest
10from music_assistant_models.enums import PlaybackState
11
12from music_assistant.providers.sendspin_source.provider import SendspinSourceProvider
13
14if TYPE_CHECKING:
15    from collections.abc import Callable
16
17    from music_assistant.mass import MusicAssistant
18
19
20class _FakeSourceRole:
21    """Source role stand-in recording start/stop commands."""
22
23    def __init__(self) -> None:
24        self.start_requests = 0
25        self.stop_requests = 0
26        self.stream_active = False
27
28    def request_start(self) -> None:
29        self.start_requests += 1
30
31    def request_stop(self) -> None:
32        self.stop_requests += 1
33
34
35class _FakeClient:
36    """Server-side SendspinClient stand-in."""
37
38    def __init__(
39        self,
40        client_id: str,
41        *,
42        name: str | None = None,
43        has_source_role: bool = True,
44        connected: bool = True,
45    ) -> None:
46        self.client_id = client_id
47        self.info_or_none = None
48        if name is not None:
49            self.info_or_none = type("Info", (), {"name": name})()
50        self.is_connected = connected
51        self.negotiated_role_ids = ["source@v1"] if has_source_role else ["player@v1"]
52        self.source_role = _FakeSourceRole() if has_source_role else None
53        self.listeners: list[Callable[[Any, Any], None]] = []
54
55    def roles_by_family(self, family: str) -> list[Any]:
56        if family == "source" and self.source_role is not None:
57            return [self.source_role]
58        return []
59
60    def detach_roles(self) -> _FakeSourceRole | None:
61        """Drop the role instances, as a cold reconnect does before re-attaching them."""
62        role, self.source_role = self.source_role, None
63        return role
64
65    def attach_roles(self, role: _FakeSourceRole | None) -> None:
66        """Re-attach role instances, which the server does after signalling connected."""
67        self.source_role = role
68
69    def add_event_listener(self, callback: Callable[[Any, Any], None]) -> Callable[[], None]:
70        self.listeners.append(callback)
71        return lambda: self.listeners.remove(callback)
72
73    def emit(self, event: Any) -> None:
74        for callback in list(self.listeners):
75            callback(self, event)
76
77
78class _FakeServerApi:
79    """SendspinServer stand-in serving a fixed client set."""
80
81    def __init__(self, clients: list[_FakeClient]) -> None:
82        self._clients = {client.client_id: client for client in clients}
83        self.listeners: list[Callable[[Any, Any], None]] = []
84
85    @property
86    def connected_clients(self) -> list[_FakeClient]:
87        return [c for c in self._clients.values() if c.is_connected]
88
89    def get_client(self, client_id: str) -> _FakeClient | None:
90        return self._clients.get(client_id)
91
92    def add_event_listener(self, callback: Callable[[Any, Any], None]) -> Callable[[], None]:
93        self.listeners.append(callback)
94        return lambda: self.listeners.remove(callback)
95
96    def emit(self, event: Any) -> None:
97        for callback in list(self.listeners):
98            callback(self, event)
99
100
101class _FakeQueue:
102    """PlayerQueue stand-in."""
103
104    def __init__(self, queue_id: str, state: PlaybackState = PlaybackState.IDLE) -> None:
105        self.queue_id = queue_id
106        self.state = state
107
108
109class _FakePlayer:
110    """Player stand-in owning a queue."""
111
112    def __init__(self, player_id: str, state: PlaybackState = PlaybackState.IDLE) -> None:
113        self.player_id = player_id
114        self.queue = _FakeQueue(player_id, state)
115
116
117class _FakePlayers:
118    """Players controller stand-in recording stop commands."""
119
120    def __init__(self) -> None:
121        self.stopped: list[str] = []
122        self.players: dict[str, _FakePlayer] = {}
123        self.stop_started: asyncio.Event | None = None
124        self.release_stop: asyncio.Event | None = None
125        # live source sessions, keyed on the player playing one
126        self.source_sessions: dict[str, Any] = {}
127        self.deselected: list[str] = []
128
129    def get_player(self, player_id: str, *args: Any, **kwargs: Any) -> _FakePlayer | None:
130        return self.players.get(player_id)
131
132    def get_active_queue(self, player: _FakePlayer) -> _FakeQueue:
133        return player.queue
134
135    def get_audio_source_session(self, player_id: str) -> Any:
136        return self.source_sessions.get(player_id)
137
138    async def deselect_source(
139        self,
140        player_id: str,
141        stop_playback: bool = True,
142        provider_instance_id: str | None = None,
143        source_id: str | None = None,
144        playback_session_id: str | None = None,
145    ) -> None:
146        self.deselected.append(player_id)
147        if stop_playback:
148            await self.cmd_stop(player_id)
149
150    async def cmd_stop(self, player_id: str) -> None:
151        self.stopped.append(player_id)
152        if self.stop_started is not None:
153            self.stop_started.set()
154        if self.release_stop is not None:
155            await self.release_stop.wait()
156
157
158class _FakePlayerQueues:
159    """Queue controller stand-in recording play_media calls."""
160
161    def __init__(self) -> None:
162        self.played: list[tuple[str, str, Any]] = []
163        self.stopped: list[str] = []
164        self.play_started: asyncio.Event | None = None
165        self.release_play: asyncio.Event | None = None
166        self.stop_started: asyncio.Event | None = None
167        self.release_stop: asyncio.Event | None = None
168        self._sessions: dict[str, str | None] = {}
169        self._session_counter = 0
170
171    async def play_media(self, queue_id: str, media: Any, option: Any = None, **_: Any) -> None:
172        self._session_counter += 1
173        self._sessions[queue_id] = f"session-{self._session_counter}"
174        if self.play_started is not None:
175            self.play_started.set()
176        if self.release_play is not None:
177            await self.release_play.wait()
178        self.played.append((queue_id, media, option))
179
180    async def stop(self, queue_id: str) -> None:
181        self.stopped.append(queue_id)
182        if self.stop_started is not None:
183            self.stop_started.set()
184        if self.release_stop is not None:
185            await self.release_stop.wait()
186        self._sessions[queue_id] = None
187
188    def queue_data(self, queue_id: str) -> Any:
189        return SimpleNamespace(session_id=self._sessions.get(queue_id))
190
191
192class _FakeConfigController:
193    """Config controller stand-in separating stored values from entry defaults."""
194
195    def __init__(self) -> None:
196        self.values: dict[tuple[str, str], Any] = {}
197        self.defaults: dict[tuple[str, str], Any] = {}
198
199    def get_raw_player_config_value(self, player_id: str, key: str, default: Any = None) -> Any:
200        return self.values.get((player_id, key), default)
201
202    async def get_player_config_value(
203        self, player_id: str, key: str, *, default: Any = None
204    ) -> Any:
205        if (stored := self.values.get((player_id, key))) is not None:
206            return stored
207        return self.defaults.get((player_id, key), default)
208
209
210class _FakeMass:
211    """MusicAssistant stand-in providing loop, task creation and provider lookup."""
212
213    cache = None
214
215    def __init__(self, sendspin_provider: Any) -> None:
216        self.loop = asyncio.get_running_loop()
217        self.players = _FakePlayers()
218        self.player_queues = _FakePlayerQueues()
219        self.config = _FakeConfigController()
220        self._sendspin_provider = sendspin_provider
221        self._timers: dict[str, asyncio.TimerHandle] = {}
222        self._tasks: dict[str, asyncio.Task[Any]] = {}
223
224    def get_provider(self, domain: str) -> Any:
225        if domain == "sendspin":
226            return self._sendspin_provider
227        return None
228
229    def create_task(
230        self,
231        coro: Any,
232        *,
233        task_id: str | None = None,
234        abort_existing: bool = False,
235        eager_start: bool = True,
236    ) -> asyncio.Task[Any]:
237        # Mirror the real controller's eager default, which decides whether a task
238        # body runs inside the event callback that created it.
239        if task_id is not None and (existing := self._tasks.get(task_id)) is not None:
240            if not abort_existing:
241                coro.close()
242                return existing
243            existing.cancel()
244        task = asyncio.Task(coro, loop=self.loop, eager_start=eager_start)
245        if task_id is not None:
246            self._tasks[task_id] = task
247            task.add_done_callback(
248                lambda completed: (
249                    self._tasks.pop(task_id, None)
250                    if self._tasks.get(task_id) is completed
251                    else None
252                )
253            )
254        return task
255
256    def call_later(
257        self,
258        delay: float,
259        target: Callable[..., Any],
260        *args: Any,
261        task_id: str | None = None,
262        **kwargs: Any,
263    ) -> asyncio.TimerHandle:
264        timer_id = task_id or str(id(target))
265        self.cancel_timer(timer_id)
266
267        def run() -> None:
268            self._timers.pop(timer_id, None)
269            self.create_task(target(*args, **kwargs), task_id=timer_id, abort_existing=True)
270
271        handle = self.loop.call_later(delay, run)
272        self._timers[timer_id] = handle
273        return handle
274
275    def cancel_timer(self, task_id: str) -> None:
276        if (timer := self._timers.pop(task_id, None)) is not None:
277            timer.cancel()
278
279    def get_task(self, task_id: str) -> asyncio.Task[Any] | None:
280        return self._tasks.get(task_id)
281
282    def cancel_task(self, task_id: str) -> None:
283        if (task := self._tasks.pop(task_id, None)) is not None:
284            task.cancel()
285
286
287class _FakeConfig:
288    """
289    Provider config stand-in.
290
291    Values default to unset, as they are at load time: a provider's own entries are
292    only resolved once its instance is loaded, so nothing has been parsed yet.
293    """
294
295    instance_id = "sendspin_source"
296
297    def __init__(self, values: dict[str, Any] | None = None) -> None:
298        self.values = values or {}
299
300    def get_value(self, key: str, default: Any = None) -> Any:
301        return self.values.get(key, default)
302
303
304async def make_provider(
305    clients: list[_FakeClient], config_values: dict[str, Any] | None = None
306) -> SendspinSourceProvider:
307    """Build a provider wired to fake mass/server_api around the given clients."""
308    server_api = _FakeServerApi(clients)
309    sendspin_provider = type("FakeSendspinProvider", (), {"server_api": server_api})()
310    # Constructed rather than hand-populated, so a new instance attribute cannot go
311    # missing here and take the fake out of step with the provider.
312    provider = SendspinSourceProvider(
313        cast("MusicAssistant", _FakeMass(sendspin_provider)),
314        cast("Any", type("Manifest", (), {"domain": "sendspin_source"})()),
315        cast("Any", _FakeConfig(config_values)),
316    )
317    for client in clients:
318        get_players(provider).players[client.client_id] = _FakePlayer(client.client_id)
319    await provider.loaded_in_mass()
320    return provider
321
322
323def get_config(provider: SendspinSourceProvider) -> _FakeConfigController:
324    """Return the fake config controller the given provider is wired to."""
325    return cast("_FakeConfigController", provider.mass.config)
326
327
328def get_queues(provider: SendspinSourceProvider) -> _FakePlayerQueues:
329    """Return the fake queue controller the given provider is wired to."""
330    return cast("_FakePlayerQueues", provider.mass.player_queues)
331
332
333def get_server_api(provider: SendspinSourceProvider) -> _FakeServerApi:
334    """Return the fake server api the given provider is wired to."""
335    sendspin = cast("Any", provider.mass.get_provider("sendspin"))
336    return cast("_FakeServerApi", sendspin.server_api)
337
338
339def get_players(provider: SendspinSourceProvider) -> _FakePlayers:
340    """Return the fake players controller the given provider is wired to."""
341    return cast("_FakePlayers", provider.mass.players)
342
343
344@pytest.fixture
345def fake_client() -> _FakeClient:
346    """Return a connected client with an active source role."""
347    return _FakeClient("client-1", name="Turntable")
348