music-assistant-server

1"""Tests for the core Music Assistant server object."""
2
3import asyncio
4
5from music_assistant_models.enums import EventType
6from music_assistant_models.event import MassEvent
7
8from music_assistant.mass import MusicAssistant
9
10
11async def test_start_and_stop_server(mass: MusicAssistant) -> None:
12    """Test that music assistant starts and stops cleanly."""
13    domains = frozenset(p.domain for p in mass.get_provider_manifests())
14    core_providers = frozenset(
15        ("builtin", "cache", "metadata", "music", "player_queues", "players", "streams")
16    )
17    assert domains.issuperset(core_providers)
18
19
20async def test_events(mass: MusicAssistant) -> None:
21    """Test that events sent by signal_event can be seen by subscribe."""
22    filters: list[tuple[EventType | tuple[EventType, ...] | None, str | tuple[str, ...] | None]] = [
23        (None, None),
24        (EventType.UNKNOWN, None),
25        ((EventType.UNKNOWN, EventType.AUTH_SESSION), None),
26        (None, "myid1"),
27        (None, ("myid1", "myid2")),
28        (EventType.UNKNOWN, "myid1"),
29    ]
30
31    for event_filter, id_filter in filters:
32        flag = False
33
34        def _ev(event: MassEvent) -> None:
35            assert event.event == EventType.UNKNOWN
36            assert event.data == "mytestdata"
37            assert event.object_id == "myid1"
38            nonlocal flag
39            flag = True
40
41        remove_cb = mass.subscribe(_ev, event_filter, id_filter)
42
43        mass.signal_event(EventType.UNKNOWN, "myid1", "mytestdata")
44        await asyncio.sleep(0)
45        assert flag is True
46
47        flag = False
48        remove_cb()
49        mass.signal_event(EventType.UNKNOWN)
50        await asyncio.sleep(0)
51        assert flag is False
52