/
/
/
1"""Tests for Sendspin playback cancellation."""
2
3from __future__ import annotations
4
5import asyncio
6from collections import deque
7from typing import cast
8from unittest.mock import MagicMock
9
10import pytest
11
12from music_assistant.providers.sendspin.playback import SendspinPlaybackSession
13
14
15def _session_mock(
16 *, allow_noncompliant: bool
17) -> tuple[SendspinPlaybackSession, list[str | tuple[str, bool]]]:
18 """Create a session with a live push stream recording the order of teardown calls."""
19 session = object.__new__(SendspinPlaybackSession)
20 session.player = MagicMock()
21 session.player.provider.server_api.allow_noncompliant_clients = allow_noncompliant
22 session._cancel_requested = False
23 calls: list[str | tuple[str, bool]] = []
24 push_stream = MagicMock()
25 push_stream.is_stopped = False
26 push_stream.clear.side_effect = lambda: calls.append("stream_clear")
27 push_stream.stop.side_effect = lambda *, keep_stream: calls.append(("stream_stop", keep_stream))
28 session._push_stream = push_stream
29
30 async def playback() -> None:
31 try:
32 await asyncio.Future()
33 finally:
34 calls.append("task_cleanup")
35
36 session.playback_task = asyncio.create_task(playback())
37 return session, calls
38
39
40async def test_transition_clears_stream_before_task_cleanup() -> None:
41 """A spec-only fleet gets the stream cleared and preserved before task teardown."""
42 session, calls = _session_mock(allow_noncompliant=False)
43 await asyncio.sleep(0)
44
45 await session.cancel("new media requested", keep_stream=True)
46
47 assert calls == ["stream_clear", ("stream_stop", True), "task_cleanup"]
48
49
50async def test_transition_leaves_teardown_to_task_while_noncompliant_allowed() -> None:
51 """Legacy clients mishandle stream/clear, so cancel leaves the stop to task teardown."""
52 session, calls = _session_mock(allow_noncompliant=True)
53 await asyncio.sleep(0)
54
55 await session.cancel("new media requested", keep_stream=True)
56
57 assert calls == ["task_cleanup"]
58
59
60async def test_transition_awaits_task_when_stopping_the_stream_fails() -> None:
61 """A stream stop failing on a dead connection must not abort the track change."""
62 session, calls = _session_mock(allow_noncompliant=False)
63 cast("MagicMock", session._push_stream).stop.side_effect = ConnectionResetError("client gone")
64 await asyncio.sleep(0)
65
66 await session.cancel("new media requested", keep_stream=True)
67
68 assert calls == ["stream_clear", "task_cleanup"]
69 assert session.playback_task is None
70
71
72async def test_setup_failure_ends_the_new_stream(monkeypatch: pytest.MonkeyPatch) -> None:
73 """A session that fails during setup ends its stream instead of leaving clients hanging."""
74 session = object.__new__(SendspinPlaybackSession)
75 session.player = MagicMock()
76 session._state_lock = asyncio.Lock()
77 session._pipeline_config_cache = {}
78 session._push_stream = None
79 session._history = deque()
80 session._produced_audio_us = 0
81 session._timeline_start_us = None
82 session._first_commit_monotonic_us = None
83 push_stream = MagicMock()
84 push_stream.set_live_source.side_effect = RuntimeError("no clients")
85 monkeypatch.setattr(
86 session, "_select_session_pcm_formats", MagicMock(return_value=(MagicMock(), MagicMock()))
87 )
88 monkeypatch.setattr(session, "_create_push_stream", MagicMock(return_value=push_stream))
89
90 with pytest.raises(RuntimeError):
91 await session._run_playback(MagicMock())
92
93 push_stream.stop.assert_called_once_with()
94 assert session._push_stream is None
95 assert session._playback_running is False
96