music-assistant-server

1"""
2Tests for the Universal Group Player lifecycle.
3
4Mirrors the structure of ``tests/providers/sync_group/test_sync_group.py`` but covers the
5multicast-stream variant (Universal Group). The new lifecycle replaces the
6mandatory power control with form-on-play, dissolve-on-stop, and a debounced
7idle grace window; these tests pin that behavior in place.
8"""
9
10from __future__ import annotations
11
12import asyncio
13from typing import Any
14from unittest.mock import AsyncMock, MagicMock, patch
15
16import pytest
17from music_assistant_models.constants import (
18    PLAYER_CONTROL_FAKE,
19    PLAYER_CONTROL_NATIVE,
20    PLAYER_CONTROL_NONE,
21)
22from music_assistant_models.enums import PlaybackState, PlayerFeature
23
24from music_assistant.providers.universal_group.player import UniversalGroupPlayer
25
26
27def _make_mock_mass() -> MagicMock:
28    """Create a minimal mock MusicAssistant for the UGP tests."""
29    mass = MagicMock()
30    mass.players = MagicMock()
31    mass.players.get_player = MagicMock(return_value=None)
32    mass.players._handle_cmd_stop = AsyncMock()
33    mass.players._handle_play_media = AsyncMock()
34    mass.players.cmd_power = AsyncMock()
35    mass.players.iter_group_members = MagicMock(return_value=[])
36
37    wait_ctx = AsyncMock()
38    wait_ctx.__aenter__.return_value = None
39    wait_ctx.__aexit__.return_value = False
40    mass.players.wait_for_player_update = MagicMock(return_value=wait_ctx)
41
42    mass.config = MagicMock()
43
44    def _config_value(key: str, default: object = None) -> object:
45        if key == "group_members":
46            return []
47        if key == "dynamic_members":
48            return False
49        return default
50
51    mass.config.get_base_player_config.return_value = MagicMock(
52        name=None, default_name="Test UGP", get_value=_config_value
53    )
54    mass.config.get_raw_player_config_value = MagicMock(return_value=None)
55
56    mass.streams = MagicMock()
57    mass.streams.base_url = "http://test"
58    mass.streams.register_dynamic_route = MagicMock(return_value=lambda: None)
59    mass.streams.get_stream = MagicMock(return_value=MagicMock())
60
61    # mass.create_task must return a real asyncio.Task so callers that do
62    # `await asyncio.wait([task])` (TaskManager.__aexit__) don't hang on a
63    # MagicMock that pretends to be a task. We wrap coroutines into real
64    # tasks; the underlying mocked methods (_handle_play_media, etc.)
65    # resolve immediately, so the tasks complete fast.
66    def _create_task(target: Any, *_args: Any, **_kwargs: Any) -> asyncio.Task[Any]:
67        if asyncio.iscoroutine(target):
68            return asyncio.ensure_future(target)
69
70        # not a coroutine - return a finished task so callers can still
71        # treat it as awaitable / cancellable.
72        async def _noop() -> None:
73            return None
74
75        return asyncio.ensure_future(_noop())
76
77    mass.create_task = MagicMock(side_effect=_create_task)
78    mass.cache = MagicMock()
79    return mass
80
81
82def _make_ugp(mass: MagicMock, player_id: str = "ugp_test") -> UniversalGroupPlayer:
83    """Create a UniversalGroupPlayer with a mock provider."""
84    provider = MagicMock()
85    provider.domain = "universal_group"
86    provider.instance_id = "universal_group_test"
87    provider.name = "Universal Group"
88    provider.mass = mass
89
90    return UniversalGroupPlayer(provider, player_id)
91
92
93def _make_mock_player(
94    player_id: str,
95    powered: bool = True,
96    available: bool = True,
97    enabled: bool = True,
98    active_group: str | None = None,
99    synced_to: str | None = None,
100    playback_state: PlaybackState = PlaybackState.IDLE,
101) -> MagicMock:
102    """Create a mock child player."""
103    player = MagicMock()
104    player.player_id = player_id
105    player.display_name = player_id
106    player.available = available
107    player.enabled = enabled
108    player.powered = powered
109    player.power_control = "native"
110    player.synced_to = synced_to
111    player.playback_state = playback_state
112    player.active_source = None
113    player.state = MagicMock()
114    player.state.available = available
115    player.state.enabled = enabled
116    player.state.powered = powered
117    player.state.power_control = "native"
118    player.state.active_group = active_group
119    player.state.synced_to = synced_to
120    player.state.playback_state = playback_state
121    player.state.supported_features = {PlayerFeature.PLAY_MEDIA}
122    player.state.elapsed_time = 0
123    player.state.elapsed_time_last_updated = None
124    player.ungroup = AsyncMock()
125    return player
126
127
128class TestIsActiveSession:
129    """is_active_session governs whether children report active_group to this UGP."""
130
131    def test_dormant_group_is_not_active(self) -> None:
132        """A freshly-initialized UGP has no stream and no grace task."""
133        mass = _make_mock_mass()
134        ugp = _make_ugp(mass)
135        assert ugp.is_active_session is False
136
137    def test_group_with_live_stream_is_active(self) -> None:
138        """A UGP with a live (not-done) stream is considered active."""
139        mass = _make_mock_mass()
140        ugp = _make_ugp(mass)
141        ugp.stream = MagicMock()
142        ugp.stream.done = False
143        assert ugp.is_active_session is True
144
145    def test_group_with_done_stream_is_not_active(self) -> None:
146        """If the stream has already ended, the session is not considered active."""
147        mass = _make_mock_mass()
148        ugp = _make_ugp(mass)
149        ugp.stream = MagicMock()
150        ugp.stream.done = True
151        assert ugp.is_active_session is False
152
153    def test_group_in_grace_window_is_active(self) -> None:
154        """While the idle-grace task is pending, the group still claims captured members."""
155        mass = _make_mock_mass()
156        ugp = _make_ugp(mass)
157        ugp.stream = None
158        ugp._idle_grace_task = MagicMock()
159        assert ugp.is_active_session is True
160
161
162class TestPowerlessLifecycle:
163    """Form-on-play / dissolve-on-stop without explicit power control."""
164
165    @pytest.mark.asyncio
166    async def test_play_media_captures_members(self) -> None:
167        """play_media should set up a stream and forward to members without a power command."""
168        mass = _make_mock_mass()
169        ugp = _make_ugp(mass)
170        ugp._attr_static_group_members = ["m1", "m2"]
171        member1 = _make_mock_player("m1")
172        member2 = _make_mock_player("m2")
173        mass.players.get_player = MagicMock(
174            side_effect=lambda pid, *_args, **_kwargs: {"m1": member1, "m2": member2}.get(pid)
175        )
176        # iter_group_members must yield the captured members after _capture_members runs
177        mass.players.iter_group_members.side_effect = lambda *_args, **_kwargs: iter(
178            [member1, member2]
179        )
180
181        with patch.object(ugp, "update_state"):
182            await ugp.play_media(
183                MagicMock(
184                    uri="track://x",
185                    source_id="src",
186                    custom_data={"session_id": "session-1"},
187                )
188            )
189
190        assert ugp.stream is not None
191        assert mass.players._handle_play_media.await_count == 2
192        for call in mass.players._handle_play_media.await_args_list:
193            assert call.args[1].custom_data["session_id"] == "session-1"
194        # power command was NOT used to capture the members
195        mass.players.cmd_power.assert_not_awaited()
196
197    @pytest.mark.asyncio
198    async def test_stop_releases_members_when_powerless(self) -> None:
199        """stop() on a powerless UGP should tear down the stream and release members."""
200        mass = _make_mock_mass()
201        ugp = _make_ugp(mass)
202        stream = MagicMock()
203        stream.done = False
204        stream.stop = AsyncMock()
205        ugp.stream = stream
206        ugp._attr_group_members = ["m1"]
207        ugp._attr_static_group_members = ["m1"]
208        member = _make_mock_player("m1")
209        mass.players.iter_group_members.side_effect = lambda *_args, **_kwargs: iter([member])
210        # _attr_powered defaults to None (no power control)
211        assert ugp._attr_powered is None
212
213        with patch.object(ugp, "update_state"):
214            await ugp.stop()
215
216        stream.stop.assert_awaited()
217        # session is no longer active
218        assert ugp.stream is None
219        # mypy narrows ugp.stream to None above; the property check below is
220        # technically static after that narrowing, hence the ignore.
221        assert ugp.is_active_session is False  # type: ignore[unreachable]
222
223    @pytest.mark.asyncio
224    async def test_stop_preserves_session_when_pinned(self) -> None:
225        """stop() on a UGP pinned with Fake power should NOT reset _attr_group_members."""
226        mass = _make_mock_mass()
227        ugp = _make_ugp(mass)
228        stream = MagicMock()
229        stream.done = False
230        stream.stop = AsyncMock()
231        ugp.stream = stream
232        ugp._attr_group_members = ["m1", "extra"]
233        ugp._attr_static_group_members = ["m1"]
234        ugp._attr_powered = True  # user explicitly pinned via Fake control
235        member = _make_mock_player("m1")
236        mass.players.iter_group_members.side_effect = lambda *_args, **_kwargs: iter([member])
237
238        with patch.object(ugp, "update_state"):
239            await ugp.stop()
240
241        # we keep the dynamic adds because the user is pinning this group
242        assert "extra" in ugp._attr_group_members
243
244
245class TestIdleGraceTimer:
246    """The idle-grace timer absorbs natural PLAYING→IDLE transitions for UGP."""
247
248    def test_grace_scheduled_on_playing_to_idle(self) -> None:
249        """_set_attributes schedules a grace task on transition while stream is still live."""
250        mass = _make_mock_mass()
251        sentinel = MagicMock()
252        mass.create_task = MagicMock(return_value=sentinel)
253        ugp = _make_ugp(mass)
254        # simulate a live stream and a prior PLAYING state
255        ugp.stream = MagicMock()
256        ugp.stream.done = False
257        ugp._attr_playback_state = PlaybackState.PLAYING
258        # no active children → state derives IDLE
259        mass.players.iter_group_members.side_effect = lambda *_args, **_kwargs: iter([])
260
261        with patch.object(ugp, "update_state"):
262            ugp._set_attributes()
263
264        assert ugp._idle_grace_task is sentinel
265        mass.create_task.assert_called_once()
266
267    def test_grace_not_scheduled_when_pinned(self) -> None:
268        """No grace task when the user has pinned the UGP with Fake power on."""
269        mass = _make_mock_mass()
270        mass.create_task = MagicMock()
271        ugp = _make_ugp(mass)
272        ugp.stream = MagicMock()
273        ugp.stream.done = False
274        ugp._attr_playback_state = PlaybackState.PLAYING
275        ugp._attr_powered = True  # explicit pin
276        mass.players.iter_group_members.side_effect = lambda *_args, **_kwargs: iter([])
277
278        with patch.object(ugp, "update_state"):
279            ugp._set_attributes()
280
281        assert ugp._idle_grace_task is None
282        mass.create_task.assert_not_called()
283
284    def test_grace_cancelled_on_resume(self) -> None:
285        """A pending grace task is cancelled when a member resumes playback."""
286        mass = _make_mock_mass()
287        mass.create_task = MagicMock()
288        ugp = _make_ugp(mass)
289        ugp.stream = MagicMock()
290        ugp.stream.done = False
291        prior = MagicMock()
292        prior.done.return_value = False
293        ugp._idle_grace_task = prior
294        ugp._attr_playback_state = PlaybackState.IDLE
295        playing = _make_mock_player(
296            "m1", playback_state=PlaybackState.PLAYING, active_group="ugp_test"
297        )
298        mass.players.iter_group_members.side_effect = lambda *_args, **_kwargs: iter([playing])
299
300        with patch.object(ugp, "update_state"):
301            ugp._set_attributes()
302
303        prior.cancel.assert_called_once()
304        assert ugp._idle_grace_task is None
305
306
307class TestFakePowerLifecycle:
308    """When the user assigns Fake power control, power(True/False) drives form/dissolve."""
309
310    @pytest.mark.asyncio
311    async def test_power_on_captures_static_members(self) -> None:
312        """power(True) should populate the effective group_members from the configured static set."""
313        mass = _make_mock_mass()
314        ugp = _make_ugp(mass)
315        ugp._attr_static_group_members = ["m1", "m2"]
316        member1 = _make_mock_player("m1")
317        member2 = _make_mock_player("m2")
318        mass.players.get_player = MagicMock(
319            side_effect=lambda pid, *_args, **_kwargs: {"m1": member1, "m2": member2}.get(pid)
320        )
321        mass.players.iter_group_members.side_effect = lambda *_args, **_kwargs: iter(
322            [member1, member2]
323        )
324
325        with patch.object(ugp, "update_state"):
326            await ugp.power(True)
327
328        assert ugp._attr_group_members == ["m1", "m2"]
329        assert ugp._attr_powered is True
330
331    @pytest.mark.asyncio
332    async def test_power_off_releases_and_resets_members(self) -> None:
333        """power(False) should reset group_members to static and clear powered."""
334        mass = _make_mock_mass()
335        ugp = _make_ugp(mass)
336        ugp._attr_static_group_members = ["m1"]
337        ugp._attr_group_members = ["m1", "extra"]
338        ugp._attr_powered = True
339        # quietly powered-on member that should get cmd_power(False) on dissolve
340        member = _make_mock_player("m1", powered=True)
341        mass.players.iter_group_members.side_effect = lambda *_args, **_kwargs: iter([member])
342
343        with patch.object(ugp, "update_state"):
344            await ugp.power(False)
345
346        assert ugp._attr_group_members == ["m1"]
347        assert ugp._attr_powered is False
348
349
350class TestSupportedFeaturesPower:
351    """POWER feature is only advertised when the user opts in via Fake control."""
352
353    @pytest.mark.asyncio
354    async def test_power_not_in_features_by_default(self) -> None:
355        """No power_control config → no POWER feature."""
356        mass = _make_mock_mass()
357        # explicitly ensure power_control config is empty
358        mass.config.get_raw_player_config_value = MagicMock(return_value=None)
359        ugp = _make_ugp(mass)
360        # on_config_updated is what (re)evaluates the feature set
361        await ugp.on_config_updated()
362        assert PlayerFeature.POWER not in ugp.supported_features
363
364    @pytest.mark.asyncio
365    async def test_power_advertised_when_fake_control_assigned(self) -> None:
366        """User assigns Fake power control → POWER feature shows up."""
367        mass = _make_mock_mass()
368
369        def _get_raw(_player_id: str, key: str, default: object = None) -> object:
370            if key == "power_control":
371                return PLAYER_CONTROL_FAKE
372            return default
373
374        mass.config.get_raw_player_config_value = MagicMock(side_effect=_get_raw)
375        ugp = _make_ugp(mass)
376        await ugp.on_config_updated()
377        assert PlayerFeature.POWER in ugp.supported_features
378
379
380class TestSupportedFeaturesSetMembers:
381    """SET_MEMBERS tracks the dynamic-members config option."""
382
383    @staticmethod
384    def _make_mass(dynamic: bool) -> MagicMock:
385        """Build a mock mass whose group is (or isn't) a dynamic group."""
386        mass = _make_mock_mass()
387
388        def _config_value(key: str, default: object = None) -> object:
389            if key == "group_members":
390                return []
391            if key == "dynamic_members":
392                return dynamic
393            return default
394
395        mass.config.get_base_player_config.return_value = MagicMock(
396            name=None, default_name="Test UGP", get_value=_config_value
397        )
398        return mass
399
400    def test_dynamic_group_advertises_set_members(self) -> None:
401        """A dynamic group lets the user change its members."""
402        ugp = _make_ugp(self._make_mass(dynamic=True))
403        assert PlayerFeature.SET_MEMBERS in ugp.supported_features
404
405    def test_static_group_does_not_advertise_set_members(self) -> None:
406        """A static group's members are fixed by its config."""
407        ugp = _make_ugp(self._make_mass(dynamic=False))
408        assert PlayerFeature.SET_MEMBERS not in ugp.supported_features
409
410
411class TestSupportedFeaturesFromMembers:
412    """Volume and mute are inherited from the members those commands are fanned out to."""
413
414    @staticmethod
415    def _attach_members(mass: MagicMock, *members: MagicMock) -> None:
416        """Register the given mock members on the mock player controller."""
417        by_id = {member.player_id: member for member in members}
418        mass.players.get_player = MagicMock(
419            side_effect=lambda pid, *_args, **_kwargs: by_id.get(pid)
420        )
421
422    def test_volume_advertised_when_a_member_supports_it(self) -> None:
423        """A volume-capable member gives the group a volume capability."""
424        mass = _make_mock_mass()
425        ugp = _make_ugp(mass)
426        member = _make_mock_player("m1")
427        member.state.supported_features = {PlayerFeature.PLAY_MEDIA, PlayerFeature.VOLUME_SET}
428        self._attach_members(mass, member)
429        ugp._attr_group_members = ["m1"]
430
431        assert PlayerFeature.VOLUME_SET in ugp.supported_features
432
433    def test_volume_not_advertised_without_capable_members(self) -> None:
434        """An empty group has nothing to send a volume command to."""
435        mass = _make_mock_mass()
436        ugp = _make_ugp(mass)
437
438        assert ugp._attr_group_members == []
439        assert PlayerFeature.VOLUME_SET not in ugp.supported_features
440
441    def test_mute_not_advertised_without_capable_members(self) -> None:
442        """Members that can't be muted leave the group without a mute capability."""
443        mass = _make_mock_mass()
444        ugp = _make_ugp(mass)
445        member = _make_mock_player("m1")
446        self._attach_members(mass, member)
447        ugp._attr_group_members = ["m1"]
448
449        assert PlayerFeature.VOLUME_MUTE not in ugp.supported_features
450
451    def test_mute_advertised_when_a_member_supports_it(self) -> None:
452        """A single mute-capable member is enough to advertise mute on the group."""
453        mass = _make_mock_mass()
454        ugp = _make_ugp(mass)
455        plain_member = _make_mock_player("m1")
456        mute_member = _make_mock_player("m2")
457        mute_member.state.supported_features = {
458            PlayerFeature.PLAY_MEDIA,
459            PlayerFeature.VOLUME_MUTE,
460        }
461        self._attach_members(mass, plain_member, mute_member)
462        ugp._attr_group_members = ["m1", "m2"]
463
464        assert PlayerFeature.VOLUME_MUTE in ugp.supported_features
465
466    def test_unavailable_members_do_not_contribute(self) -> None:
467        """An offline member's capabilities are not advertised by the group."""
468        mass = _make_mock_mass()
469        ugp = _make_ugp(mass)
470        member = _make_mock_player("m1", available=False)
471        member.state.supported_features = {PlayerFeature.PLAY_MEDIA, PlayerFeature.VOLUME_MUTE}
472        self._attach_members(mass, member)
473        ugp._attr_group_members = ["m1"]
474
475        assert PlayerFeature.VOLUME_MUTE not in ugp.supported_features
476
477    def test_dormant_group_resolves_a_native_mute_control(self) -> None:
478        """
479        An idle group resolves a mute control, which is what HA gates its mute button on.
480
481        Without an inherited VOLUME_MUTE this lands on PLAYER_CONTROL_NONE and the
482        capability is dropped from the player state again.
483        """
484        mass = _make_mock_mass()
485        ugp = _make_ugp(mass)
486        member = _make_mock_player("m1")
487        member.state.supported_features = {PlayerFeature.PLAY_MEDIA, PlayerFeature.VOLUME_MUTE}
488        self._attach_members(mass, member)
489        ugp._attr_static_group_members = ["m1"]
490        ugp._attr_group_members = ["m1"]
491        # a member update reaches the group as trigger_player_update -> update_state,
492        # which is what drops the cached mute_control so it can be re-resolved
493        ugp.update_state()
494
495        assert not ugp.is_active_session
496        assert ugp.mute_control == PLAYER_CONTROL_NATIVE
497
498    def test_mute_control_stays_none_without_capable_members(self) -> None:
499        """No member can be muted → no mute control to hand to HA."""
500        mass = _make_mock_mass()
501        ugp = _make_ugp(mass)
502        member = _make_mock_player("m1")
503        self._attach_members(mass, member)
504        ugp._attr_group_members = ["m1"]
505        ugp.update_state()
506
507        assert ugp.mute_control == PLAYER_CONTROL_NONE
508