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