music-assistant-server

1"""Tests for the core Music Assistant server object."""
2
3import asyncio
4import logging
5from typing import TYPE_CHECKING
6
7from music_assistant_models.enums import EventType
8
9from music_assistant.constants import MASS_LOGGER_NAME
10from music_assistant.mass import MusicAssistant
11
12if TYPE_CHECKING:
13    import pytest
14    from music_assistant_models.event import MassEvent
15
16
17async def test_start_and_stop_server(mass: MusicAssistant) -> None:
18    """Test that music assistant starts and stops cleanly."""
19    domains = frozenset(p.domain for p in mass.get_provider_manifests())
20    core_providers = frozenset(
21        (
22            "builtin",
23            "cache",
24            "discovery",
25            "metadata",
26            "music",
27            "player_queues",
28            "players",
29            "streams",
30        )
31    )
32    assert domains.issuperset(core_providers)
33
34
35async def test_events(mass: MusicAssistant) -> None:
36    """Test that events sent by signal_event can be seen by subscribe."""
37    filters: list[tuple[EventType | tuple[EventType, ...] | None, str | tuple[str, ...] | None]] = [
38        (None, None),
39        (EventType.UNKNOWN, None),
40        ((EventType.UNKNOWN, EventType.PLAYER_ADDED), None),
41        (None, "myid1"),
42        (None, ("myid1", "myid2")),
43        (EventType.UNKNOWN, "myid1"),
44    ]
45
46    for event_filter, id_filter in filters:
47        flag = False
48
49        def _ev(event: MassEvent) -> None:
50            assert event.event == EventType.UNKNOWN
51            assert event.data == "mytestdata"
52            assert event.object_id == "myid1"
53            nonlocal flag
54            flag = True
55
56        remove_cb = mass.subscribe(_ev, event_filter, id_filter)
57
58        mass.signal_event(EventType.UNKNOWN, "myid1", "mytestdata")
59        await asyncio.sleep(0)
60        assert flag is True
61
62        flag = False
63        remove_cb()
64        mass.signal_event(EventType.UNKNOWN)
65        await asyncio.sleep(0)
66        assert flag is False
67
68
69async def test_create_task_failure_logged(
70    mass_minimal: MusicAssistant, caplog: pytest.LogCaptureFixture
71) -> None:
72    """Test that a failed task without awaiter is logged, also without debug logging."""
73
74    async def _boom() -> None:
75        raise ValueError("boom")
76
77    with caplog.at_level(logging.INFO, logger=MASS_LOGGER_NAME):
78        mass_minimal.create_task(_boom)
79        # allow the task's done callback to run
80        await asyncio.sleep(0)
81
82    assert any(
83        "boom" in record.getMessage()
84        for record in caplog.records
85        if record.levelno == logging.WARNING
86    )
87
88
89async def test_create_task_replacement_stays_tracked(mass_minimal: MusicAssistant) -> None:
90    """Test that a finished task does not untrack a replacement with the same task_id."""
91    task_id = "test_replacement"
92    release = asyncio.Event()
93
94    async def _instant() -> None:
95        return
96
97    async def _blocked() -> None:
98        await release.wait()
99
100    # a task that never suspends is already finished when create_task returns,
101    # so its done callback is still queued while the caller continues
102    first = mass_minimal.create_task(_instant(), task_id=task_id)
103    assert first.done()
104    second = mass_minimal.create_task(_blocked(), task_id=task_id)
105    assert second is not first
106    assert mass_minimal._tracked_tasks[task_id] is second
107
108    # allow the first task's done callback to run
109    await asyncio.sleep(0)
110
111    assert mass_minimal._tracked_tasks.get(task_id) is second
112    # a later caller must join the in-flight task instead of starting a duplicate
113    assert mass_minimal.create_task(_blocked(), task_id=task_id) is second
114
115    release.set()
116    await second
117    assert task_id not in mass_minimal._tracked_tasks
118
119
120async def test_create_task_abort_existing_tracks_replacement(
121    mass_minimal: MusicAssistant,
122) -> None:
123    """Test that a task aborted in favour of a replacement does not untrack it."""
124    task_id = "test_abort_existing"
125
126    async def _blocked() -> None:
127        await asyncio.Event().wait()
128
129    first = mass_minimal.create_task(_blocked(), task_id=task_id)
130    second = mass_minimal.create_task(_blocked(), task_id=task_id, abort_existing=True)
131    assert second is not first
132
133    # the aborted task runs its done callback only once the cancellation is delivered
134    await asyncio.wait((first,))
135    assert first.cancelled()
136    assert mass_minimal._tracked_tasks.get(task_id) is second
137
138    second.cancel()
139    await asyncio.wait((second,))
140    assert task_id not in mass_minimal._tracked_tasks
141