/
/
/
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_server_info(mass: MusicAssistant) -> None:
36 """Test that the server info advertises the webserver details."""
37 server_info = mass.get_server_info()
38 assert server_info.name is not None
39 assert "Music Assistant" in server_info.name
40 assert server_info.internal_url == mass.webserver.base_url
41 assert server_info.external_url is None
42 assert server_info.has_remote_access is False
43
44
45async def test_events(mass: MusicAssistant) -> None:
46 """Test that events sent by signal_event can be seen by subscribe."""
47 filters: list[tuple[EventType | tuple[EventType, ...] | None, str | tuple[str, ...] | None]] = [
48 (None, None),
49 (EventType.UNKNOWN, None),
50 ((EventType.UNKNOWN, EventType.PLAYER_ADDED), None),
51 (None, "myid1"),
52 (None, ("myid1", "myid2")),
53 (EventType.UNKNOWN, "myid1"),
54 ]
55
56 for event_filter, id_filter in filters:
57 flag = False
58
59 def _ev(event: MassEvent) -> None:
60 assert event.event == EventType.UNKNOWN
61 assert event.data == "mytestdata"
62 assert event.object_id == "myid1"
63 nonlocal flag
64 flag = True
65
66 remove_cb = mass.subscribe(_ev, event_filter, id_filter)
67
68 mass.signal_event(EventType.UNKNOWN, "myid1", "mytestdata")
69 await asyncio.sleep(0)
70 assert flag is True
71
72 flag = False
73 remove_cb()
74 mass.signal_event(EventType.UNKNOWN)
75 await asyncio.sleep(0)
76 assert flag is False
77
78
79async def test_create_task_failure_logged(
80 mass_minimal: MusicAssistant, caplog: pytest.LogCaptureFixture
81) -> None:
82 """Test that a failed task without awaiter is logged, also without debug logging."""
83
84 async def _boom() -> None:
85 raise ValueError("boom")
86
87 with caplog.at_level(logging.INFO, logger=MASS_LOGGER_NAME):
88 mass_minimal.create_task(_boom)
89 # allow the task's done callback to run
90 await asyncio.sleep(0)
91
92 assert any(
93 "boom" in record.getMessage()
94 for record in caplog.records
95 if record.levelno == logging.WARNING
96 )
97
98
99async def test_create_task_replacement_stays_tracked(mass_minimal: MusicAssistant) -> None:
100 """Test that a finished task does not untrack a replacement with the same task_id."""
101 task_id = "test_replacement"
102 release = asyncio.Event()
103
104 async def _instant() -> None:
105 return
106
107 async def _blocked() -> None:
108 await release.wait()
109
110 # a task that never suspends is already finished when create_task returns,
111 # so its done callback is still queued while the caller continues
112 first = mass_minimal.create_task(_instant(), task_id=task_id)
113 assert first.done()
114 second = mass_minimal.create_task(_blocked(), task_id=task_id)
115 assert second is not first
116 assert mass_minimal._tracked_tasks[task_id] is second
117
118 # allow the first task's done callback to run
119 await asyncio.sleep(0)
120
121 assert mass_minimal._tracked_tasks.get(task_id) is second
122 # a later caller must join the in-flight task instead of starting a duplicate
123 assert mass_minimal.create_task(_blocked(), task_id=task_id) is second
124
125 release.set()
126 await second
127 assert task_id not in mass_minimal._tracked_tasks
128
129
130async def test_create_task_abort_existing_tracks_replacement(
131 mass_minimal: MusicAssistant,
132) -> None:
133 """Test that a task aborted in favour of a replacement does not untrack it."""
134 task_id = "test_abort_existing"
135
136 async def _blocked() -> None:
137 await asyncio.Event().wait()
138
139 first = mass_minimal.create_task(_blocked(), task_id=task_id)
140 second = mass_minimal.create_task(_blocked(), task_id=task_id, abort_existing=True)
141 assert second is not first
142
143 # the aborted task runs its done callback only once the cancellation is delivered
144 await asyncio.wait((first,))
145 assert first.cancelled()
146 assert mass_minimal._tracked_tasks.get(task_id) is second
147
148 second.cancel()
149 await asyncio.wait((second,))
150 assert task_id not in mass_minimal._tracked_tasks
151