/
/
/
1"""Tests for the live announcements a client speaks into the webserver."""
2
3from __future__ import annotations
4
5import asyncio
6import logging
7import re
8from contextlib import aclosing
9from types import MethodType
10from typing import TYPE_CHECKING, Any, NamedTuple
11from unittest.mock import AsyncMock, MagicMock, patch
12
13import pytest
14from aiohttp import WSMsgType, web
15from aiohttp.test_utils import TestClient, TestServer, make_mocked_request
16from music_assistant_models.auth import User, UserRole
17from music_assistant_models.enums import ContentType
18from music_assistant_models.media_items import AudioFormat
19
20from music_assistant.constants import HOMEASSISTANT_SYSTEM_USER
21from music_assistant.controllers.players.controller import PlayerController
22from music_assistant.controllers.players.helpers import handle_player_command
23from music_assistant.controllers.streams import live_announcements
24from music_assistant.controllers.streams.live_announcements import (
25 LIVE_ANNOUNCEMENT_ROUTE,
26 MAX_CLOSE_REASON_BYTES,
27 LiveAnnouncementManager,
28 LiveAnnouncementSession,
29)
30from music_assistant.controllers.webserver.helpers.auth_middleware import get_current_user
31from music_assistant.helpers.audio import create_streaming_wave_header
32from tests.common import use_real_create_task
33
34if TYPE_CHECKING:
35 from collections.abc import AsyncGenerator
36
37 from aiohttp import ClientWebSocketResponse
38
39PLAYER_ID = "player1"
40BASE_URL = "http://ma.local:8097"
41VALID_TOKEN = "valid-token"
42SAMPLE_RATE = 16000
43LIVE_FORMAT = AudioFormat(
44 content_type=ContentType.PCM_S16LE,
45 sample_rate=SAMPLE_RATE,
46 bit_depth=16,
47 channels=1,
48)
49# the close code a client that may not (or cannot) announce is disconnected with
50REJECTED = 4001
51# upper bound on anything the server is expected to answer right away
52REPLY_TIMEOUT = 5
53
54
55class Harness(NamedTuple):
56 """A live announcement manager, its MusicAssistant stub and a client for its route."""
57
58 manager: LiveAnnouncementManager
59 mass: MagicMock
60 client: TestClient[web.Request, web.Application]
61 player: MagicMock
62
63
64class BlockedAnnouncement(NamedTuple):
65 """An announcement that reports its progress and plays for as long as the test wants."""
66
67 started: asyncio.Event
68 release: asyncio.Event
69 done: asyncio.Event
70
71
72@pytest.fixture(name="harness")
73async def harness_fixture() -> AsyncGenerator[Harness]:
74 """Yield a live announcement manager served on a real websocket route."""
75 mass = MagicMock()
76 mass.streams.base_url = BASE_URL
77 mass.webserver.auth.authenticate_with_token = AsyncMock(
78 return_value=User(user_id="user_1", username="listener", role=UserRole.USER)
79 )
80 player = _player()
81 mass.players.get_player = MagicMock(
82 side_effect=lambda player_id: player if player_id == PLAYER_ID else None
83 )
84 mass.players.play_announcement = AsyncMock()
85 # the announcement is dispatched as a task, which must actually run
86 use_real_create_task(mass)
87 manager = LiveAnnouncementManager(mass, logging.getLogger("test.streams.live_announcements"))
88 app = web.Application()
89 app.router.add_get(LIVE_ANNOUNCEMENT_ROUTE, manager.handle_ws)
90 client: TestClient[web.Request, web.Application] = TestClient(TestServer(app))
91 await client.start_server()
92 try:
93 yield Harness(manager, mass, client, player)
94 finally:
95 await client.close()
96
97
98@pytest.mark.asyncio
99async def test_read_replays_the_clip_from_the_start() -> None:
100 """
101 A reader gets the whole clip from the start and follows the audio still coming in.
102
103 The clip is replayed rather than consumed, so it can be served again whenever the
104 renderer comes back for it.
105 """
106 session = _session()
107 await session.write(b"first")
108
109 stream = session.read()
110 async with aclosing(stream):
111 assert await anext(stream) == b"first"
112 reader = asyncio.ensure_future(anext(stream))
113 await asyncio.sleep(0)
114 assert not reader.done()
115 await session.write(b"second")
116 assert await asyncio.wait_for(reader, timeout=REPLY_TIMEOUT) == b"second"
117
118 # the clip only ends when the client says it is done
119 end = asyncio.ensure_future(anext(stream))
120 await asyncio.sleep(0)
121 assert not end.done()
122 await session.finish()
123 with pytest.raises(StopAsyncIteration):
124 await asyncio.wait_for(end, timeout=REPLY_TIMEOUT)
125
126 # the whole clip is served again, without the audio that arrived after it ended
127 await session.write(b"late")
128 assert [chunk async for chunk in session.read()] == [b"first", b"second"]
129
130
131@pytest.mark.asyncio
132async def test_duration_follows_the_audio_that_arrived() -> None:
133 """The duration is what has been spoken so far, in seconds of PCM."""
134 session = _session()
135 assert session.duration == 0
136
137 await session.write(b"\x00" * LIVE_FORMAT.pcm_sample_size)
138 assert session.duration == 1.0
139
140 await session.write(b"\x00" * (LIVE_FORMAT.pcm_sample_size // 2))
141 assert session.duration == 1.5
142
143
144@pytest.mark.asyncio
145async def test_an_unknown_session_is_not_served() -> None:
146 """A session id that is not (or no longer) live has nothing to serve."""
147 manager = LiveAnnouncementManager(MagicMock(), logging.getLogger("test.streams.live"))
148 request, _ = _stream_request("ghost")
149
150 with pytest.raises(web.HTTPNotFound):
151 await manager.serve_stream(request)
152
153
154@pytest.mark.asyncio
155async def test_a_spoken_clip_is_announced_on_the_player(harness: Harness) -> None:
156 """A text frame completes the clip, which is then announced on the chosen player."""
157 ws = await _start_speaking(harness.client, pre_announce=True, volume_level=42)
158 assert await _reply(ws) == "started"
159
160 await ws.send_bytes(b"\x01\x02")
161 await ws.send_str('{"type": "stop"}')
162 assert await _reply(ws) == "finished"
163
164 harness.mass.players.play_announcement.assert_awaited_once()
165 call = harness.mass.players.play_announcement.call_args
166 assert call.args == (PLAYER_ID,)
167 assert call.kwargs["pre_announce"] is True
168 assert call.kwargs["volume_level"] == 42
169 assert re.fullmatch(rf"{re.escape(BASE_URL)}/live_announcement/[\w-]+\.wav", call.kwargs["url"])
170
171
172@pytest.mark.asyncio
173async def test_the_renderer_is_served_the_finished_clip(harness: Harness) -> None:
174 """The announcement url delivers a wave header followed by the clip that was spoken."""
175 served = _serve_the_clip_when_announced(harness)
176
177 ws = await _start_speaking(harness.client)
178 assert await _reply(ws) == "started"
179 await ws.send_bytes(b"\x01\x02")
180 await ws.send_bytes(b"\x03\x04")
181 await ws.send_str('{"type": "stop"}')
182 assert await _reply(ws) == "finished"
183
184 assert served == [[create_streaming_wave_header(LIVE_FORMAT), b"\x01\x02", b"\x03\x04"]]
185
186
187@pytest.mark.asyncio
188async def test_the_session_lives_as_long_as_the_announcement(harness: Harness) -> None:
189 """The audio stays available for as long as the player is playing it."""
190 blocked = _block_announcement(harness)
191
192 ws = await _start_speaking(harness.client)
193 assert await _reply(ws) == "started"
194 await ws.send_bytes(b"\x01\x02")
195 await ws.send_str('{"type": "stop"}')
196 await _wait(blocked.started)
197
198 # the player is on the clip, so it must stay available to be (re)fetched
199 session_id = _session_id(harness.mass.players.play_announcement.call_args.kwargs["url"])
200 assert harness.manager.active_sessions == 1
201 request, writer = _stream_request(session_id)
202 await harness.manager.serve_stream(request)
203 assert _written(writer) == [create_streaming_wave_header(LIVE_FORMAT), b"\x01\x02"]
204
205 blocked.release.set()
206 assert await _reply(ws) == "finished"
207
208 assert harness.manager.active_sessions == 0
209 late_request, _ = _stream_request(session_id)
210 with pytest.raises(web.HTTPNotFound):
211 await harness.manager.serve_stream(late_request)
212
213
214@pytest.mark.asyncio
215async def test_the_announcement_outlives_a_client_that_drops(harness: Harness) -> None:
216 """A client that disconnects mid-sentence still gets what it spoke played out."""
217 blocked = _block_announcement(harness)
218
219 ws = await _start_speaking(harness.client)
220 assert await _reply(ws) == "started"
221 await ws.send_bytes(b"\x01\x02")
222 await ws.close()
223
224 # the clip is announced although there is no longer anyone to report back to
225 await _wait(blocked.started)
226 assert harness.manager.active_sessions == 1
227 blocked.release.set()
228 await _wait(blocked.done)
229
230
231@pytest.mark.asyncio
232async def test_a_player_that_never_finishes_is_not_waited_on_forever(
233 harness: Harness, monkeypatch: pytest.MonkeyPatch
234) -> None:
235 """A player that never reports back ends as an error instead of holding the session."""
236 monkeypatch.setattr(live_announcements, "ANNOUNCEMENT_TIMEOUT", 0.1)
237 # never released: this stands in for a provider that hands off and never returns
238 _block_announcement(harness)
239
240 ws = await _start_speaking(harness.client)
241 assert await _reply(ws) == "started"
242 await ws.send_bytes(b"\x01\x02")
243 await ws.send_str('{"type": "stop"}')
244
245 assert await _reply(ws) == "error"
246 assert harness.manager.active_sessions == 0
247
248
249@pytest.mark.asyncio
250@pytest.mark.parametrize("player_filter", [[], [PLAYER_ID]], ids=["no filter", "player allowed"])
251async def test_the_announcement_runs_as_the_user_that_spoke_it(
252 harness: Harness, player_filter: list[str]
253) -> None:
254 """
255 The user of the connection reaches the task the announcement is dispatched on.
256
257 Player commands apply the user's restrictions from that context, so without it
258 every live announcement would run unrestricted.
259 """
260 user = User(
261 user_id="user_4", username="speaker", role=UserRole.USER, player_filter=player_filter
262 )
263 harness.mass.webserver.auth.authenticate_with_token = AsyncMock(return_value=user)
264 announced = _announce_through_player_commands(harness)
265
266 ws = await _start_speaking(harness.client)
267 assert await _reply(ws) == "started"
268 await ws.send_bytes(b"\x01\x02")
269 await ws.send_str('{"type": "stop"}')
270 assert await _reply(ws) == "finished"
271
272 assert announced == [(PLAYER_ID, user)]
273
274
275@pytest.mark.asyncio
276async def test_a_user_without_access_to_the_player_is_refused(harness: Harness) -> None:
277 """A player outside the user's filter is refused, as it is for any other command."""
278 harness.mass.webserver.auth.authenticate_with_token = AsyncMock(
279 return_value=User(
280 user_id="user_5", username="guest", role=UserRole.USER, player_filter=["other_player"]
281 )
282 )
283 announced = _announce_through_player_commands(harness)
284
285 ws = await _start_speaking(harness.client)
286 assert await _reply(ws) == "started"
287 await ws.send_bytes(b"\x01\x02")
288 await ws.send_str('{"type": "stop"}')
289 assert await _reply(ws) == "error"
290
291 assert announced == []
292
293
294@pytest.mark.asyncio
295async def test_a_client_that_goes_silent_ends_its_own_clip(
296 harness: Harness, monkeypatch: pytest.MonkeyPatch
297) -> None:
298 """Silence ends the clip on its own, so a client that stops sending is not left open."""
299 monkeypatch.setattr(live_announcements, "IDLE_TIMEOUT", 0.05)
300
301 ws = await _start_speaking(harness.client)
302 assert await _reply(ws) == "started"
303 await ws.send_bytes(b"\x01\x02")
304
305 assert await _reply(ws) == "finished"
306 harness.mass.players.play_announcement.assert_awaited_once()
307
308
309@pytest.mark.asyncio
310async def test_a_clip_that_keeps_going_is_cut_off(
311 harness: Harness, monkeypatch: pytest.MonkeyPatch
312) -> None:
313 """A client that keeps sending cannot keep its session open indefinitely."""
314 # with the idle timeout out of reach, only the wall clock bound can end this clip
315 monkeypatch.setattr(live_announcements, "MAX_SESSION_SECONDS", 0.1)
316 monkeypatch.setattr(live_announcements, "IDLE_TIMEOUT", 30)
317
318 ws = await _start_speaking(harness.client)
319 assert await _reply(ws) == "started"
320 await ws.send_bytes(b"\x01\x02")
321
322 assert await _reply(ws) == "finished"
323 harness.mass.players.play_announcement.assert_awaited_once()
324
325
326@pytest.mark.asyncio
327async def test_only_so_many_clips_can_be_in_progress_at_once(
328 harness: Harness, monkeypatch: pytest.MonkeyPatch
329) -> None:
330 """The audio buffered at the same time is bounded by a cap on the sessions."""
331 monkeypatch.setattr(live_announcements, "MAX_CONCURRENT_SESSIONS", 1)
332
333 speaking = await _start_speaking(harness.client)
334 assert await _reply(speaking) == "started"
335 await speaking.send_bytes(b"\x01\x02")
336
337 rejected = await _start_speaking(harness.client)
338 assert await _close_code(rejected) == REJECTED
339
340 await speaking.send_str('{"type": "stop"}')
341 assert await _reply(speaking) == "finished"
342 assert harness.mass.players.play_announcement.await_count == 1
343
344
345@pytest.mark.asyncio
346async def test_empty_frames_are_not_part_of_the_clip(harness: Harness) -> None:
347 """A frame without audio must not grow the clip that is held in memory."""
348 served = _serve_the_clip_when_announced(harness)
349
350 ws = await _start_speaking(harness.client)
351 assert await _reply(ws) == "started"
352 await ws.send_bytes(b"")
353 await ws.send_bytes(b"\x01\x02")
354 await ws.send_bytes(b"")
355 await ws.send_str('{"type": "stop"}')
356 assert await _reply(ws) == "finished"
357
358 assert served == [[create_streaming_wave_header(LIVE_FORMAT), b"\x01\x02"]]
359
360
361@pytest.mark.asyncio
362@pytest.mark.parametrize("frames", [[], [b"", b""]], ids=["no audio at all", "only empty frames"])
363async def test_a_clip_without_audio_is_not_announced(harness: Harness, frames: list[bytes]) -> None:
364 """
365 A clip nobody spoke into is dropped instead of played.
366
367 Announcing silence would interrupt whatever the player is doing for nothing.
368 """
369 ws = await _start_speaking(harness.client)
370 assert await _reply(ws) == "started"
371 for frame in frames:
372 await ws.send_bytes(frame)
373 await ws.send_str('{"type": "stop"}')
374
375 assert await _reply(ws) == "finished"
376 harness.mass.players.play_announcement.assert_not_called()
377 assert harness.manager.active_sessions == 0
378
379
380@pytest.mark.asyncio
381async def test_a_reloaded_manager_still_announces(harness: Harness) -> None:
382 """
383 A reload leaves the manager able to announce again.
384
385 A core controller reload closes and sets the manager up again, so a shutdown
386 that is never undone would silence every clip until the server restarts.
387 """
388 await harness.manager.close()
389 harness.manager.setup()
390
391 ws = await _start_speaking(harness.client)
392 assert await _reply(ws) == "started"
393 await ws.send_bytes(b"\x01\x02")
394 await ws.send_str('{"type": "stop"}')
395
396 assert await _reply(ws) == "finished"
397 harness.mass.players.play_announcement.assert_awaited_once()
398
399
400@pytest.mark.asyncio
401async def test_an_ingress_client_announces_without_an_auth_message(harness: Harness) -> None:
402 """Home Assistant authenticates its own users, so ingress skips the auth message."""
403 user = User(user_id="user_2", username="ha_user", role=UserRole.USER)
404 with (
405 patch.object(live_announcements, "is_request_from_ingress", return_value=True),
406 patch.object(live_announcements, "get_authenticated_user", AsyncMock(return_value=user)),
407 ):
408 ws = await harness.client.ws_connect(LIVE_ANNOUNCEMENT_ROUTE)
409 await ws.send_json({"type": "start", "player_id": PLAYER_ID, "sample_rate": SAMPLE_RATE})
410 assert await _reply(ws) == "started"
411 await ws.send_bytes(b"\x01\x02")
412 await ws.send_str('{"type": "stop"}')
413 assert await _reply(ws) == "finished"
414
415 harness.mass.webserver.auth.authenticate_with_token.assert_not_called()
416 harness.mass.players.play_announcement.assert_awaited_once()
417
418
419@pytest.mark.asyncio
420async def test_the_home_assistant_system_user_announces_over_ingress(harness: Harness) -> None:
421 """Home Assistant announces as its own system user, which ingress is the home of."""
422 user = User(user_id="user_6", username=HOMEASSISTANT_SYSTEM_USER, role=UserRole.SERVICE)
423 with (
424 patch.object(live_announcements, "is_request_from_ingress", return_value=True),
425 patch.object(live_announcements, "get_authenticated_user", AsyncMock(return_value=user)),
426 ):
427 ws = await harness.client.ws_connect(LIVE_ANNOUNCEMENT_ROUTE)
428 await ws.send_json({"type": "start", "player_id": PLAYER_ID, "sample_rate": SAMPLE_RATE})
429 assert await _reply(ws) == "started"
430 await ws.send_bytes(b"\x01\x02")
431 await ws.send_str('{"type": "stop"}')
432 assert await _reply(ws) == "finished"
433
434 harness.mass.players.play_announcement.assert_awaited_once()
435
436
437@pytest.mark.asyncio
438async def test_the_home_assistant_system_user_is_rejected_off_ingress(harness: Harness) -> None:
439 """The Home Assistant system token is not accepted on the regular webserver."""
440 harness.mass.webserver.auth.authenticate_with_token = AsyncMock(
441 return_value=User(
442 user_id="user_7", username=HOMEASSISTANT_SYSTEM_USER, role=UserRole.SERVICE
443 )
444 )
445 ws = await harness.client.ws_connect(LIVE_ANNOUNCEMENT_ROUTE)
446 await ws.send_json({"type": "auth", "token": VALID_TOKEN})
447
448 assert await _close_code(ws) == REJECTED
449 harness.mass.players.play_announcement.assert_not_called()
450
451
452@pytest.mark.asyncio
453async def test_an_auth_message_without_a_token_is_rejected(harness: Harness) -> None:
454 """A client that presents no token never gets to announce."""
455 ws = await harness.client.ws_connect(LIVE_ANNOUNCEMENT_ROUTE)
456 await ws.send_json({"type": "auth"})
457
458 assert await _close_code(ws) == REJECTED
459 harness.mass.players.play_announcement.assert_not_called()
460
461
462@pytest.mark.asyncio
463async def test_an_unknown_token_is_rejected(harness: Harness) -> None:
464 """A token that does not resolve to a user never gets to announce."""
465 harness.mass.webserver.auth.authenticate_with_token = AsyncMock(return_value=None)
466 ws = await harness.client.ws_connect(LIVE_ANNOUNCEMENT_ROUTE)
467 await ws.send_json({"type": "auth", "token": "nope"})
468
469 assert await _close_code(ws) == REJECTED
470 harness.mass.players.play_announcement.assert_not_called()
471
472
473@pytest.mark.asyncio
474async def test_a_user_that_may_not_control_players_is_rejected(harness: Harness) -> None:
475 """Announcing takes the players control scope, whatever else the token is valid for."""
476 # a role id outside ROLE_SCOPES grants no scopes at all
477 harness.mass.webserver.auth.authenticate_with_token = AsyncMock(
478 return_value=User(user_id="user_3", username="kiosk", role="kiosk")
479 )
480 ws = await harness.client.ws_connect(LIVE_ANNOUNCEMENT_ROUTE)
481 await ws.send_json({"type": "auth", "token": VALID_TOKEN})
482
483 assert await _close_code(ws) == REJECTED
484 harness.mass.players.play_announcement.assert_not_called()
485
486
487@pytest.mark.asyncio
488async def test_an_unavailable_player_is_rejected(harness: Harness) -> None:
489 """
490 A player that is offline is refused up front.
491
492 The command for such a player is silently dropped downstream, which would leave
493 the client believing its announcement played.
494 """
495 harness.player.available = False
496
497 ws = await _start_speaking(harness.client)
498
499 assert await _close_code(ws) == REJECTED
500 harness.mass.players.play_announcement.assert_not_called()
501
502
503@pytest.mark.asyncio
504async def test_a_long_rejection_reason_still_reaches_the_client(harness: Harness) -> None:
505 """The reason quotes what the client sent, which must still fit in a close frame."""
506 ws = await harness.client.ws_connect(LIVE_ANNOUNCEMENT_ROUTE)
507 await ws.send_json({"type": "auth", "token": VALID_TOKEN})
508 await ws.send_json({"type": "start", "player_id": "x" * 500, "sample_rate": SAMPLE_RATE})
509
510 msg = await asyncio.wait_for(ws.receive(), timeout=REPLY_TIMEOUT)
511 assert msg.type is WSMsgType.CLOSE
512 assert msg.data == REJECTED
513 assert msg.extra is not None
514 assert msg.extra.startswith("Unknown or unavailable player")
515 assert len(msg.extra.encode()) <= MAX_CLOSE_REASON_BYTES
516
517
518@pytest.mark.asyncio
519@pytest.mark.parametrize(
520 "start_message",
521 [
522 {"type": "start", "player_id": "ghost", "sample_rate": SAMPLE_RATE},
523 {"type": "start", "player_id": PLAYER_ID, "sample_rate": 1000},
524 {"type": "start", "player_id": PLAYER_ID, "sample_rate": SAMPLE_RATE, "channels": 3},
525 ],
526 ids=["unknown player", "sample rate out of range", "channel count out of range"],
527)
528async def test_an_unusable_start_message_is_rejected(
529 harness: Harness, start_message: dict[str, object]
530) -> None:
531 """Nothing is announced when the client cannot say what to send where."""
532 ws = await harness.client.ws_connect(LIVE_ANNOUNCEMENT_ROUTE)
533 await ws.send_json({"type": "auth", "token": VALID_TOKEN})
534 await ws.send_json(start_message)
535
536 assert await _close_code(ws) == REJECTED
537 harness.mass.players.play_announcement.assert_not_called()
538
539
540def _session(session_id: str = "session1") -> LiveAnnouncementSession:
541 """Return a session for the format the tests speak in."""
542 return LiveAnnouncementSession(
543 session_id, f"{BASE_URL}/live_announcement/{session_id}.wav", LIVE_FORMAT
544 )
545
546
547def _player() -> MagicMock:
548 """Return the player the tests announce on."""
549 player = MagicMock()
550 player.player_id = PLAYER_ID
551 player.display_name = "Kitchen"
552 player.available = True
553 player.protocol_parent_id = None
554 return player
555
556
557def _announce_through_player_commands(harness: Harness) -> list[tuple[str, User | None]]:
558 """
559 Announce through the guards every player command passes, instead of a bare mock.
560
561 :param harness: The harness whose announcements should run through the guards.
562 :return: The player and the user of every announcement that got through.
563 """
564 controller = PlayerController.__new__(PlayerController)
565 controller._players = {PLAYER_ID: harness.player}
566 controller.logger = logging.getLogger("test.streams.live_announcements.players")
567 announced: list[tuple[str, User | None]] = []
568
569 async def _play(_self: PlayerController, player_id: str, **_kwargs: Any) -> None:
570 announced.append((player_id, get_current_user()))
571
572 harness.mass.players.play_announcement = MethodType(handle_player_command(_play), controller)
573 return announced
574
575
576def _block_announcement(harness: Harness) -> BlockedAnnouncement:
577 """
578 Keep an announcement playing until the test releases it.
579
580 :param harness: The harness whose announcements should block.
581 :return: The events reporting the announcement and releasing it again.
582 """
583 blocked = BlockedAnnouncement(asyncio.Event(), asyncio.Event(), asyncio.Event())
584
585 async def _play(*_args: Any, **_kwargs: Any) -> None:
586 blocked.started.set()
587 await blocked.release.wait()
588 blocked.done.set()
589
590 harness.mass.players.play_announcement = AsyncMock(side_effect=_play)
591 return blocked
592
593
594def _serve_the_clip_when_announced(harness: Harness) -> list[list[bytes]]:
595 """
596 Fetch the announcement url the way the renderer does, whenever one is announced.
597
598 :param harness: The harness whose announcements should be fetched.
599 :return: What was written to the stream for each of them.
600 """
601 served: list[list[bytes]] = []
602
603 async def _pull(_player_id: str, url: str, **_kwargs: Any) -> None:
604 request, writer = _stream_request(_session_id(url))
605 await harness.manager.serve_stream(request)
606 served.append(_written(writer))
607
608 harness.mass.players.play_announcement = AsyncMock(side_effect=_pull)
609 return served
610
611
612async def _wait(event: asyncio.Event) -> None:
613 """Wait for something the server is expected to do right away."""
614 await asyncio.wait_for(event.wait(), timeout=REPLY_TIMEOUT)
615
616
617async def _start_speaking(
618 client: TestClient[web.Request, web.Application], **start: object
619) -> ClientWebSocketResponse:
620 """
621 Connect, authenticate and send the start message.
622
623 :param client: Test client for the manager's websocket route.
624 :param start: Extra fields to put in the start message.
625 """
626 ws = await client.ws_connect(LIVE_ANNOUNCEMENT_ROUTE)
627 await ws.send_json({"type": "auth", "token": VALID_TOKEN})
628 await ws.send_json(
629 {"type": "start", "player_id": PLAYER_ID, "sample_rate": SAMPLE_RATE, **start}
630 )
631 return ws
632
633
634async def _reply(ws: ClientWebSocketResponse) -> str:
635 """Return the type of the next message the server sends."""
636 message = await asyncio.wait_for(ws.receive_json(), timeout=REPLY_TIMEOUT)
637 return str(message["type"])
638
639
640async def _close_code(ws: ClientWebSocketResponse) -> int:
641 """Return the code the server closed the connection with."""
642 msg = await asyncio.wait_for(ws.receive(), timeout=REPLY_TIMEOUT)
643 assert msg.type is WSMsgType.CLOSE
644 return int(msg.data)
645
646
647def _stream_request(session_id: str) -> tuple[web.Request, MagicMock]:
648 """Return a request for the audio of the given session, plus the writer serving it."""
649 writer = MagicMock()
650 writer.write = AsyncMock()
651 writer.write_headers = AsyncMock()
652 writer.write_eof = AsyncMock()
653 writer.drain = AsyncMock()
654 request = make_mocked_request(
655 "GET",
656 f"/live_announcement/{session_id}.wav",
657 match_info={"session_id": session_id},
658 writer=writer,
659 )
660 return request, writer
661
662
663def _written(writer: MagicMock) -> list[bytes]:
664 """Return the chunks that were written to a served stream, header first."""
665 return [call.args[0] for call in writer.write.call_args_list]
666
667
668def _session_id(url: str) -> str:
669 """Return the session id in a live announcement url."""
670 return url.rsplit("/", maxsplit=1)[-1].removesuffix(".wav")
671