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