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                    queue_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].queue_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 TestMemberPositionPropagation:
308    """The group mirrors the playback position reported by its active member."""
309
310    def test_position_is_adopted(self) -> None:
311        """A member's position anchor becomes the group's position anchor."""
312        ugp = self._ugp_with_member(self._member(42.0, 2000.0))
313
314        with patch.object(ugp, "update_state"):
315            ugp._set_attributes()
316
317        assert ugp._attr_elapsed_time == 42.0
318        assert ugp._attr_elapsed_time_last_updated == 2000.0
319
320    def test_zero_position_is_adopted(self) -> None:
321        """Position 0 is a real position and replaces the group's own anchor."""
322        ugp = self._ugp_with_member(self._member(0.0, 2000.0))
323
324        with patch.object(ugp, "update_state"):
325            ugp._set_attributes()
326
327        assert ugp._attr_elapsed_time == 0.0
328        assert ugp._attr_elapsed_time_last_updated == 2000.0
329
330    def test_position_without_timestamp_is_ignored(self) -> None:
331        """A position without its timestamp is unusable, so the group keeps its anchor."""
332        ugp = self._ugp_with_member(self._member(42.0, None))
333
334        with patch.object(ugp, "update_state"):
335            ugp._set_attributes()
336
337        assert ugp._attr_elapsed_time == 12.0
338        assert ugp._attr_elapsed_time_last_updated == 1000.0
339
340    def test_unknown_position_is_ignored(self) -> None:
341        """A member that reports no position leaves the group's own anchor in place."""
342        ugp = self._ugp_with_member(self._member(None, 2000.0))
343
344        with patch.object(ugp, "update_state"):
345            ugp._set_attributes()
346
347        assert ugp._attr_elapsed_time == 12.0
348        assert ugp._attr_elapsed_time_last_updated == 1000.0
349
350    @staticmethod
351    def _member(elapsed_time: float | None, last_updated: float | None) -> MagicMock:
352        """Create an active member reporting the given position anchor."""
353        member = _make_mock_player(
354            "m1", playback_state=PlaybackState.PLAYING, active_group="ugp_test"
355        )
356        member.state.elapsed_time = elapsed_time
357        member.state.elapsed_time_last_updated = last_updated
358        return member
359
360    @staticmethod
361    def _ugp_with_member(member: MagicMock) -> UniversalGroupPlayer:
362        """Create a playing UGP holding a stale anchor and deriving state from the member."""
363        mass = _make_mock_mass()
364        ugp = _make_ugp(mass)
365        ugp.stream = MagicMock()
366        ugp.stream.done = False
367        ugp._attr_playback_state = PlaybackState.PLAYING
368        ugp._attr_elapsed_time = 12.0
369        ugp._attr_elapsed_time_last_updated = 1000.0
370        mass.players.iter_group_members.side_effect = lambda *_args, **_kwargs: iter([member])
371        return ugp
372
373
374class TestFakePowerLifecycle:
375    """When the user assigns Fake power control, power(True/False) drives form/dissolve."""
376
377    @pytest.mark.asyncio
378    async def test_power_on_captures_static_members(self) -> None:
379        """power(True) should populate the effective group_members from the configured static set."""
380        mass = _make_mock_mass()
381        ugp = _make_ugp(mass)
382        ugp._attr_static_group_members = ["m1", "m2"]
383        member1 = _make_mock_player("m1")
384        member2 = _make_mock_player("m2")
385        mass.players.get_player = MagicMock(
386            side_effect=lambda pid, *_args, **_kwargs: {"m1": member1, "m2": member2}.get(pid)
387        )
388        mass.players.iter_group_members.side_effect = lambda *_args, **_kwargs: iter(
389            [member1, member2]
390        )
391
392        with patch.object(ugp, "update_state"):
393            await ugp.power(True)
394
395        assert ugp._attr_group_members == ["m1", "m2"]
396        assert ugp._attr_powered is True
397
398    @pytest.mark.asyncio
399    async def test_power_off_releases_and_resets_members(self) -> None:
400        """power(False) should reset group_members to static and clear powered."""
401        mass = _make_mock_mass()
402        ugp = _make_ugp(mass)
403        ugp._attr_static_group_members = ["m1"]
404        ugp._attr_group_members = ["m1", "extra"]
405        ugp._attr_powered = True
406        # quietly powered-on member that should get cmd_power(False) on dissolve
407        member = _make_mock_player("m1", powered=True)
408        mass.players.iter_group_members.side_effect = lambda *_args, **_kwargs: iter([member])
409
410        with patch.object(ugp, "update_state"):
411            await ugp.power(False)
412
413        assert ugp._attr_group_members == ["m1"]
414        assert ugp._attr_powered is False
415
416
417class TestSupportedFeaturesPower:
418    """POWER feature is only advertised when the user opts in via Fake control."""
419
420    @pytest.mark.asyncio
421    async def test_power_not_in_features_by_default(self) -> None:
422        """No power_control config → no POWER feature."""
423        mass = _make_mock_mass()
424        # explicitly ensure power_control config is empty
425        mass.config.get_raw_player_config_value = MagicMock(return_value=None)
426        ugp = _make_ugp(mass)
427        # on_config_updated is what (re)evaluates the feature set
428        await ugp.on_config_updated()
429        assert PlayerFeature.POWER not in ugp.supported_features
430
431    @pytest.mark.asyncio
432    async def test_power_advertised_when_fake_control_assigned(self) -> None:
433        """User assigns Fake power control → POWER feature shows up."""
434        mass = _make_mock_mass()
435
436        def _get_raw(_player_id: str, key: str, default: object = None) -> object:
437            if key == "power_control":
438                return PLAYER_CONTROL_FAKE
439            return default
440
441        mass.config.get_raw_player_config_value = MagicMock(side_effect=_get_raw)
442        ugp = _make_ugp(mass)
443        await ugp.on_config_updated()
444        assert PlayerFeature.POWER in ugp.supported_features
445
446
447class TestSupportedFeaturesSetMembers:
448    """SET_MEMBERS tracks the dynamic-members config option."""
449
450    @staticmethod
451    def _make_mass(dynamic: bool) -> MagicMock:
452        """Build a mock mass whose group is (or isn't) a dynamic group."""
453        mass = _make_mock_mass()
454
455        def _config_value(key: str, default: object = None) -> object:
456            if key == "group_members":
457                return []
458            if key == "dynamic_members":
459                return dynamic
460            return default
461
462        mass.config.get_base_player_config.return_value = MagicMock(
463            name=None, default_name="Test UGP", get_value=_config_value
464        )
465        return mass
466
467    def test_dynamic_group_advertises_set_members(self) -> None:
468        """A dynamic group lets the user change its members."""
469        ugp = _make_ugp(self._make_mass(dynamic=True))
470        assert PlayerFeature.SET_MEMBERS in ugp.supported_features
471
472    def test_static_group_does_not_advertise_set_members(self) -> None:
473        """A static group's members are fixed by its config."""
474        ugp = _make_ugp(self._make_mass(dynamic=False))
475        assert PlayerFeature.SET_MEMBERS not in ugp.supported_features
476
477
478class TestSupportedFeaturesFromMembers:
479    """Volume and mute are inherited from the members those commands are fanned out to."""
480
481    @staticmethod
482    def _attach_members(mass: MagicMock, *members: MagicMock) -> None:
483        """Register the given mock members on the mock player controller."""
484        by_id = {member.player_id: member for member in members}
485        mass.players.get_player = MagicMock(
486            side_effect=lambda pid, *_args, **_kwargs: by_id.get(pid)
487        )
488
489    def test_volume_advertised_when_a_member_supports_it(self) -> None:
490        """A volume-capable member gives the group a volume capability."""
491        mass = _make_mock_mass()
492        ugp = _make_ugp(mass)
493        member = _make_mock_player("m1")
494        member.state.supported_features = {PlayerFeature.PLAY_MEDIA, PlayerFeature.VOLUME_SET}
495        self._attach_members(mass, member)
496        ugp._attr_group_members = ["m1"]
497
498        assert PlayerFeature.VOLUME_SET in ugp.supported_features
499
500    def test_volume_not_advertised_without_capable_members(self) -> None:
501        """An empty group has nothing to send a volume command to."""
502        mass = _make_mock_mass()
503        ugp = _make_ugp(mass)
504
505        assert ugp._attr_group_members == []
506        assert PlayerFeature.VOLUME_SET not in ugp.supported_features
507
508    def test_mute_not_advertised_without_capable_members(self) -> None:
509        """Members that can't be muted leave the group without a mute capability."""
510        mass = _make_mock_mass()
511        ugp = _make_ugp(mass)
512        member = _make_mock_player("m1")
513        self._attach_members(mass, member)
514        ugp._attr_group_members = ["m1"]
515
516        assert PlayerFeature.VOLUME_MUTE not in ugp.supported_features
517
518    def test_mute_advertised_when_a_member_supports_it(self) -> None:
519        """A single mute-capable member is enough to advertise mute on the group."""
520        mass = _make_mock_mass()
521        ugp = _make_ugp(mass)
522        plain_member = _make_mock_player("m1")
523        mute_member = _make_mock_player("m2")
524        mute_member.state.supported_features = {
525            PlayerFeature.PLAY_MEDIA,
526            PlayerFeature.VOLUME_MUTE,
527        }
528        self._attach_members(mass, plain_member, mute_member)
529        ugp._attr_group_members = ["m1", "m2"]
530
531        assert PlayerFeature.VOLUME_MUTE in ugp.supported_features
532
533    def test_unavailable_members_do_not_contribute(self) -> None:
534        """An offline member's capabilities are not advertised by the group."""
535        mass = _make_mock_mass()
536        ugp = _make_ugp(mass)
537        member = _make_mock_player("m1", available=False)
538        member.state.supported_features = {PlayerFeature.PLAY_MEDIA, PlayerFeature.VOLUME_MUTE}
539        self._attach_members(mass, member)
540        ugp._attr_group_members = ["m1"]
541
542        assert PlayerFeature.VOLUME_MUTE not in ugp.supported_features
543
544    def test_dormant_group_resolves_a_native_mute_control(self) -> None:
545        """
546        An idle group resolves a mute control, which is what HA gates its mute button on.
547
548        Without an inherited VOLUME_MUTE this lands on PLAYER_CONTROL_NONE and the
549        capability is dropped from the player state again.
550        """
551        mass = _make_mock_mass()
552        ugp = _make_ugp(mass)
553        member = _make_mock_player("m1")
554        member.state.supported_features = {PlayerFeature.PLAY_MEDIA, PlayerFeature.VOLUME_MUTE}
555        self._attach_members(mass, member)
556        ugp._attr_static_group_members = ["m1"]
557        ugp._attr_group_members = ["m1"]
558        # a member update reaches the group as trigger_player_update -> update_state,
559        # which is what drops the cached mute_control so it can be re-resolved
560        ugp.update_state()
561
562        assert not ugp.is_active_session
563        assert ugp.mute_control == PLAYER_CONTROL_NATIVE
564
565    def test_mute_control_stays_none_without_capable_members(self) -> None:
566        """No member can be muted → no mute control to hand to HA."""
567        mass = _make_mock_mass()
568        ugp = _make_ugp(mass)
569        member = _make_mock_player("m1")
570        self._attach_members(mass, member)
571        ugp._attr_group_members = ["m1"]
572        ugp.update_state()
573
574        assert ugp.mute_control == PLAYER_CONTROL_NONE
575