/
/
/
1"""
2Tests for snapshotting progress at a natural end of stream (support#5813).
3
4The group can only resolve the live playback position while its push stream is
5up; the group.stop() that follows teardown would otherwise re-emit whatever
6anchor was last pushed. Freezing is limited to a clean end of stream - doing it
7for a superseded stream would flash a paused position between tracks.
8"""
9
10from __future__ import annotations
11
12from unittest.mock import MagicMock
13
14from music_assistant.providers.sendspin.playback import SendspinPlaybackSession
15
16
17def _session_mock() -> tuple[MagicMock, list[str]]:
18 """Create a session mock recording the order of progress freeze vs transport stop."""
19 session = MagicMock()
20 calls: list[str] = []
21 push_stream = MagicMock()
22 push_stream.is_stopped = False
23 push_stream.stop.side_effect = lambda *, keep_stream=False: calls.append("stream_stop") # noqa: ARG005
24 session._push_stream = push_stream
25 metadata_role = session.player._metadata_role
26 metadata_role.freeze_progress.side_effect = lambda: calls.append("freeze")
27 return session, calls
28
29
30def test_clean_eof_snapshots_progress_before_stopping_transport() -> None:
31 """The snapshot must happen while the push stream is still live."""
32 session, calls = _session_mock()
33
34 SendspinPlaybackSession._stop_push_stream(session, snapshot_progress=True)
35
36 assert calls == ["freeze", "stream_stop"]
37
38
39def test_superseded_stream_is_not_snapshotted() -> None:
40 """A stream torn down for a transition must not freeze clients at a paused position."""
41 session, calls = _session_mock()
42
43 SendspinPlaybackSession._stop_push_stream(session, snapshot_progress=False)
44
45 assert calls == ["stream_stop"]
46
47
48def test_snapshot_without_metadata_role_still_stops_transport() -> None:
49 """A group with no metadata role has nothing to freeze, but must still stop."""
50 session, calls = _session_mock()
51 session.player._metadata_role = None
52
53 SendspinPlaybackSession._stop_push_stream(session, snapshot_progress=True)
54
55 assert calls == ["stream_stop"]
56
57
58def test_already_stopped_stream_is_not_snapshotted() -> None:
59 """Nothing to freeze or stop once the stream is already down."""
60 session, calls = _session_mock()
61 session._push_stream.is_stopped = True
62
63 SendspinPlaybackSession._stop_push_stream(session, snapshot_progress=True)
64
65 assert calls == []
66