music-assistant-server

1"""
2Tests for __final_playback_state when an output protocol is active.
3
4Covers the propagation of the protocol player's state (including IDLE) up to
5the parent player that has it set as ``active_output_protocol``.
6"""
7
8from __future__ import annotations
9
10import time
11from unittest.mock import MagicMock
12
13import pytest
14from music_assistant_models.enums import PlaybackState, PlayerType
15
16from music_assistant.controllers.players import PlayerController
17from tests.common import MockPlayer, MockProvider
18
19
20@pytest.fixture
21def mock_mass() -> MagicMock:
22    """Create a mock MusicAssistant instance."""
23    mass = MagicMock()
24    mass.closing = False
25    mass.loop = None
26    mass.config = MagicMock()
27    mass.config.get = MagicMock(return_value=[])
28    mass.config.get_raw_player_config_value = MagicMock(
29        side_effect=lambda _player_id, _key, default=None: default
30    )
31    mass.config.get_raw_core_config_value = MagicMock(return_value="GLOBAL")
32    mass.config.set = MagicMock()
33    mass.signal_event = MagicMock()
34    mass.get_providers = MagicMock(return_value=[])
35    mass.player_queues = MagicMock()
36    mass.player_queues.get = MagicMock(return_value=None)
37    return mass
38
39
40@pytest.fixture
41def controller(mock_mass: MagicMock) -> PlayerController:
42    """Create a PlayerController instance."""
43    ctrl = PlayerController(mock_mass)
44    mock_mass.players = ctrl
45    return ctrl
46
47
48class TestFinalPlaybackStateWithActiveProtocol:
49    """When an active output protocol is set the protocol is the source of truth."""
50
51    def test_playing_protocol_propagates_playing(
52        self,
53        provider: MockProvider,
54        controller: PlayerController,
55    ) -> None:
56        """A PLAYING protocol player makes the parent report PLAYING."""
57        protocol_player = MockPlayer(provider, "ap_1", "AirPlay", player_type=PlayerType.PROTOCOL)
58        protocol_player._attr_playback_state = PlaybackState.PLAYING
59        protocol_player._attr_elapsed_time = 20.0
60        protocol_player._attr_elapsed_time_last_updated = time.time()
61
62        player = MockPlayer(provider, "player_1", "Test Player")
63        player._attr_playback_state = PlaybackState.IDLE  # native is idle
64        player.set_active_output_protocol("ap_1")
65
66        controller._players = {"player_1": player, "ap_1": protocol_player}
67
68        protocol_player.update_state(signal_event=False)
69        # refresh: the protocol player is cross-player state and the production
70        # fan-out (which marks the parent dirty) is suppressed in this test
71        player.refresh_state(signal_event=False)
72
73        assert player.state.playback_state == PlaybackState.PLAYING
74
75    def test_idle_protocol_propagates_idle(
76        self,
77        provider: MockProvider,
78        controller: PlayerController,
79    ) -> None:
80        """
81        An IDLE protocol player makes the parent report IDLE (regression guard).
82
83        Previously the parent would fall through to the parent/group state when
84        the protocol was IDLE, which could create a circular state inheritance
85        in sync groups and leave the group stuck in PLAYING forever. The
86        protocol player is now the source of truth regardless of its state.
87        """
88        protocol_player = MockPlayer(provider, "ap_1", "AirPlay", player_type=PlayerType.PROTOCOL)
89        protocol_player._attr_playback_state = PlaybackState.IDLE
90        protocol_player._attr_elapsed_time = 0.0
91        protocol_player._attr_elapsed_time_last_updated = time.time()
92
93        player = MockPlayer(provider, "player_1", "Test Player")
94        # Native player still reports PLAYING (e.g. a Sonos that's outputting
95        # someone else's AirPlay audio) — but the active output protocol is
96        # the live audio source and it has gone IDLE, which is what matters.
97        player._attr_playback_state = PlaybackState.PLAYING
98        player.set_active_output_protocol("ap_1")
99
100        controller._players = {"player_1": player, "ap_1": protocol_player}
101
102        protocol_player.update_state(signal_event=False)
103        player.refresh_state(signal_event=False)
104
105        assert player.state.playback_state == PlaybackState.IDLE
106
107    def test_no_active_protocol_uses_native_state(
108        self,
109        provider: MockProvider,
110        controller: PlayerController,
111    ) -> None:
112        """Without an active output protocol, fall through to the native state."""
113        player = MockPlayer(provider, "player_1", "Test Player")
114        player._attr_playback_state = PlaybackState.PLAYING
115        player._attr_elapsed_time = 5.0
116        player._attr_elapsed_time_last_updated = time.time()
117
118        controller._players = {"player_1": player}
119
120        player.update_state(signal_event=False)
121
122        assert player.state.playback_state == PlaybackState.PLAYING
123
124    def test_negative_elapsed_time_is_clamped(
125        self,
126        provider: MockProvider,
127        controller: PlayerController,
128    ) -> None:
129        """A negative provider elapsed time is clamped at the elapsed_time accessor."""
130        player = MockPlayer(provider, "player_1", "Test Player")
131        player._attr_playback_state = PlaybackState.PAUSED
132        player._attr_elapsed_time = -1.0
133        player._attr_elapsed_time_last_updated = time.time()
134
135        controller._players = {"player_1": player}
136
137        player.update_state(signal_event=False)
138
139        assert player.elapsed_time == 0
140        assert player.corrected_elapsed_time == 0
141        assert player.state.elapsed_time == 0
142
143
144def _source_session(elapsed: float | None, updated: float | None, player_id: str) -> MagicMock:
145    """Build a mock session whose source reports the given clock."""
146    stream_metadata = MagicMock()
147    stream_metadata.elapsed_time = elapsed
148    stream_metadata.elapsed_time_last_updated = updated
149    session = MagicMock()
150    session.player_id = player_id
151    session.stream_metadata = stream_metadata
152    session.source_uri = f"test_instance://audio_source/{player_id}"
153    session.source.image = None
154    return session
155
156
157class TestFinalPlaybackStateAudioSourceElapsed:
158    """The source-clock override for a live external source playing on a player."""
159
160    def test_standalone_uses_its_sources_clock(
161        self,
162        provider: MockProvider,
163        controller: PlayerController,
164        mock_mass: MagicMock,
165    ) -> None:
166        """A standalone player reports the position its live source reports."""
167        player = MockPlayer(provider, "player_1", "Player")
168        player._attr_playback_state = PlaybackState.PLAYING
169        player._attr_active_source = "player_1"
170        player._attr_elapsed_time = 0.0
171        player._attr_elapsed_time_last_updated = time.time()
172
173        session = _source_session(42.0, time.time(), "player_1")
174        mock_mass.players.get_audio_source_session = MagicMock(
175            side_effect=lambda pid: {"player_1": session}.get(pid)
176        )
177        controller._players = {"player_1": player}
178
179        player.update_state(signal_event=False)
180
181        assert player.state.elapsed_time == 42.0
182
183    def test_group_uses_its_sources_clock_not_the_leaders(
184        self,
185        provider: MockProvider,
186        controller: PlayerController,
187        mock_mass: MagicMock,
188    ) -> None:
189        """
190        A group reports its own source's position, not the clock copied from its leader.
191
192        The session belongs to the group that owns the source, so it is found directly
193        on the group and the leader never has to be consulted.
194        """
195        group = MockPlayer(provider, "group_1", "Group", player_type=PlayerType.GROUP)
196        group._attr_playback_state = PlaybackState.PLAYING
197        group._attr_active_source = "leader_1"
198        # stale leader byte clock that gets copied into the group
199        group._attr_elapsed_time = 999.0
200        group._attr_elapsed_time_last_updated = time.time()
201
202        session = _source_session(42.0, time.time(), "group_1")
203        mock_mass.players.get_audio_source_session = MagicMock(
204            side_effect=lambda pid: {"group_1": session}.get(pid)
205        )
206        controller._players = {"group_1": group}
207
208        group.update_state(signal_event=False)
209
210        assert group.state.elapsed_time == 42.0
211
212    def test_a_player_without_a_source_keeps_its_own_clock(
213        self,
214        provider: MockProvider,
215        controller: PlayerController,
216        mock_mass: MagicMock,
217    ) -> None:
218        """A player with no live source of its own is not given another player's position."""
219        player = MockPlayer(provider, "player_1", "Player")
220        player._attr_playback_state = PlaybackState.PLAYING
221        player._attr_active_source = "other_src"
222        player._attr_elapsed_time = 5.0
223        player._attr_elapsed_time_last_updated = time.time()
224
225        elsewhere = _source_session(42.0, time.time(), "other_player")
226        mock_mass.players.get_audio_source_session = MagicMock(
227            side_effect=lambda pid: {"other_player": elsewhere}.get(pid)
228        )
229        controller._players = {"player_1": player}
230
231        player.update_state(signal_event=False)
232
233        assert player.state.elapsed_time == 5.0
234