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