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(self, player_id: str, stop_playback: bool = True) -> None:
139        self.deselected.append(player_id)
140        if stop_playback:
141            await self.cmd_stop(player_id)
142
143    async def cmd_stop(self, player_id: str) -> None:
144        self.stopped.append(player_id)
145        if self.stop_started is not None:
146            self.stop_started.set()
147        if self.release_stop is not None:
148            await self.release_stop.wait()
149
150
151class _FakePlayerQueues:
152    """Queue controller stand-in recording play_media calls."""
153
154    def __init__(self) -> None:
155        self.played: list[tuple[str, str, Any]] = []
156        self.stopped: list[str] = []
157        self.play_started: asyncio.Event | None = None
158        self.release_play: asyncio.Event | None = None
159        self.stop_started: asyncio.Event | None = None
160        self.release_stop: asyncio.Event | None = None
161        self._sessions: dict[str, str | None] = {}
162        self._session_counter = 0
163
164    async def play_media(self, queue_id: str, media: Any, option: Any = None, **_: Any) -> None:
165        self._session_counter += 1
166        self._sessions[queue_id] = f"session-{self._session_counter}"
167        if self.play_started is not None:
168            self.play_started.set()
169        if self.release_play is not None:
170            await self.release_play.wait()
171        self.played.append((queue_id, media, option))
172
173    async def stop(self, queue_id: str) -> None:
174        self.stopped.append(queue_id)
175        if self.stop_started is not None:
176            self.stop_started.set()
177        if self.release_stop is not None:
178            await self.release_stop.wait()
179        self._sessions[queue_id] = None
180
181    def queue_data(self, queue_id: str) -> Any:
182        return SimpleNamespace(session_id=self._sessions.get(queue_id))
183
184
185class _FakeConfigController:
186    """Config controller stand-in separating stored values from entry defaults."""
187
188    def __init__(self) -> None:
189        self.values: dict[tuple[str, str], Any] = {}
190        self.defaults: dict[tuple[str, str], Any] = {}
191
192    def get_raw_player_config_value(self, player_id: str, key: str, default: Any = None) -> Any:
193        return self.values.get((player_id, key), default)
194
195    async def get_player_config_value(
196        self, player_id: str, key: str, *, default: Any = None
197    ) -> Any:
198        if (stored := self.values.get((player_id, key))) is not None:
199            return stored
200        return self.defaults.get((player_id, key), default)
201
202
203class _FakeMass:
204    """MusicAssistant stand-in providing loop, task creation and provider lookup."""
205
206    cache = None
207
208    def __init__(self, sendspin_provider: Any) -> None:
209        self.loop = asyncio.get_running_loop()
210        self.players = _FakePlayers()
211        self.player_queues = _FakePlayerQueues()
212        self.config = _FakeConfigController()
213        self._sendspin_provider = sendspin_provider
214        self._timers: dict[str, asyncio.TimerHandle] = {}
215        self._tasks: dict[str, asyncio.Task[Any]] = {}
216
217    def get_provider(self, domain: str) -> Any:
218        if domain == "sendspin":
219            return self._sendspin_provider
220        return None
221
222    def create_task(
223        self,
224        coro: Any,
225        *,
226        task_id: str | None = None,
227        abort_existing: bool = False,
228        eager_start: bool = True,
229    ) -> asyncio.Task[Any]:
230        # Mirror the real controller's eager default, which decides whether a task
231        # body runs inside the event callback that created it.
232        if task_id is not None and (existing := self._tasks.get(task_id)) is not None:
233            if not abort_existing:
234                coro.close()
235                return existing
236            existing.cancel()
237        task = asyncio.Task(coro, loop=self.loop, eager_start=eager_start)
238        if task_id is not None:
239            self._tasks[task_id] = task
240            task.add_done_callback(
241                lambda completed: (
242                    self._tasks.pop(task_id, None)
243                    if self._tasks.get(task_id) is completed
244                    else None
245                )
246            )
247        return task
248
249    def call_later(
250        self,
251        delay: float,
252        target: Callable[..., Any],
253        *args: Any,
254        task_id: str | None = None,
255        **kwargs: Any,
256    ) -> asyncio.TimerHandle:
257        timer_id = task_id or str(id(target))
258        self.cancel_timer(timer_id)
259
260        def run() -> None:
261            self._timers.pop(timer_id, None)
262            self.create_task(target(*args, **kwargs), task_id=timer_id, abort_existing=True)
263
264        handle = self.loop.call_later(delay, run)
265        self._timers[timer_id] = handle
266        return handle
267
268    def cancel_timer(self, task_id: str) -> None:
269        if (timer := self._timers.pop(task_id, None)) is not None:
270            timer.cancel()
271
272    def get_task(self, task_id: str) -> asyncio.Task[Any] | None:
273        return self._tasks.get(task_id)
274
275    def cancel_task(self, task_id: str) -> None:
276        if (task := self._tasks.pop(task_id, None)) is not None:
277            task.cancel()
278
279
280class _FakeConfig:
281    """
282    Provider config stand-in.
283
284    Values default to unset, as they are at load time: a provider's own entries are
285    only resolved once its instance is loaded, so nothing has been parsed yet.
286    """
287
288    instance_id = "sendspin_source"
289
290    def __init__(self, values: dict[str, Any] | None = None) -> None:
291        self.values = values or {}
292
293    def get_value(self, key: str, default: Any = None) -> Any:
294        return self.values.get(key, default)
295
296
297async def make_provider(
298    clients: list[_FakeClient], config_values: dict[str, Any] | None = None
299) -> SendspinSourceProvider:
300    """Build a provider wired to fake mass/server_api around the given clients."""
301    server_api = _FakeServerApi(clients)
302    sendspin_provider = type("FakeSendspinProvider", (), {"server_api": server_api})()
303    # Constructed rather than hand-populated, so a new instance attribute cannot go
304    # missing here and take the fake out of step with the provider.
305    provider = SendspinSourceProvider(
306        cast("MusicAssistant", _FakeMass(sendspin_provider)),
307        cast("Any", type("Manifest", (), {"domain": "sendspin_source"})()),
308        cast("Any", _FakeConfig(config_values)),
309    )
310    for client in clients:
311        get_players(provider).players[client.client_id] = _FakePlayer(client.client_id)
312    await provider.loaded_in_mass()
313    return provider
314
315
316def get_config(provider: SendspinSourceProvider) -> _FakeConfigController:
317    """Return the fake config controller the given provider is wired to."""
318    return cast("_FakeConfigController", provider.mass.config)
319
320
321def get_queues(provider: SendspinSourceProvider) -> _FakePlayerQueues:
322    """Return the fake queue controller the given provider is wired to."""
323    return cast("_FakePlayerQueues", provider.mass.player_queues)
324
325
326def get_server_api(provider: SendspinSourceProvider) -> _FakeServerApi:
327    """Return the fake server api the given provider is wired to."""
328    sendspin = cast("Any", provider.mass.get_provider("sendspin"))
329    return cast("_FakeServerApi", sendspin.server_api)
330
331
332def get_players(provider: SendspinSourceProvider) -> _FakePlayers:
333    """Return the fake players controller the given provider is wired to."""
334    return cast("_FakePlayers", provider.mass.players)
335
336
337@pytest.fixture
338def fake_client() -> _FakeClient:
339    """Return a connected client with an active source role."""
340    return _FakeClient("client-1", name="Turntable")
341