music-assistant-server

1"""Tests for player grouping logic (independent of protocols).
2
3This module tests the core grouping behavior including:
4- can_group_with filtering logic
5- Group member inclusion/exclusion
6- Sync leader behavior
7- Group state transitions
8- Cache invalidation
9"""
10
11from __future__ import annotations
12
13from unittest.mock import MagicMock
14
15import pytest
16from music_assistant_models.enums import PlaybackState, PlayerFeature
17
18from music_assistant.controllers.players import PlayerController
19from tests.common import MockPlayer, MockProvider
20
21
22@pytest.fixture
23def mock_mass() -> MagicMock:
24    """Create a mock MusicAssistant instance."""
25    mass = MagicMock()
26    mass.closing = False
27    mass.config = MagicMock()
28    mass.config.get = MagicMock(return_value=[])
29    mass.config.get_raw_player_config_value = MagicMock(return_value="auto")
30    # Return "GLOBAL" for log level config (standard default)
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    return mass
36
37
38@pytest.fixture
39def controller(mock_mass: MagicMock) -> PlayerController:
40    """Create a PlayerController instance."""
41    return PlayerController(mock_mass)
42
43
44class TestCanGroupWithBasics:
45    """Test basic can_group_with filtering logic."""
46
47    def test_ungrouped_players_can_group(self, mock_mass: MagicMock) -> None:
48        """Test that two ungrouped players can group with each other."""
49        controller = PlayerController(mock_mass)
50        provider = MockProvider("test_provider", instance_id="test", mass=mock_mass)
51
52        player_a = MockPlayer(provider, "player_a", "Player A")
53        player_a._attr_supported_features.add(PlayerFeature.SET_MEMBERS)
54        # Use explicit player IDs instead of provider instance ID for simpler test
55        player_a._attr_can_group_with = {"player_b"}
56
57        player_b = MockPlayer(provider, "player_b", "Player B")
58        player_b._attr_supported_features.add(PlayerFeature.SET_MEMBERS)
59        player_b._attr_can_group_with = {"player_a"}
60
61        controller._players = {"player_a": player_a, "player_b": player_b}
62        mock_mass.players = controller
63
64        # Trigger state calculation
65        player_a.update_state(signal_event=False)
66        player_b.update_state(signal_event=False)
67
68        # Both players should be able to group with each other
69        assert "player_b" in player_a.state.can_group_with
70        assert "player_a" in player_b.state.can_group_with
71
72    def test_unavailable_players_excluded(self, mock_mass: MagicMock) -> None:
73        """Test that unavailable players are excluded from can_group_with."""
74        controller = PlayerController(mock_mass)
75        provider = MockProvider("test_provider", instance_id="test", mass=mock_mass)
76
77        player_a = MockPlayer(provider, "player_a", "Player A")
78        player_a._attr_supported_features.add(PlayerFeature.SET_MEMBERS)
79        player_a._attr_can_group_with = {"player_b"}
80
81        player_b = MockPlayer(provider, "player_b", "Player B")
82        player_b._attr_available = False  # Mark as unavailable
83
84        controller._players = {"player_a": player_a, "player_b": player_b}
85        mock_mass.players = controller
86
87        # Trigger state calculation
88        player_a.update_state(signal_event=False)
89        player_b.update_state(signal_event=False)
90
91        # Unavailable player should be excluded
92        assert "player_b" not in player_a.state.can_group_with
93
94    def test_playing_players_with_different_source_excluded(self, mock_mass: MagicMock) -> None:
95        """Test that players playing different sources are NOT excluded (behavior changed).
96
97        Note: Previously, players with different active sources were excluded from grouping,
98        but this was removed as it was difficult to track reliably.
99        """
100        controller = PlayerController(mock_mass)
101        provider = MockProvider("test_provider", instance_id="test", mass=mock_mass)
102
103        player_a = MockPlayer(provider, "player_a", "Player A")
104        player_a._attr_supported_features.add(PlayerFeature.SET_MEMBERS)
105        player_a._attr_can_group_with = {"player_b"}
106        player_a._attr_playback_state = PlaybackState.PLAYING
107        player_a._attr_active_source = "player_a"
108
109        player_b = MockPlayer(provider, "player_b", "Player B")
110        player_b._attr_playback_state = PlaybackState.PLAYING
111        player_b._attr_active_source = "player_b"  # Different source
112
113        controller._players = {"player_a": player_a, "player_b": player_b}
114        mock_mass.players = controller
115
116        # Trigger state calculation
117        player_a.update_state(signal_event=False)
118        player_b.update_state(signal_event=False)
119
120        # Player with different active source is now ALLOWED (behavior changed)
121        assert "player_b" in player_a.state.can_group_with
122
123
124class TestSyncedPlayers:
125    """Test behavior with synced/grouped players."""
126
127    def test_synced_player_excluded_from_others(self, mock_mass: MagicMock) -> None:
128        """
129        Test that a player synced to another is excluded from other players' can_group_with.
130
131        Regression test for: Player synced to another showing up in third player's can_group_with.
132        """
133        controller = PlayerController(mock_mass)
134        provider = MockProvider("test_provider", instance_id="test", mass=mock_mass)
135
136        # Sync leader
137        leader = MockPlayer(provider, "leader", "Leader")
138        leader._attr_supported_features.add(PlayerFeature.SET_MEMBERS)
139        leader._attr_can_group_with = {"synced", "other"}
140        leader._attr_group_members = ["leader", "synced"]
141        leader._attr_playback_state = PlaybackState.PLAYING  # Make it playing so it gets excluded
142
143        # Synced player
144        synced = MockPlayer(provider, "synced", "Synced")
145
146        # Third player
147        other = MockPlayer(provider, "other", "Other")
148        other._attr_supported_features.add(PlayerFeature.SET_MEMBERS)
149        other._attr_can_group_with = {"leader", "synced"}
150
151        controller._players = {"leader": leader, "synced": synced, "other": other}
152        mock_mass.players = controller
153
154        # Trigger synced_to calculation
155        leader.update_state(signal_event=False)
156        synced.update_state(signal_event=False)
157        other.update_state(signal_event=False)
158
159        # The synced player should NOT appear in other's can_group_with
160        assert "synced" not in other.state.can_group_with
161        # The leader should also NOT appear (has group members)
162        assert "leader" not in other.state.can_group_with
163        # Other should only see itself as ungrouped
164        assert other.state.can_group_with == set()
165
166    def test_sync_leader_excludes_itself_from_members_can_group_with(
167        self, mock_mass: MagicMock
168    ) -> None:
169        """Test that sync leader doesn't appear in its members' can_group_with."""
170        controller = PlayerController(mock_mass)
171        provider = MockProvider("test_provider", instance_id="test", mass=mock_mass)
172
173        leader = MockPlayer(provider, "leader", "Leader")
174        leader._attr_supported_features.add(PlayerFeature.SET_MEMBERS)
175        leader._attr_can_group_with = {"member"}
176        leader._attr_group_members = ["leader", "member"]
177
178        member = MockPlayer(provider, "member", "Member")
179
180        controller._players = {"leader": leader, "member": member}
181        mock_mass.players = controller
182
183        # Trigger synced_to calculation
184        leader.update_state(signal_event=False)
185        member.update_state(signal_event=False)
186
187        # Member is synced, so can_group_with should be empty
188        assert member.state.can_group_with == set()
189
190    def test_group_members_included_in_leader_can_group_with(self, mock_mass: MagicMock) -> None:
191        """
192        Test that group members appear in sync leader's can_group_with.
193
194        This allows ungrouping members from the leader.
195        """
196        controller = PlayerController(mock_mass)
197        provider = MockProvider("test_provider", instance_id="test", mass=mock_mass)
198
199        leader = MockPlayer(provider, "leader", "Leader")
200        leader._attr_supported_features.add(PlayerFeature.SET_MEMBERS)
201        leader._attr_can_group_with = {"member_a", "member_b"}
202        leader._attr_group_members = ["leader", "member_a", "member_b"]
203
204        member_a = MockPlayer(provider, "member_a", "Member A")
205        member_b = MockPlayer(provider, "member_b", "Member B")
206
207        controller._players = {
208            "leader": leader,
209            "member_a": member_a,
210            "member_b": member_b,
211        }
212        mock_mass.players = controller
213
214        # Trigger synced_to calculation
215        leader.update_state(signal_event=False)
216        member_a.update_state(signal_event=False)
217        member_b.update_state(signal_event=False)
218
219        # Leader should be able to see its own members (for ungrouping)
220        assert "member_a" in leader.state.can_group_with
221        assert "member_b" in leader.state.can_group_with
222
223
224class TestSyncLeaderBehavior:
225    """Test sync leader specific behavior."""
226
227    def test_sync_leader_excluded_from_can_group_with(self, mock_mass: MagicMock) -> None:
228        """Test that players with group members (sync leaders) are excluded."""
229        controller = PlayerController(mock_mass)
230        provider = MockProvider("test_provider", instance_id="test", mass=mock_mass)
231
232        leader = MockPlayer(provider, "leader", "Leader")
233        leader._attr_supported_features.add(PlayerFeature.SET_MEMBERS)
234        leader._attr_can_group_with = {"member", "other"}
235        leader._attr_group_members = ["leader", "member"]
236        leader._attr_playback_state = PlaybackState.PLAYING  # Make it playing so it gets excluded
237
238        member = MockPlayer(provider, "member", "Member")
239
240        other = MockPlayer(provider, "other", "Other")
241        other._attr_supported_features.add(PlayerFeature.SET_MEMBERS)
242        other._attr_can_group_with = {"leader", "member"}
243
244        controller._players = {"leader": leader, "member": member, "other": other}
245        mock_mass.players = controller
246
247        # Trigger synced_to calculation
248        leader.update_state(signal_event=False)
249        member.update_state(signal_event=False)
250        other.update_state(signal_event=False)
251
252        # Leader should NOT appear in other's can_group_with (has group members)
253        assert "leader" not in other.state.can_group_with
254
255
256class TestCircularDependency:
257    """Test that circular dependencies are avoided."""
258
259    def test_no_circular_dependency_in_synced_to(self, mock_mass: MagicMock) -> None:
260        """
261        Test that synced_to calculation doesn't cause circular dependency.
262
263        Regression test for: synced_to calling group_members causing infinite recursion.
264        """
265        controller = PlayerController(mock_mass)
266        provider = MockProvider("test_provider", instance_id="test", mass=mock_mass)
267
268        leader = MockPlayer(provider, "leader", "Leader")
269        leader._attr_group_members = ["leader", "member"]
270
271        member = MockPlayer(provider, "member", "Member")
272
273        controller._players = {"leader": leader, "member": member}
274        mock_mass.players = controller
275
276        # Trigger synced_to calculation via update_state
277        leader.update_state(signal_event=False)
278        member.update_state(signal_event=False)
279
280        # This should not cause infinite recursion
281        assert member.state.synced_to == "leader"
282        assert leader.state.synced_to is None
283
284
285class TestCacheInvalidation:
286    """Test that caches are invalidated correctly."""
287
288    def test_can_group_with_cache_cleared_on_update_state(self, mock_mass: MagicMock) -> None:
289        """Test that can_group_with cache is cleared when update_state is called."""
290        controller = PlayerController(mock_mass)
291        provider = MockProvider("test_provider", instance_id="test", mass=mock_mass)
292
293        player_a = MockPlayer(provider, "player_a", "Player A")
294        player_a._attr_supported_features.add(PlayerFeature.SET_MEMBERS)
295        player_a._attr_can_group_with = {"player_b"}
296
297        player_b = MockPlayer(provider, "player_b", "Player B")
298
299        controller._players = {"player_a": player_a, "player_b": player_b}
300        mock_mass.players = controller
301
302        # Update state after setting attributes and registering with controller
303        player_a.update_state(signal_event=False)
304        player_b.update_state(signal_event=False)
305
306        # Get can_group_with to populate cache
307        initial = player_a.state.can_group_with
308        assert "player_b" in initial
309
310        # Modify underlying data
311        player_a._attr_can_group_with = set()
312
313        # Cache should still have old value
314        assert player_a.state.can_group_with == initial
315
316        # Clear cache via update_state
317        player_a.update_state(signal_event=False)
318
319        # Cache should be cleared, new value should be returned
320        assert player_a.state.can_group_with == set()
321
322
323class TestProviderInstanceIdExpansion:
324    """Test expansion of provider instance IDs in can_group_with."""
325
326    def test_provider_instance_id_expands_to_all_players(self, mock_mass: MagicMock) -> None:
327        """Test that provider instance IDs expand to all available players from that provider."""
328        controller = PlayerController(mock_mass)
329        provider = MockProvider("test_provider", instance_id="test", mass=mock_mass)
330
331        player_a = MockPlayer(provider, "player_a", "Player A")
332        player_a._attr_supported_features.add(PlayerFeature.SET_MEMBERS)
333        player_a._attr_can_group_with = {"test"}  # Provider instance ID
334
335        player_b = MockPlayer(provider, "player_b", "Player B")
336        player_c = MockPlayer(provider, "player_c", "Player C")
337
338        controller._players = {
339            "player_a": player_a,
340            "player_b": player_b,
341            "player_c": player_c,
342        }
343        mock_mass.players = controller
344        # Set up get_provider to return the provider for instance ID
345        mock_mass.get_provider = MagicMock(return_value=provider)
346
347        # Trigger state calculation
348        player_a.update_state(signal_event=False)
349        player_b.update_state(signal_event=False)
350        player_c.update_state(signal_event=False)
351
352        # Provider instance ID should expand to include all players from that provider
353        can_group = player_a.state.can_group_with
354        assert "player_b" in can_group
355        assert "player_c" in can_group
356
357
358if __name__ == "__main__":
359    pytest.main([__file__, "-v"])
360