/
/
1"""
2Hermetic end-to-end fixtures backed by the fake `test` + demo player providers.
3
4These boot a real MusicAssistant with only the fake music (`test`) and fake player
5(`_demo_player_provider`) providers, with all real network discovery disabled, so
6end-to-end behaviour (grouping, queue, current-media propagation) can be exercised
7without any hardware or LAN access.
8"""
9
10from __future__ import annotations
11
12import asyncio
13import logging
14import pathlib
15from collections.abc import AsyncGenerator, Callable
16from typing import cast
17from unittest.mock import AsyncMock, MagicMock, NonCallableMagicMock, patch
18
19import pytest
20from zeroconf.asyncio import AsyncZeroconf
21
22from music_assistant.mass import MusicAssistant
23from music_assistant.models.music_provider import MusicProvider
24from music_assistant.models.player import Player
25from tests.common import suppress_auto_loaded_providers, use_ephemeral_server_ports
26
27NUM_DEMO_PLAYERS = 3
28
29
30async def wait_for(predicate: Callable[[], bool], timeout: float = 25.0) -> bool:
31 """Poll ``predicate`` until it is truthy or ``timeout`` elapses; return the final value."""
32 elapsed = 0.0
33 while elapsed < timeout:
34 if predicate():
35 return True
36 await asyncio.sleep(0.25)
37 elapsed += 0.25
38 return predicate()
39
40
41def demo_players(mass: MusicAssistant) -> list[Player]:
42 """Return the registered demo players, sorted by id."""
43 return sorted(
44 (p for p in mass.players if p.player_id.startswith("demo_")),
45 key=lambda p: p.player_id,
46 )
47
48
49async def group_players(mass: MusicAssistant, leader: Player, members: list[Player]) -> None:
50 """
51 Form a sync group of demo players under ``leader``.
52
53 :param leader: The player that becomes the sync leader.
54 :param members: The players to join to the leader.
55 """
56 # refresh every player's state so can_group_with re-expands now that all players
57 # are registered (the cross-player refresh is otherwise debounced)
58 for player in [leader, *members]:
59 player.update_state(force_update=True)
60 await asyncio.sleep(0.5)
61 for member in members:
62 await mass.players.cmd_group(member.player_id, leader.player_id)
63 assert await wait_for(lambda: len(leader.state.group_members) >= 1 + len(members)), (
64 f"group did not form: {leader.state.group_members}"
65 )
66
67
68async def play_test_track(mass: MusicAssistant, queue_id: str, track_id: str = "0_0_0") -> None:
69 """Play a fake `test`-provider track (which carries artwork) on the given queue."""
70 test_prov = cast("MusicProvider", mass.get_provider("test"))
71 assert test_prov is not None
72 track = await test_prov.get_track(track_id)
73 await mass.player_queues.play_media(queue_id, track)
74
75
76def _create_mock_zeroconf() -> MagicMock:
77 """Create a mock AsyncZeroconf that prevents real mDNS network I/O."""
78 mock_zc = MagicMock(spec=AsyncZeroconf)
79 mock_inner_zc = NonCallableMagicMock()
80 mock_inner_zc.cache = NonCallableMagicMock()
81 mock_inner_zc.cache.cache = {} # empty cache - no discovered services
82 mock_zc.zeroconf = mock_inner_zc
83 mock_zc.async_register_service = AsyncMock()
84 mock_zc.async_update_service = AsyncMock()
85 mock_zc.async_unregister_service = AsyncMock()
86 mock_zc.async_close = AsyncMock()
87 return mock_zc
88
89
90@pytest.fixture
91async def e2e_mass(tmp_path: pathlib.Path) -> AsyncGenerator[MusicAssistant]:
92 """
93 Boot a hermetic MusicAssistant with only the fake `test` + demo player providers.
94
95 No real network discovery happens: mDNS (zeroconf) and SSDP are mocked, the
96 default device providers (dlna/sonos/...) are suppressed and so is local_audio,
97 which would otherwise register the host's sound devices as players. The `test`
98 music provider and three grouped-capable demo players are configured and ready.
99 """
100 storage_path = tmp_path / "data"
101 cache_path = tmp_path / "cache"
102 storage_path.mkdir(parents=True)
103 cache_path.mkdir(parents=True)
104 logging.getLogger("aiosqlite").level = logging.INFO
105
106 mass_instance = MusicAssistant(str(storage_path), str(cache_path))
107 # load the `_`-prefixed demo player provider (gated on dev mode)
108 mass_instance.dev_mode = True
109
110 with (
111 use_ephemeral_server_ports(),
112 patch(
113 "music_assistant.controllers.discovery.controller.AsyncZeroconf",
114 return_value=_create_mock_zeroconf(),
115 ),
116 patch(
117 "music_assistant.controllers.discovery.controller.AsyncServiceBrowser",
118 return_value=NonCallableMagicMock(),
119 ),
120 patch(
121 "music_assistant.controllers.streams.controller.check_ffmpeg_version",
122 new=AsyncMock(),
123 ),
124 # hermetic: no real SSDP search
125 patch(
126 "music_assistant.controllers.discovery.controller.async_upnp_search",
127 new=AsyncMock(),
128 ),
129 # hermetic: no auto-loaded device providers and no host-audio bridging
130 suppress_auto_loaded_providers(),
131 ):
132 try:
133 await mass_instance.start()
134 # configure the fake music + player providers
135 await mass_instance.config._create_provider_instance("test", {})
136 await mass_instance.config._create_provider_instance(
137 "_demo_player_provider", {"number_of_players": NUM_DEMO_PLAYERS}
138 )
139 await wait_for(lambda: len(demo_players(mass_instance)) >= NUM_DEMO_PLAYERS)
140 yield mass_instance
141 finally:
142 # also stop after a failed boot, or the half-started server's open database
143 # connections keep their non-daemon threads alive and hang the interpreter
144 # at exit (see the same note in tests/conftest.py)
145 await mass_instance.stop()
146