/
/
/
1"""Common test helpers for Music Assistant tests."""
2
3import asyncio
4import contextlib
5import pathlib
6from collections.abc import AsyncGenerator
7
8import aiofiles.os
9from music_assistant_models.enums import EventType
10from music_assistant_models.event import MassEvent
11
12from music_assistant.mass import MusicAssistant
13
14
15def _get_fixture_folder(provider: str | None = None) -> pathlib.Path:
16 tests_base = pathlib.Path(__file__).parent
17 if provider:
18 return tests_base / "providers" / provider / "fixtures"
19 return tests_base / "fixtures"
20
21
22async def get_fixtures_dir(
23 subdir: str, provider: str | None = None
24) -> AsyncGenerator[tuple[str, bytes], None]:
25 """Yield the contents of every fixture in a fixtures folder."""
26 dir_path = _get_fixture_folder(provider) / subdir
27 for file in await aiofiles.os.listdir(dir_path):
28 async with aiofiles.open(dir_path / file, "rb") as fp:
29 yield (file, await fp.read())
30
31
32@contextlib.asynccontextmanager
33async def wait_for_sync_completion(mass: MusicAssistant) -> AsyncGenerator[None, None]:
34 """Wait for a sync to finish."""
35 flag = asyncio.Event()
36
37 def _event(event: MassEvent) -> None:
38 if not event.data:
39 flag.set()
40
41 release_cb = mass.subscribe(_event, EventType.SYNC_TASKS_UPDATED)
42
43 try:
44 yield
45 finally:
46 await flag.wait()
47 release_cb()
48