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