/
/
/
1"""Common test helpers for Music Assistant tests."""
2
3import asyncio
4import contextlib
5import logging
6import pathlib
7from collections.abc import AsyncGenerator
8from unittest.mock import MagicMock
9
10import aiofiles.os
11from music_assistant_models.enums import EventType, IdentifierType, PlayerFeature, PlayerType
12from music_assistant_models.event import MassEvent
13from music_assistant_models.player import DeviceInfo
14
15from music_assistant.mass import MusicAssistant
16from music_assistant.models.player import Player
17
18
19def _get_fixture_folder(provider: str | None = None) -> pathlib.Path:
20 tests_base = pathlib.Path(__file__).parent
21 if provider:
22 return tests_base / "providers" / provider / "fixtures"
23 return tests_base / "fixtures"
24
25
26async def get_fixtures_dir(
27 subdir: str, provider: str | None = None
28) -> AsyncGenerator[tuple[str, bytes], None]:
29 """Yield the contents of every fixture in a fixtures folder."""
30 dir_path = _get_fixture_folder(provider) / subdir
31 for file in await aiofiles.os.listdir(dir_path):
32 async with aiofiles.open(dir_path / file, "rb") as fp:
33 yield (file, await fp.read())
34
35
36@contextlib.asynccontextmanager
37async def wait_for_sync_completion(mass: MusicAssistant) -> AsyncGenerator[None, None]:
38 """Wait for a sync to finish."""
39 flag = asyncio.Event()
40
41 def _event(event: MassEvent) -> None:
42 if not event.data:
43 flag.set()
44
45 release_cb = mass.subscribe(_event, EventType.SYNC_TASKS_UPDATED)
46
47 try:
48 yield
49 finally:
50 await flag.wait()
51 release_cb()
52
53
54# Mock classes for testing
55
56
57def create_mock_config(name: str) -> MagicMock:
58 """Create a mock player config with the given name."""
59 config = MagicMock()
60 config.name = None # No custom name, use default
61 config.default_name = name
62 config.get_value = MagicMock(return_value="none") # Default to no power control
63 return config
64
65
66class MockProvider:
67 """Mock player provider for testing."""
68
69 def __init__(
70 self, domain: str, instance_id: str = "test_instance", mass: MagicMock | None = None
71 ) -> None:
72 """Initialize the mock provider."""
73 self.domain = domain
74 self.instance_id = instance_id
75 self.name = f"Mock {domain.title()}"
76 self.manifest = MagicMock()
77 self.manifest.name = f"Mock {domain} Provider"
78 self.mass = mass or MagicMock()
79 self.logger = logging.getLogger(f"test.{domain}")
80
81
82class MockPlayer(Player):
83 """Mock player for testing."""
84
85 def __init__(
86 self,
87 provider: MockProvider,
88 player_id: str,
89 name: str,
90 player_type: PlayerType = PlayerType.PLAYER,
91 identifiers: dict[IdentifierType, str] | None = None,
92 ) -> None:
93 """Initialize the mock player."""
94 # Set up the mock config before calling super().__init__
95 # because the parent __init__ accesses config
96 provider.mass.config.get_base_player_config.return_value = create_mock_config(name)
97
98 super().__init__(provider, player_id) # type: ignore[arg-type]
99 self._attr_name = name
100 # Set type as instance attribute (overrides class attribute)
101 self._attr_type = player_type
102 self._attr_available = True
103 self._attr_powered = True
104 self._attr_supported_features = {PlayerFeature.VOLUME_SET}
105 self._attr_can_group_with = set()
106 self._attr_group_members = []
107
108 # Set up device info with identifiers
109 self._attr_device_info = DeviceInfo(
110 model="Test Model",
111 manufacturer="Test Manufacturer",
112 )
113 if identifiers:
114 for conn_type, value in identifiers.items():
115 self._attr_device_info.add_identifier(conn_type, value)
116
117 # Clear cached properties after modifying attributes
118 self._cache.clear()
119
120 async def set_members(
121 self,
122 player_ids_to_add: list[str] | None = None,
123 player_ids_to_remove: list[str] | None = None,
124 ) -> None:
125 """Mock implementation of set_members."""
126 current_members = set(self._attr_group_members)
127
128 if player_ids_to_add:
129 current_members.update(player_ids_to_add)
130
131 if player_ids_to_remove:
132 current_members.difference_update(player_ids_to_remove)
133
134 # Always include self as first member if there are members
135 if current_members:
136 self._attr_group_members = [self.player_id] + [
137 pid for pid in current_members if pid != self.player_id
138 ]
139 else:
140 self._attr_group_members = []
141
142 # Clear cache to reflect changes
143 self._cache.clear()
144
145 async def stop(self) -> None:
146 """Stop playback - required abstract method."""
147
148
149class MockMass:
150 """Type hint for mocked MusicAssistant instance."""
151