music-assistant-server

1"""Tests for the Sendspin Source provider."""
2
3from __future__ import annotations
4
5import asyncio
6from typing import TYPE_CHECKING, Any
7
8import pytest
9from aiosendspin.audio import AudioFormat as SendspinAudioFormat
10from aiosendspin.server import (
11    ClientConnectedEvent,
12    SignalState,
13    SourceSignalChangedEvent,
14    SourceStreamStartedEvent,
15)
16from music_assistant_models.enums import MediaType, PlaybackState, QueueOption, StreamType
17from music_assistant_models.errors import AudioError, MediaNotFoundError
18from music_assistant_models.streamdetails import StreamDetails
19
20import music_assistant.providers.sendspin_source.provider as provider_module
21from music_assistant.constants import CONF_ENTRY_WARN_PREVIEW
22from music_assistant.providers.sendspin.constants import CONF_SOURCE_AUTOSTART_TARGET
23from music_assistant.providers.sendspin_source import get_config_entries
24from music_assistant.providers.sendspin_source.constants import (
25    CONF_TARGET_LATENCY,
26    DEFAULT_TARGET_LATENCY_MS,
27)
28from music_assistant.providers.sendspin_source.provider import OUTPUT_FORMAT
29
30from .conftest import (
31    _FakeClient,
32    get_config,
33    get_players,
34    get_queues,
35    get_server_api,
36    make_provider,
37)
38
39if TYPE_CHECKING:
40    from collections.abc import AsyncGenerator
41
42MARKER_BYTE = b"\x7f"
43NATIVE_FORMAT = SendspinAudioFormat(sample_rate=44100, bit_depth=16, channels=2)
44
45
46class _StubBridge:
47    """Bridge stand-in returning marker bytes and recording feeds."""
48
49    occupancy_us = 0
50
51    def __init__(self) -> None:
52        self.fed: list[tuple[bytes, int]] = []
53        self.flush_count = 0
54
55    def feed(self, pcm: bytes, capture_timestamp_us: int) -> None:
56        self.fed.append((pcm, capture_timestamp_us))
57
58    def read(self, frames: int) -> bytes:
59        return MARKER_BYTE * (frames * 4)
60
61    def flush(self) -> None:
62        self.flush_count += 1
63
64
65def _stream_details(item_id: str = "client-1") -> StreamDetails:
66    return StreamDetails(
67        provider="sendspin_source",
68        item_id=item_id,
69        audio_format=OUTPUT_FORMAT,
70        media_type=MediaType.AUDIO_SOURCE,
71        stream_type=StreamType.CUSTOM,
72    )
73
74
75async def test_config_entries_start_with_preview_warning() -> None:
76    """The alpha provider warns users before showing its options."""
77    mass: Any = None
78    entries = await get_config_entries(mass)
79    assert entries[0] is CONF_ENTRY_WARN_PREVIEW
80
81
82async def _take(stream: AsyncGenerator[bytes], count: int) -> list[bytes]:
83    chunks: list[bytes] = []
84    try:
85        async for chunk in stream:
86            chunks.append(chunk)
87            if len(chunks) >= count:
88                break
89    finally:
90        await stream.aclose()
91    return chunks
92
93
94async def _fake_handle(chunks: list[tuple[bytes, int]]) -> Any:
95    for chunk in chunks:
96        yield chunk
97
98
99async def test_audio_sources_follow_role_activation() -> None:
100    """Only connected clients with an active source role are listed."""
101    provider = await make_provider(
102        [
103            _FakeClient("with-role", name="Turntable"),
104            _FakeClient("no-role", name="Speaker", has_source_role=False),
105            _FakeClient("offline", name="Gone", connected=False),
106        ]
107    )
108    sources = await provider.get_audio_sources()
109    assert [s.item_id for s in sources] == ["with-role"]
110    assert sources[0].name == "Turntable"
111    assert sources[0].exclusive is True
112    assert sources[0].can_initiate is True
113
114
115async def test_stream_details_rejects_unknown_source() -> None:
116    """get_stream_details raises for a client that is not a connected source."""
117    provider = await make_provider([_FakeClient("no-role", has_source_role=False)])
118    with pytest.raises(MediaNotFoundError):
119        await provider.get_stream_details("no-role", MediaType.AUDIO_SOURCE)
120    with pytest.raises(MediaNotFoundError):
121        await provider.get_stream_details("unknown", MediaType.AUDIO_SOURCE)
122
123
124async def test_select_requests_start_and_subscribes(fake_client: _FakeClient) -> None:
125    """Selecting a source sends server/command start and attaches a client listener."""
126    provider = await make_provider([fake_client])
127    await provider.on_source_selected("client-1", "player-1", "queue-1", "session-1")
128    assert fake_client.source_role is not None
129    assert fake_client.source_role.start_requests == 1
130    assert len(fake_client.listeners) == 1
131
132
133async def test_unselect_stops_and_releases(fake_client: _FakeClient) -> None:
134    """Unselecting with the live session id sends stop and drops the session."""
135    provider = await make_provider([fake_client])
136    await provider.on_source_selected("client-1", "player-1", "queue-1", "session-1")
137    await provider.on_source_unselected("client-1", "queue-1", "session-1")
138    assert fake_client.source_role is not None
139    assert fake_client.source_role.stop_requests == 1
140    with pytest.raises(AudioError):
141        await _take(provider.get_audio_stream(_stream_details()), 1)
142    assert get_players(provider).stopped == []
143    # The signal watcher outlives the session, so autostart still works afterwards.
144    assert len(fake_client.listeners) == 1
145
146
147async def test_unselect_ignores_stale_session_id(
148    fake_client: _FakeClient, monkeypatch: pytest.MonkeyPatch
149) -> None:
150    """A stale unselect callback must not tear down the live session."""
151    provider = await make_provider([fake_client])
152    await _start_streaming(provider, fake_client, monkeypatch)
153    await provider.on_source_selected("client-1", "player-1", "queue-1", "session-2")
154    await provider.on_source_unselected("client-1", "queue-1", "session-1")
155    assert fake_client.source_role is not None
156    assert fake_client.source_role.stop_requests == 0
157    assert await _take(provider.get_audio_stream(_stream_details()), 1)
158    assert len(fake_client.listeners) == 1
159
160
161async def _start_streaming(
162    provider: Any,
163    client: _FakeClient,
164    monkeypatch: pytest.MonkeyPatch,
165    chunks: list[tuple[bytes, int]] | None = None,
166) -> _StubBridge:
167    """Select the source and let one chunk through, so the stream is past its cold start."""
168    bridge = _StubBridge()
169    monkeypatch.setattr(provider, "_create_bridge", lambda *_args: bridge)
170    await provider.on_source_selected(client.client_id, "player-1", "queue-1", "session-1")
171    handle = _fake_handle(chunks if chunks is not None else [(b"\x01\x02\x03\x04", 1_000_000)])
172    client.emit(SourceStreamStartedEvent(audio_format=NATIVE_FORMAT, handle=handle))
173    await _settle()
174    return bridge
175
176
177async def _latency_passed_to_bridge(
178    provider: Any, client: _FakeClient, monkeypatch: pytest.MonkeyPatch
179) -> list[int]:
180    """Select the source and return the target latencies the bridge was built with."""
181    latencies: list[int] = []
182
183    def _create(_audio_format: Any, target_latency_ms: int) -> _StubBridge:
184        latencies.append(target_latency_ms)
185        return _StubBridge()
186
187    monkeypatch.setattr(provider, "_create_bridge", _create)
188    await provider.on_source_selected(client.client_id, "player-1", "queue-1", "session-1")
189    client.emit(SourceStreamStartedEvent(audio_format=NATIVE_FORMAT, handle=_fake_handle([])))
190    await _settle()
191    return latencies
192
193
194async def test_bridge_falls_back_to_the_default_latency(
195    fake_client: _FakeClient, monkeypatch: pytest.MonkeyPatch
196) -> None:
197    """A provider starts up before its own options are resolved, so the value is unset."""
198    provider = await make_provider([fake_client])
199    assert await _latency_passed_to_bridge(provider, fake_client, monkeypatch) == [
200        DEFAULT_TARGET_LATENCY_MS
201    ]
202
203
204async def test_a_configured_latency_reaches_the_bridge(
205    fake_client: _FakeClient, monkeypatch: pytest.MonkeyPatch
206) -> None:
207    """Once the user sets a latency it must win over the fallback."""
208    provider = await make_provider([fake_client], {CONF_TARGET_LATENCY: 1200})
209    assert await _latency_passed_to_bridge(provider, fake_client, monkeypatch) == [1200]
210
211
212async def test_stream_fails_when_the_source_never_starts(
213    fake_client: _FakeClient, monkeypatch: pytest.MonkeyPatch
214) -> None:
215    """A client that never answers the start command is a failed acquisition, not silence."""
216    monkeypatch.setattr(provider_module, "COLD_START_TIMEOUT_S", 0.01)
217    provider = await make_provider([fake_client])
218    await provider.on_source_selected("client-1", "player-1", "queue-1", "session-1")
219    with pytest.raises(AudioError):
220        await _take(provider.get_audio_stream(_stream_details()), 1)
221
222
223async def test_stream_switches_to_bridge_audio_after_stream_start(
224    fake_client: _FakeClient, monkeypatch: pytest.MonkeyPatch
225) -> None:
226    """A client_stream/start routes decoded chunks through the bridge into the stream."""
227    provider = await make_provider([fake_client])
228    bridge = await _start_streaming(provider, fake_client, monkeypatch)
229
230    chunks = await _take(provider.get_audio_stream(_stream_details()), 5)
231    assert all(chunk == MARKER_BYTE * (48000 * 25 // 1000 * 4) for chunk in chunks)
232    assert bridge.fed == [(b"\x01\x02\x03\x04", 1_000_000)]
233    await provider.on_source_unselected("client-1", "queue-1", "session-1")
234
235
236async def test_source_stream_end_flushes_the_live_bridge(
237    fake_client: _FakeClient, monkeypatch: pytest.MonkeyPatch
238) -> None:
239    """A normally ended source stream emits the resampler tail."""
240    provider = await make_provider([fake_client])
241    bridge = await _start_streaming(provider, fake_client, monkeypatch)
242    assert bridge.flush_count == 1
243
244
245async def test_stream_ends_after_source_timeout(
246    fake_client: _FakeClient, monkeypatch: pytest.MonkeyPatch
247) -> None:
248    """The generator ends on its own once no source audio arrives for the timeout."""
249    provider = await make_provider([fake_client])
250    await _start_streaming(provider, fake_client, monkeypatch)
251    monkeypatch.setattr(provider_module, "SOURCE_TIMEOUT_S", 0.05)
252    chunks = [chunk async for chunk in provider.get_audio_stream(_stream_details())]
253    assert 1 <= len(chunks) <= 10
254
255
256async def test_reconnect_re_requests_start(fake_client: _FakeClient) -> None:
257    """A reconnect clears the client's start request, so the provider sends it again."""
258    provider = await make_provider([fake_client])
259    await provider.on_source_selected("client-1", "player-1", "queue-1", "session-1")
260    assert fake_client.source_role is not None
261    assert fake_client.source_role.start_requests == 1
262    get_server_api(provider).emit(ClientConnectedEvent("client-1"))
263    await asyncio.sleep(0)
264    assert fake_client.source_role.start_requests == 2
265    await provider.on_source_unselected("client-1", "queue-1", "session-1")
266
267
268async def test_cold_reconnect_re_requests_start_once_roles_are_back(
269    fake_client: _FakeClient,
270) -> None:
271    """Roles attach after the connected signal, so the re-request must not run inside it."""
272    provider = await make_provider([fake_client])
273    await provider.on_source_selected("client-1", "player-1", "queue-1", "session-1")
274    role = fake_client.detach_roles()
275    get_server_api(provider).emit(ClientConnectedEvent("client-1"))
276    fake_client.attach_roles(role)
277    await _settle()
278    assert role is not None
279    assert role.start_requests == 2
280
281
282async def test_signal_watcher_arms_before_roles_attach() -> None:
283    """Watching keys off negotiated roles, which are known before the instances attach."""
284    client = _FakeClient("client-1", name="Turntable", connected=False)
285    provider = await make_provider([client])
286    assert client.listeners == []
287    client.is_connected = True
288    client.detach_roles()
289    get_server_api(provider).emit(ClientConnectedEvent("client-1"))
290    await _settle()
291    assert len(client.listeners) == 1
292
293
294async def test_reconnect_leaves_an_open_stream_alone(fake_client: _FakeClient) -> None:
295    """A client that kept its input stream across the event is not asked to start again."""
296    provider = await make_provider([fake_client])
297    await provider.on_source_selected("client-1", "player-1", "queue-1", "session-1")
298    assert fake_client.source_role is not None
299    fake_client.source_role.stream_active = True
300    get_server_api(provider).emit(ClientConnectedEvent("client-1"))
301    await asyncio.sleep(0)
302    assert fake_client.source_role.start_requests == 1
303    await provider.on_source_unselected("client-1", "queue-1", "session-1")
304
305
306async def test_handoff_stops_the_player_it_was_taken_from(fake_client: _FakeClient) -> None:
307    """Moving a source to another player stops the first, which would drain its buffer."""
308    provider = await make_provider([fake_client])
309    await provider.on_source_selected("client-1", "player-1", "queue-1", "session-1")
310    await provider.on_source_selected("client-1", "player-2", "queue-2", "session-2")
311    assert get_players(provider).stopped == ["player-1"]
312    # The client must be stopped too: only a fresh client_stream/start gives the
313    # replacement session a bridge, and the client sends one after a stop/start.
314    assert fake_client.source_role is not None
315    assert fake_client.source_role.stop_requests == 1
316
317
318async def test_reclaim_by_the_same_queue_keeps_the_client_streaming(
319    fake_client: _FakeClient, monkeypatch: pytest.MonkeyPatch
320) -> None:
321    """A renderer opening the stream url twice must not cost a stop/start of the source."""
322    provider = await make_provider([fake_client])
323    await _start_streaming(provider, fake_client, monkeypatch)
324    await provider.on_source_selected("client-1", "player-1", "queue-1", "session-2")
325    assert fake_client.source_role is not None
326    assert fake_client.source_role.stop_requests == 0
327    assert fake_client.source_role.start_requests == 1
328    assert get_players(provider).stopped == []
329    assert await _take(provider.get_audio_stream(_stream_details()), 1)
330
331
332async def test_reclaim_by_the_same_queue_retires_the_previous_generator(
333    fake_client: _FakeClient, monkeypatch: pytest.MonkeyPatch
334) -> None:
335    """Only one generator may read the bridge, or the two requests split the audio."""
336    provider = await make_provider([fake_client])
337    await _start_streaming(provider, fake_client, monkeypatch)
338    stream = provider.get_audio_stream(_stream_details())
339    assert await anext(stream) is not None
340    await provider.on_source_selected("client-1", "player-1", "queue-1", "session-2")
341    assert [chunk async for chunk in stream] == []
342
343
344async def test_reclaim_while_waiting_for_audio_retires_the_previous_generator(
345    fake_client: _FakeClient, monkeypatch: pytest.MonkeyPatch
346) -> None:
347    """A same-queue reclaim retires a generator that is still waiting for first audio."""
348    provider = await make_provider([fake_client])
349    bridge = _StubBridge()
350    monkeypatch.setattr(provider, "_create_bridge", lambda *_args: bridge)
351    await provider.on_source_selected("client-1", "player-1", "queue-1", "session-1")
352    previous = provider.get_audio_stream(_stream_details())
353    previous_chunk = asyncio.create_task(anext(previous))
354    await asyncio.sleep(0)
355
356    await provider.on_source_selected("client-1", "player-1", "queue-1", "session-2")
357    replacement = provider.get_audio_stream(_stream_details())
358    replacement_chunk = asyncio.create_task(anext(replacement))
359    fake_client.emit(
360        SourceStreamStartedEvent(
361            audio_format=NATIVE_FORMAT,
362            handle=_fake_handle([(b"\x01\x02\x03\x04", 1_000_000)]),
363        )
364    )
365    await _settle()
366
367    with pytest.raises(StopAsyncIteration):
368        await previous_chunk
369    assert await replacement_chunk == MARKER_BYTE * (48000 * 25 // 1000 * 4)
370    await replacement.aclose()
371
372
373async def test_reclaim_by_the_same_player_on_another_queue_still_hands_off(
374    fake_client: _FakeClient,
375) -> None:
376    """A different queue is a real re-target, so the previous claim is torn down."""
377    provider = await make_provider([fake_client])
378    await provider.on_source_selected("client-1", "player-1", "queue-1", "session-1")
379    await provider.on_source_selected("client-1", "player-1", "queue-2", "session-2")
380    assert fake_client.source_role is not None
381    assert fake_client.source_role.stop_requests == 1
382    assert get_players(provider).stopped == []
383
384
385async def test_concurrent_handoffs_leave_the_newest_selection_active(
386    fake_client: _FakeClient,
387) -> None:
388    """Concurrent handoffs serialize so an older stop cannot overwrite the latest claim."""
389    provider = await make_provider([fake_client])
390    await provider.on_source_selected("client-1", "player-1", "queue-1", "session-1")
391    players = get_players(provider)
392    players.stop_started = asyncio.Event()
393    players.release_stop = asyncio.Event()
394
395    first = asyncio.create_task(
396        provider.on_source_selected("client-1", "player-2", "queue-2", "session-2")
397    )
398    await players.stop_started.wait()
399    second = asyncio.create_task(
400        provider.on_source_selected("client-1", "player-3", "queue-3", "session-3")
401    )
402    await asyncio.sleep(0)
403    players.release_stop.set()
404    await asyncio.gather(first, second)
405
406    await provider.on_source_unselected("client-1", "queue-3", "session-3")
407    assert players.stopped == ["player-1", "player-2"]
408    assert fake_client.source_role is not None
409    assert fake_client.source_role.stop_requests == 3
410
411
412async def test_two_sources_stream_concurrently(fake_client: _FakeClient) -> None:
413    """Selecting a second source must not stop the first: exclusivity is per source."""
414    other = _FakeClient("client-2", name="Aux")
415    provider = await make_provider([fake_client, other])
416    await provider.on_source_selected("client-1", "player-1", "queue-1", "session-1")
417    await provider.on_source_selected("client-2", "player-2", "queue-2", "session-2")
418    assert fake_client.source_role is not None
419    assert other.source_role is not None
420    assert fake_client.source_role.stop_requests == 0
421    assert other.source_role.start_requests == 1
422    assert get_players(provider).stopped == []
423
424
425async def test_new_selection_supersedes_running_stream(
426    fake_client: _FakeClient, monkeypatch: pytest.MonkeyPatch
427) -> None:
428    """Re-selecting the same source elsewhere makes the previous generator terminate."""
429    provider = await make_provider([fake_client])
430    await _start_streaming(provider, fake_client, monkeypatch)
431    stream = provider.get_audio_stream(_stream_details())
432    assert await anext(stream) is not None
433    await provider.on_source_selected("client-1", "player-2", "queue-2", "session-2")
434    chunks = [chunk async for chunk in stream]
435    assert len(chunks) <= 1
436
437
438def _signal(state: SignalState) -> SourceSignalChangedEvent:
439    return SourceSignalChangedEvent(signal=state)
440
441
442@pytest.fixture
443def fast_autostart(monkeypatch: pytest.MonkeyPatch) -> None:
444    """Collapse the debounce and hold so autostart timing is not a test dependency."""
445    monkeypatch.setattr(provider_module, "AUTOSTART_SIGNAL_DEBOUNCE_S", 0.0)
446    monkeypatch.setattr(provider_module, "AUTOSTART_SIGNAL_ABSENT_HOLD_S", 0.0)
447
448
449async def _settle() -> None:
450    for _ in range(4):
451        await asyncio.sleep(0)
452
453
454@pytest.mark.usefixtures("fast_autostart")
455async def test_signal_returning_autostarts_configured_target(fake_client: _FakeClient) -> None:
456    """A signal appearing after being absent starts the source on the configured player."""
457    provider = await make_provider([fake_client])
458    get_config(provider).values[("client-1", CONF_SOURCE_AUTOSTART_TARGET)] = "client-1"
459    fake_client.emit(_signal(SignalState.ABSENT))
460    fake_client.emit(_signal(SignalState.PRESENT))
461    await _settle()
462    assert get_queues(provider).played == [
463        ("client-1", "sendspin_source://audio_source/client-1", QueueOption.PLAY)
464    ]
465
466
467@pytest.mark.usefixtures("fast_autostart")
468async def test_autostart_uses_the_entry_default_when_nothing_was_saved(
469    fake_client: _FakeClient,
470) -> None:
471    """A device that plays its own line-in must work before the user ever saves the page."""
472    provider = await make_provider([fake_client])
473    get_config(provider).defaults[("client-1", CONF_SOURCE_AUTOSTART_TARGET)] = "client-1"
474    fake_client.emit(_signal(SignalState.ABSENT))
475    fake_client.emit(_signal(SignalState.PRESENT))
476    await _settle()
477    assert get_queues(provider).played == [
478        ("client-1", "sendspin_source://audio_source/client-1", QueueOption.PLAY)
479    ]
480
481
482@pytest.mark.usefixtures("fast_autostart")
483async def test_first_signal_report_never_autostarts(fake_client: _FakeClient) -> None:
484    """A needle already down at startup must not start playing by itself."""
485    provider = await make_provider([fake_client])
486    get_config(provider).values[("client-1", CONF_SOURCE_AUTOSTART_TARGET)] = "client-1"
487    fake_client.emit(_signal(SignalState.PRESENT))
488    await _settle()
489    assert get_queues(provider).played == []
490
491
492async def test_transient_signal_does_not_autostart(
493    fake_client: _FakeClient, monkeypatch: pytest.MonkeyPatch
494) -> None:
495    """A signal that disappears during the debounce never starts playback."""
496    monkeypatch.setattr(provider_module, "AUTOSTART_SIGNAL_DEBOUNCE_S", 60.0)
497    provider = await make_provider([fake_client])
498    get_config(provider).values[("client-1", CONF_SOURCE_AUTOSTART_TARGET)] = "client-1"
499    fake_client.emit(_signal(SignalState.ABSENT))
500    fake_client.emit(_signal(SignalState.PRESENT))
501    fake_client.emit(_signal(SignalState.ABSENT))
502    await _settle()
503    assert get_queues(provider).played == []
504    await provider.unload()
505
506
507async def test_manual_selection_cancels_a_fired_autostart(
508    fake_client: _FakeClient, monkeypatch: pytest.MonkeyPatch
509) -> None:
510    """A manual selection wins while a fired autostart is starting playback."""
511    provider = await make_provider([fake_client])
512    get_config(provider).values[("client-1", CONF_SOURCE_AUTOSTART_TARGET)] = "client-1"
513    queues = get_queues(provider)
514    queues.play_started = asyncio.Event()
515    queues.release_play = asyncio.Event()
516    monkeypatch.setattr(provider_module, "AUTOSTART_SIGNAL_DEBOUNCE_S", 0.0)
517    fake_client.emit(_signal(SignalState.ABSENT))
518    fake_client.emit(_signal(SignalState.PRESENT))
519    await queues.play_started.wait()
520
521    await provider.on_source_selected("client-1", "player-2", "queue-2", "session-1")
522    queues.release_play.set()
523    await _settle()
524    assert queues.played == []
525
526
527@pytest.mark.usefixtures("fast_autostart")
528async def test_delayed_autostart_stream_cannot_override_manual_selection(
529    fake_client: _FakeClient,
530) -> None:
531    """A delayed stream request from a completed autostart cannot reclaim the source."""
532    provider = await make_provider([fake_client])
533    get_config(provider).values[("client-1", CONF_SOURCE_AUTOSTART_TARGET)] = "client-1"
534    fake_client.emit(_signal(SignalState.ABSENT))
535    fake_client.emit(_signal(SignalState.PRESENT))
536    await _settle()
537
538    queues = get_queues(provider)
539    # the handoff suspends while it releases the autostart player, which is where the
540    # delayed request has to arrive for this to be the race it is testing
541    players = get_players(provider)
542    players.stop_started = asyncio.Event()
543    players.release_stop = asyncio.Event()
544    manual = asyncio.create_task(
545        provider.on_source_selected("client-1", "player-2", "queue-2", "session-1")
546    )
547    await players.stop_started.wait()
548    delayed = asyncio.create_task(
549        provider.on_source_selected("client-1", "client-1", "client-1", "session-2")
550    )
551    await asyncio.sleep(0)
552    players.release_stop.set()
553    await manual
554    with pytest.raises(RuntimeError, match="Superseded autostart"):
555        await delayed
556
557    await queues.play_media("client-1", "sendspin_source://audio_source/client-1")
558    await provider.on_source_selected("client-1", "client-1", "client-1", "session-3")
559    assert fake_client.source_role is not None
560    assert fake_client.source_role.start_requests == 2
561
562
563@pytest.mark.usefixtures("fast_autostart")
564async def test_autostart_stays_off_without_a_configured_target(fake_client: _FakeClient) -> None:
565    """With no target configured the signal is observed but never acted on."""
566    provider = await make_provider([fake_client])
567    fake_client.emit(_signal(SignalState.ABSENT))
568    fake_client.emit(_signal(SignalState.PRESENT))
569    await _settle()
570    assert get_queues(provider).played == []
571
572
573@pytest.mark.usefixtures("fast_autostart")
574async def test_autostart_interrupts_a_busy_target(fake_client: _FakeClient) -> None:
575    """A detected signal takes over a target that is already playing."""
576    provider = await make_provider([fake_client])
577    get_config(provider).values[("client-1", CONF_SOURCE_AUTOSTART_TARGET)] = "client-1"
578    get_players(provider).players["client-1"].queue.state = PlaybackState.PLAYING
579    fake_client.emit(_signal(SignalState.ABSENT))
580    fake_client.emit(_signal(SignalState.PRESENT))
581    await _settle()
582    assert get_queues(provider).played == [
583        ("client-1", "sendspin_source://audio_source/client-1", QueueOption.PLAY)
584    ]
585
586
587@pytest.mark.usefixtures("fast_autostart")
588async def test_autostart_skipped_while_the_source_already_streams(fake_client: _FakeClient) -> None:
589    """A source the user already routed somewhere must not be stolen back by a signal."""
590    provider = await make_provider([fake_client])
591    get_config(provider).values[("client-1", CONF_SOURCE_AUTOSTART_TARGET)] = "client-1"
592    await provider.on_source_selected("client-1", "player-2", "queue-2", "session-1")
593    fake_client.emit(_signal(SignalState.ABSENT))
594    fake_client.emit(_signal(SignalState.PRESENT))
595    await _settle()
596    assert get_queues(provider).played == []
597
598
599@pytest.mark.usefixtures("fast_autostart")
600async def test_signal_loss_stops_a_running_source(fake_client: _FakeClient) -> None:
601    """A record reaching its end stops the stream instead of playing silence forever."""
602    provider = await make_provider([fake_client])
603    await provider.on_source_selected("client-1", "player-1", "queue-1", "session-1")
604    fake_client.emit(_signal(SignalState.PRESENT))
605    fake_client.emit(_signal(SignalState.ABSENT))
606    await _settle()
607    # the source is given up on the player that owns it, so the player stops saying
608    # it is playing something that has gone quiet
609    assert get_players(provider).deselected == ["queue-1"]
610
611
612async def test_signal_return_cancels_pending_autostop(
613    fake_client: _FakeClient, monkeypatch: pytest.MonkeyPatch
614) -> None:
615    """A signal returning during the hold keeps the running source playing."""
616    monkeypatch.setattr(provider_module, "AUTOSTART_SIGNAL_ABSENT_HOLD_S", 60.0)
617    provider = await make_provider([fake_client])
618    await provider.on_source_selected("client-1", "player-1", "queue-1", "session-1")
619    fake_client.emit(_signal(SignalState.PRESENT))
620    fake_client.emit(_signal(SignalState.ABSENT))
621    fake_client.emit(_signal(SignalState.PRESENT))
622    await _settle()
623    assert get_queues(provider).stopped == []
624    await provider.unload()
625
626
627@pytest.mark.usefixtures("fast_autostart")
628async def test_a_reconnect_does_not_defuse_a_pending_autostop(fake_client: _FakeClient) -> None:
629    """A same-queue reconnect re-claims the source and must not cancel the pending stop."""
630    provider = await make_provider([fake_client])
631    await provider.on_source_selected("client-1", "player-1", "queue-1", "session-1")
632    fake_client.emit(_signal(SignalState.PRESENT))
633    fake_client.emit(_signal(SignalState.ABSENT))
634    await provider.on_source_selected("client-1", "player-1", "queue-1", "session-2")
635    await _settle()
636    assert get_players(provider).deselected == ["queue-1"]
637
638
639async def test_deferred_reconnect_does_not_rewatch_after_unload(fake_client: _FakeClient) -> None:
640    """A connected callback queued before unload cannot restore its client listener."""
641    provider = await make_provider([fake_client])
642    get_server_api(provider).emit(ClientConnectedEvent("client-1"))
643    await provider.unload()
644    await _settle()
645    assert fake_client.listeners == []
646