/
/
/
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. The `test` music
102 provider and three grouped-capable demo players are configured and ready.
103 """
104 storage_path = tmp_path / "data"
105 cache_path = tmp_path / "cache"
106 storage_path.mkdir(parents=True)
107 cache_path.mkdir(parents=True)
108 logging.getLogger("aiosqlite").level = logging.INFO
109
110 mass_instance = MusicAssistant(str(storage_path), str(cache_path))
111 # load the `_`-prefixed demo player provider (gated on dev mode)
112 mass_instance.dev_mode = True
113
114 with (
115 use_ephemeral_server_ports(),
116 patch(
117 "music_assistant.controllers.discovery.controller.AsyncZeroconf",
118 return_value=_create_mock_zeroconf(),
119 ),
120 patch(
121 "music_assistant.controllers.discovery.controller.AsyncServiceBrowser",
122 return_value=NonCallableMagicMock(),
123 ),
124 patch(
125 "music_assistant.controllers.streams.controller.check_ffmpeg_version",
126 new=AsyncMock(),
127 ),
128 # hermetic: no real SSDP search
129 patch(
130 "music_assistant.controllers.discovery.controller.async_upnp_search",
131 new=AsyncMock(),
132 ),
133 # hermetic: no auto-loaded device providers and no host-audio bridging
134 suppress_auto_loaded_providers(),
135 # no library sync starting on its own inside a running test
136 suppress_initial_library_sync(),
137 ):
138 try:
139 await mass_instance.start()
140 # configure the fake music + player providers
141 await mass_instance.config._create_provider_instance("test", {})
142 await mass_instance.config._create_provider_instance(
143 "_demo_player_provider", {"number_of_players": NUM_DEMO_PLAYERS}
144 )
145 await wait_for(lambda: len(demo_players(mass_instance)) >= NUM_DEMO_PLAYERS)
146 await wait_for_boot_to_settle(mass_instance)
147 yield mass_instance
148 finally:
149 # also stop after a failed boot, or the half-started server's open database
150 # connections keep their non-daemon threads alive and hang the interpreter
151 # at exit (see the same note in tests/conftest.py)
152 await mass_instance.stop()
153