/
/
/
1"""
2End-to-end demo: a sync group of fake players mirrors the leader's media.
3
4Exercises the full stack with no hardware/network: form a sync group of demo
5players, play a fake `test`-provider track, and assert the group's current media
6propagates to every member.
7"""
8
9from __future__ import annotations
10
11from functools import partial
12
13import pytest
14
15from music_assistant.mass import MusicAssistant
16from music_assistant.models.player import Player
17
18from .conftest import demo_players, group_players, play_test_track, wait_for
19
20
21def _mirrors_uri(player: Player, uri: str) -> bool:
22 """Return True if the player's current media matches the given uri."""
23 current = player.state.current_media
24 return current is not None and current.uri == uri
25
26
27@pytest.mark.asyncio
28async def test_demo_group_mirrors_current_media(e2e_mass: MusicAssistant) -> None:
29 """A grouped set of demo players all report the sync leader's current media."""
30 # hermetic boot: only the fake demo players exist (no real LAN devices discovered)
31 assert sorted(p.player_id for p in e2e_mass.players) == ["demo_0", "demo_1", "demo_2"]
32
33 players = demo_players(e2e_mass)
34 assert len(players) >= 3
35 leader, *members = players[:3]
36
37 # 1. form the sync group
38 await group_players(e2e_mass, leader, members)
39 assert await wait_for(lambda: len(leader.state.group_members) == 3), (
40 f"group not formed: {leader.state.group_members}"
41 )
42 assert leader.state.group_members[0] == leader.player_id
43 for member in members:
44 assert member.state.synced_to == leader.player_id
45
46 # 2. play a fake track (carries artwork) on the group
47 await play_test_track(e2e_mass, leader.player_id)
48 assert await wait_for(lambda: leader.state.current_media is not None), (
49 "leader current_media never populated"
50 )
51 leader_media = leader.state.current_media
52 assert leader_media is not None
53 assert leader_media.image_url
54
55 # 3. every member mirrors the leader's media end-to-end
56 for member in members:
57 assert await wait_for(partial(_mirrors_uri, member, leader_media.uri)), (
58 f"{member.player_id} did not mirror the leader's media"
59 )
60 member_media = member.state.current_media
61 assert member_media is not None
62 assert member_media.image_url == leader_media.image_url
63