/
/
/
1"""
2Tests for how BluOS transport states map onto playback states.
3
4BluOS reports 'connecting' while it (re)fills its buffer, which includes the stretch where
5it plays out the tail of a stream that stopped sending. Reporting that as idle ends the
6queue while the player is still making sound, so a running stream stays 'playing' until
7BluOS reports a real stop.
8"""
9
10from __future__ import annotations
11
12from typing import cast
13from unittest.mock import MagicMock
14
15import pytest
16from music_assistant_models.enums import PlaybackState
17
18from music_assistant.providers.bluesound.const import MAX_CONNECTING_POLLS
19from music_assistant.providers.bluesound.player import BluesoundPlayer
20
21
22def _resolve(bluos_state: str, current_state: PlaybackState) -> PlaybackState:
23 """Resolve the playback state for a BluOS state reported on top of a current state."""
24 fake = MagicMock()
25 fake.status.state = bluos_state
26 fake._attr_playback_state = current_state
27 fake._connecting_polls = 0
28 return BluesoundPlayer._resolve_playback_state(cast("BluesoundPlayer", fake))
29
30
31def _poll(fake: MagicMock, bluos_state: str) -> PlaybackState:
32 """Report one poll of the given BluOS state and feed the result back as current state."""
33 fake.status.state = bluos_state
34 state = BluesoundPlayer._resolve_playback_state(cast("BluesoundPlayer", fake))
35 fake._attr_playback_state = state
36 return state
37
38
39def test_connecting_while_playing_stays_playing() -> None:
40 """A buffering stream must not be mistaken for the end of the queue."""
41 assert _resolve("connecting", PlaybackState.PLAYING) == PlaybackState.PLAYING
42
43
44@pytest.mark.parametrize("current", [PlaybackState.IDLE, PlaybackState.PAUSED])
45def test_connecting_outside_playback_is_idle(current: PlaybackState) -> None:
46 """Connecting from a standstill is the device starting up, not playback."""
47 assert _resolve("connecting", current) == PlaybackState.IDLE
48
49
50@pytest.mark.parametrize(
51 ("bluos_state", "expected"),
52 [
53 ("play", PlaybackState.PLAYING),
54 ("stream", PlaybackState.PLAYING),
55 ("stop", PlaybackState.IDLE),
56 ("pause", PlaybackState.PAUSED),
57 ],
58)
59def test_reported_states_are_mapped(bluos_state: str, expected: PlaybackState) -> None:
60 """Every other BluOS state maps straight through, including a stop while playing."""
61 assert _resolve(bluos_state, PlaybackState.PLAYING) == expected
62
63
64def test_a_player_stuck_connecting_ends_up_idle() -> None:
65 """A player that never resumes must be reported idle so playback can recover."""
66 fake = MagicMock()
67 fake._attr_playback_state = PlaybackState.PLAYING
68 fake._connecting_polls = 0
69
70 states = [_poll(fake, "connecting") for _ in range(MAX_CONNECTING_POLLS + 1)]
71
72 assert states[:-1] == [PlaybackState.PLAYING] * MAX_CONNECTING_POLLS
73 assert states[-1] == PlaybackState.IDLE
74
75
76def test_resumed_playback_clears_the_connecting_grace() -> None:
77 """Buffering that resolves must not spend the grace of a later interruption."""
78 fake = MagicMock()
79 fake._attr_playback_state = PlaybackState.PLAYING
80 fake._connecting_polls = 0
81
82 assert _poll(fake, "connecting") == PlaybackState.PLAYING
83 assert _poll(fake, "stream") == PlaybackState.PLAYING
84 assert fake._connecting_polls == 0
85 # the full grace is available again
86 for _ in range(MAX_CONNECTING_POLLS):
87 assert _poll(fake, "connecting") == PlaybackState.PLAYING
88