/
/
/
1"""Tests for PlayerController high-level operations.
2
3This module tests:
4- cmd_set_members validation and execution
5- Group/ungroup commands
6- Player state management
7- Cache invalidation after grouping operations
8"""
9
10from __future__ import annotations
11
12import asyncio
13import contextlib
14from unittest.mock import MagicMock
15
16import pytest
17from music_assistant_models.enums import PlayerFeature
18from music_assistant_models.errors import UnsupportedFeaturedException
19
20from music_assistant.controllers.players import PlayerController
21from music_assistant.helpers.throttle_retry import Throttler
22from tests.common import MockPlayer, MockProvider
23
24
25@pytest.fixture
26def mock_mass() -> MagicMock:
27 """Create a mock MusicAssistant instance."""
28 mass = MagicMock()
29 mass.closing = False
30 mass.loop = None
31 mass.config = MagicMock()
32 mass.config.get = MagicMock(return_value=[])
33 mass.config.get_raw_player_config_value = MagicMock(return_value="auto")
34 # Return "GLOBAL" for log level config (standard default)
35 mass.config.get_raw_core_config_value = MagicMock(return_value="GLOBAL")
36 mass.config.set = MagicMock()
37 mass.signal_event = MagicMock()
38 mass.get_providers = MagicMock(return_value=[])
39 return mass
40
41
42@pytest.fixture
43def controller(mock_mass: MagicMock) -> PlayerController:
44 """Create a PlayerController instance."""
45 return PlayerController(mock_mass)
46
47
48class TestSetMembersValidation:
49 """Test cmd_set_members validation logic."""
50
51 def test_set_members_requires_feature(self, mock_mass: MagicMock) -> None:
52 """Test that set_members requires SET_MEMBERS feature."""
53 controller = PlayerController(mock_mass)
54 provider = MockProvider("test_provider", instance_id="test", mass=mock_mass)
55
56 leader = MockPlayer(provider, "leader", "Leader")
57 # Note: NOT adding SET_MEMBERS feature
58
59 member = MockPlayer(provider, "member", "Member")
60
61 controller._players = {"leader": leader, "member": member}
62 controller._player_throttlers = {
63 "leader": Throttler(1, 0.05),
64 "member": Throttler(1, 0.05),
65 }
66 mock_mass.players = controller
67
68 # Should raise exception because leader doesn't support SET_MEMBERS
69 with pytest.raises(UnsupportedFeaturedException):
70 asyncio.run(controller.cmd_set_members("leader", player_ids_to_add=["member"]))
71
72 def test_cannot_group_incompatible_players(self, mock_mass: MagicMock) -> None:
73 """Test that incompatible players cannot be grouped."""
74 controller = PlayerController(mock_mass)
75 provider_a = MockProvider("provider_a", instance_id="provider_a", mass=mock_mass)
76 provider_b = MockProvider("provider_b", instance_id="provider_b", mass=mock_mass)
77
78 player_a = MockPlayer(provider_a, "player_a", "Player A")
79 player_a._attr_supported_features.add(PlayerFeature.SET_MEMBERS)
80 player_a._attr_can_group_with = {"provider_a"} # Only same provider
81
82 player_b = MockPlayer(provider_b, "player_b", "Player B")
83
84 controller._players = {"player_a": player_a, "player_b": player_b}
85 controller._player_throttlers = {
86 "player_a": Throttler(1, 0.05),
87 "player_b": Throttler(1, 0.05),
88 }
89 mock_mass.players = controller
90
91 # Should raise exception because players are incompatible
92 with pytest.raises(UnsupportedFeaturedException):
93 asyncio.run(controller.cmd_set_members("player_a", player_ids_to_add=["player_b"]))
94
95
96class TestCacheInvalidationAfterGrouping:
97 """Test that caches are invalidated after grouping operations."""
98
99 async def test_all_players_cache_cleared_after_set_members(self, mock_mass: MagicMock) -> None:
100 """
101 Test that all players' caches are cleared after set_members.
102
103 Regression test for: Stale can_group_with cache after grouping changes.
104 """
105 controller = PlayerController(mock_mass)
106 provider = MockProvider("test_provider", instance_id="test", mass=mock_mass)
107
108 leader = MockPlayer(provider, "leader", "Leader")
109 leader._attr_supported_features.add(PlayerFeature.SET_MEMBERS)
110 leader._attr_can_group_with = {"test"}
111 leader._attr_group_members = []
112
113 member = MockPlayer(provider, "member", "Member")
114
115 other = MockPlayer(provider, "other", "Other")
116 other._attr_supported_features.add(PlayerFeature.SET_MEMBERS)
117 other._attr_can_group_with = {"test"}
118
119 controller._players = {"leader": leader, "member": member, "other": other}
120 controller._player_throttlers = {
121 "leader": Throttler(1, 0.05),
122 "member": Throttler(1, 0.05),
123 "other": Throttler(1, 0.05),
124 }
125 mock_mass.players = controller
126
127 # Populate caches
128 _ = leader.state.can_group_with
129 _ = other.state.can_group_with
130
131 # Simulate grouping (normally done by provider's set_members implementation)
132 leader._attr_group_members = ["leader", "member"]
133
134 # Call set_members to trigger cache invalidation
135 await controller._handle_set_members_with_protocols(
136 leader, player_ids_to_add=["member"], player_ids_to_remove=[]
137 )
138
139 # Note: The actual cache clearing happens via trigger_player_update
140 # which schedules update_state to be called later
141 # In a real scenario, this would clear all players' caches
142
143
144class TestGroupUngroup:
145 """Test group and ungroup commands."""
146
147 async def test_group_command(self, mock_mass: MagicMock) -> None:
148 """Test the group command (cmd_group)."""
149 controller = PlayerController(mock_mass)
150 provider = MockProvider("test_provider", instance_id="test", mass=mock_mass)
151
152 leader = MockPlayer(provider, "leader", "Leader")
153 leader._attr_supported_features.add(PlayerFeature.SET_MEMBERS)
154 leader._attr_can_group_with = {"member"} # Leader can group with member
155
156 member = MockPlayer(provider, "member", "Member")
157 # Make sure member is already powered on to skip power handling
158 member._attr_powered = True
159
160 controller._players = {"leader": leader, "member": member}
161 controller._player_throttlers = {
162 "leader": Throttler(1, 0.05),
163 "member": Throttler(1, 0.05),
164 }
165 mock_mass.players = controller
166
167 # Update state after modifying attributes and registering with controller
168 leader.update_state(signal_event=False)
169 member.update_state(signal_event=False)
170
171 # Track if set_members was called
172 set_members_called = False
173 original_set_members = leader.set_members
174
175 async def mock_set_members(
176 player_ids_to_add: list[str] | None = None,
177 player_ids_to_remove: list[str] | None = None,
178 ) -> None:
179 nonlocal set_members_called
180 set_members_called = True
181 # Call the original to update group_members
182 await original_set_members(player_ids_to_add, player_ids_to_remove)
183
184 leader.set_members = mock_set_members # type: ignore[method-assign]
185
186 # Mock power handling to skip power control (focus is on grouping logic)
187 async def mock_handle_cmd_power(player_id: str, powered: bool) -> None:
188 pass
189
190 controller._handle_cmd_power = mock_handle_cmd_power # type: ignore[method-assign]
191
192 # Execute group command
193 await controller.cmd_group("member", "leader")
194
195 # Verify set_members was called
196 assert set_members_called
197 # Verify member was added to leader's group
198 assert "member" in leader._attr_group_members
199
200
201class TestPlayerAvailability:
202 """Test player availability checks in grouping."""
203
204 def test_unavailable_player_rejected(self, mock_mass: MagicMock) -> None:
205 """Test that unavailable players are rejected when grouping."""
206 controller = PlayerController(mock_mass)
207 provider = MockProvider("test_provider", instance_id="test", mass=mock_mass)
208
209 leader = MockPlayer(provider, "leader", "Leader")
210 leader._attr_supported_features.add(PlayerFeature.SET_MEMBERS)
211 leader._attr_can_group_with = {"test"}
212
213 member = MockPlayer(provider, "member", "Member")
214 member._attr_available = False # Mark as unavailable
215
216 controller._players = {"leader": leader, "member": member}
217 controller._player_throttlers = {
218 "leader": Throttler(1, 0.05),
219 "member": Throttler(1, 0.05),
220 }
221 mock_mass.players = controller
222
223 # Attempting to group with unavailable player should be handled
224 # (either silently ignored or raise exception depending on implementation)
225 # This should either skip the unavailable player or raise an exception
226 with contextlib.suppress(Exception):
227 asyncio.run(controller.cmd_set_members("leader", player_ids_to_add=["member"]))
228
229
230if __name__ == "__main__":
231 pytest.main([__file__, "-v"])
232