/
/
/
1"""Tests for event dispatch and fan-out towards websocket clients."""
2
3from __future__ import annotations
4
5import asyncio
6from types import SimpleNamespace
7from typing import TYPE_CHECKING
8
9import pytest
10from aiohttp import web
11from aiohttp.test_utils import make_mocked_request
12from music_assistant_models.auth import User, UserRole
13from music_assistant_models.enums import EventType
14
15from music_assistant.controllers.webserver.controller import WebserverController
16from music_assistant.controllers.webserver.websocket_client import WebsocketClientHandler
17
18if TYPE_CHECKING:
19 from music_assistant.mass import MusicAssistant
20
21
22async def drain_event_callbacks() -> None:
23 """Yield to the event loop so pending event subscriber callbacks run."""
24 await asyncio.sleep(0)
25
26
27@pytest.fixture
28def webserver(mass_minimal: MusicAssistant) -> WebserverController:
29 """Return a WebserverController with stubbed serialization dependencies."""
30 # stub the controllers referenced by the serialization resolvers
31 # (mass_minimal does not set up metadata/translations/tasks)
32 mass_minimal.metadata = SimpleNamespace( # type: ignore[assignment]
33 compute_image_id=lambda provider, path: f"{provider}--{path}"
34 )
35 mass_minimal.translations = SimpleNamespace( # type: ignore[assignment]
36 get_translation=lambda _key, **_kwargs: None
37 )
38 webserver = WebserverController(mass_minimal)
39 mass_minimal.webserver = webserver
40 return webserver
41
42
43def create_ws_client(
44 webserver: WebserverController, username: str, role: UserRole = UserRole.ADMIN
45) -> WebsocketClientHandler:
46 """Create an authenticated + event-subscribed websocket client handler (no real socket)."""
47 request = make_mocked_request("GET", "/ws", app=web.Application())
48 client = WebsocketClientHandler(webserver, request)
49 client._authenticated_user = User(user_id=username, username=username, role=role)
50 client._subscribe_to_events()
51 return client
52
53
54def get_written_message(client: WebsocketClientHandler) -> str:
55 """Pop the next message queued for the client's writer."""
56 message = client._to_write.get_nowait()
57 assert message is not None
58 return message
59
60
61async def test_event_delivered_to_all_clients(
62 mass_minimal: MusicAssistant,
63 webserver: WebserverController,
64) -> None:
65 """An event signalled on the loop thread reaches every subscribed websocket client."""
66 client1 = create_ws_client(webserver, "user1")
67 client2 = create_ws_client(webserver, "user2")
68
69 mass_minimal.signal_event(EventType.PLAYER_UPDATED, "player1", {"name": "Test Player"})
70 await drain_event_callbacks()
71
72 msg1 = get_written_message(client1)
73 msg2 = get_written_message(client2)
74 assert msg1 == msg2
75 assert "player1" in msg1
76
77
78async def test_tasks_updated_payload_per_user(
79 mass_minimal: MusicAssistant,
80 webserver: WebserverController,
81) -> None:
82 """TASKS_UPDATED events carry a per-user payload."""
83 mass_minimal.tasks = SimpleNamespace( # type: ignore[assignment]
84 list_tasks_for_user=lambda user: [{"task_id": f"task-for-{user.username}"}]
85 )
86 client1 = create_ws_client(webserver, "user1")
87 client2 = create_ws_client(webserver, "user2")
88
89 mass_minimal.signal_event(EventType.TASKS_UPDATED)
90 await drain_event_callbacks()
91
92 msg1 = get_written_message(client1)
93 msg2 = get_written_message(client2)
94 assert "task-for-user1" in msg1
95 assert "task-for-user2" not in msg1
96 assert "task-for-user2" in msg2
97 assert "task-for-user1" not in msg2
98
99
100async def test_provider_event_delivered_to_guest_clients(
101 mass_minimal: MusicAssistant,
102 webserver: WebserverController,
103) -> None:
104 """PROVIDER_EVENT reaches all clients, including guest-scoped ones."""
105 guest = create_ws_client(webserver, "guest1", role=UserRole.GUEST)
106 admin = create_ws_client(webserver, "admin1")
107
108 mass_minimal.signal_event(EventType.PROVIDER_EVENT, "music_quiz--abcd/game_state", {"round": 1})
109 await drain_event_callbacks()
110
111 msg_guest = get_written_message(guest)
112 msg_admin = get_written_message(admin)
113 assert msg_guest == msg_admin
114 assert "music_quiz--abcd/game_state" in msg_guest
115