music-assistant-server
17.9 KB•PY
test_party_qr_cover.py
17.9 KB • 506 lines • python
1"""Tests for the party QR cover compositor (spec 0004)."""
2
3from __future__ import annotations
4
5import asyncio
6import io
7import json
8import threading
9import time
10from typing import Any
11from unittest.mock import AsyncMock, MagicMock, Mock
12from urllib.parse import parse_qs, urlencode, urlsplit
13
14import pytest
15import segno
16from aiohttp import web
17from aiohttp.test_utils import TestClient as AiohttpTestClient
18from aiohttp.test_utils import TestServer, make_mocked_request
19from PIL import Image
20
21from music_assistant.helpers.util import join_task
22from music_assistant.providers.msx_bridge import http_server as http_server_module
23from music_assistant.providers.msx_bridge.http_server import (
24 MSXHTTPServer,
25 PartyInfo,
26 _stamp_qr_on_cover,
27)
28from music_assistant.providers.msx_bridge.mappers import map_tracks_to_msx_playlist
29from music_assistant.providers.msx_bridge.provider import MSXBridgeProvider
30from tests.common import collect_loop_errors
31
32JOIN_URL = "http://ma.local:8095/?join=ABC123"
33COVER_URL = "http://ma.local:8095/imageproxy?path=cover.jpg"
34
35
36def _party_mock(url: str | None = JOIN_URL) -> Mock:
37 """Return a mock Party plugin provider."""
38 party = Mock()
39 party.get_party_url = AsyncMock(return_value=url)
40 config = Mock()
41 config.party_name = "My Party"
42 config.qr_text = "Scan to join!"
43 party.get_party_config = AsyncMock(return_value=config)
44 return party
45
46
47def _black_cover_png(size: int = 200) -> bytes:
48 """Return a solid black square PNG."""
49 buf = io.BytesIO()
50 Image.new("RGB", (size, size), (0, 0, 0)).save(buf, format="PNG")
51 return buf.getvalue()
52
53
54def _qr_png() -> bytes:
55 """Return a small QR PNG."""
56 buf = io.BytesIO()
57 segno.make(JOIN_URL, error="m").save(buf, kind="png", scale=4)
58 return buf.getvalue()
59
60
61def _http_session_mock(body: bytes, status: int = 200) -> Mock:
62 """Return a mock aiohttp session whose get() yields the given body."""
63 resp = AsyncMock()
64 resp.status = status
65 resp.read = AsyncMock(return_value=body)
66 cm = MagicMock()
67 cm.__aenter__ = AsyncMock(return_value=resp)
68 cm.__aexit__ = AsyncMock(return_value=False)
69 session = Mock()
70 session.get = Mock(return_value=cm)
71 return session
72
73
74def _failing_http_session_mock(release: asyncio.Event) -> Mock:
75 """Return a mock session whose get() fails once released."""
76
77 async def _gated_read() -> bytes:
78 await release.wait()
79 raise ConnectionResetError("connection reset while fetching the cover")
80
81 resp = AsyncMock()
82 resp.status = 200
83 resp.read = _gated_read
84 cm = MagicMock()
85 cm.__aenter__ = AsyncMock(return_value=resp)
86 cm.__aexit__ = AsyncMock(return_value=False)
87 session = Mock()
88 session.get = Mock(return_value=cm)
89 return session
90
91
92def _slow_http_session_mock(body: bytes, release: asyncio.Event) -> Mock:
93 """Return a mock session whose get() blocks reading the body until released."""
94
95 async def _gated_read() -> bytes:
96 await release.wait()
97 return body
98
99 resp = AsyncMock()
100 resp.status = 200
101 resp.read = _gated_read
102 cm = MagicMock()
103 cm.__aenter__ = AsyncMock(return_value=resp)
104 cm.__aexit__ = AsyncMock(return_value=False)
105 session = Mock()
106 session.get = Mock(return_value=cm)
107 return session
108
109
110# --- Compositor ---
111
112
113def test_stamp_qr_on_cover_composites() -> None:
114 """The QR lands bottom-right with a white quiet zone; dimensions are preserved."""
115 cover = _black_cover_png(200)
116 stamped = _stamp_qr_on_cover(cover, _qr_png())
117
118 assert stamped != cover
119 img = Image.open(io.BytesIO(stamped))
120 assert img.size == (200, 200)
121 # top-left quadrant stays untouched cover (black)
122 assert img.convert("RGB").getpixel((10, 10)) == (0, 0, 0)
123 # bottom-right quadrant contains white QR quiet-zone pixels
124 quadrant = img.convert("RGB").crop((100, 100, 200, 200))
125 assert any(px == (255, 255, 255) for px in list(quadrant.getdata()))
126
127
128# --- /api/party/qr-cover.png endpoint ---
129
130
131async def test_qr_cover_active_party_returns_png(
132 provider: MSXBridgeProvider, mass_mock: Mock
133) -> None:
134 """With an active party and an allowed source, the composited PNG is served."""
135 mass_mock.get_provider = Mock(return_value=_party_mock())
136 mass_mock.webserver.base_url = "http://ma.local:8095"
137 mass_mock.http_session = _http_session_mock(_black_cover_png())
138 server = MSXHTTPServer(provider, 0)
139 client = AiohttpTestClient(TestServer(server.app))
140 await client.start_server()
141 try:
142 resp = await client.get(
143 "/api/party/qr-cover.png", params={"image": COVER_URL}, allow_redirects=False
144 )
145 assert resp.status == 200
146 assert resp.headers["Content-Type"] == "image/png"
147 body = await resp.read()
148 assert Image.open(io.BytesIO(body)).size == (200, 200)
149 finally:
150 await client.close()
151
152
153async def test_qr_cover_composite_runs_off_event_loop(
154 provider: MSXBridgeProvider, mass_mock: Mock, monkeypatch: pytest.MonkeyPatch
155) -> None:
156 """PIL compositing must run in a worker thread, never on the event loop."""
157 mass_mock.get_provider = Mock(return_value=_party_mock())
158 mass_mock.webserver.base_url = "http://ma.local:8095"
159 mass_mock.http_session = _http_session_mock(_black_cover_png())
160 loop_thread = threading.get_ident()
161 stamp_threads: list[int] = []
162
163 def _tracking_stamp(cover_bytes: bytes, qr_bytes: bytes) -> bytes:
164 stamp_threads.append(threading.get_ident())
165 return _stamp_qr_on_cover(cover_bytes, qr_bytes)
166
167 monkeypatch.setattr(http_server_module, "_stamp_qr_on_cover", _tracking_stamp)
168 server = MSXHTTPServer(provider, 0)
169 client = AiohttpTestClient(TestServer(server.app))
170 await client.start_server()
171 try:
172 resp = await client.get(
173 "/api/party/qr-cover.png", params={"image": COVER_URL}, allow_redirects=False
174 )
175 assert resp.status == 200
176 assert stamp_threads
177 assert loop_thread not in stamp_threads
178 finally:
179 await client.close()
180
181
182async def test_qr_cover_concurrent_misses_coalesce(
183 provider: MSXBridgeProvider, mass_mock: Mock, monkeypatch: pytest.MonkeyPatch
184) -> None:
185 """Concurrent requests for the same cover share one fetch and one composite."""
186 mass_mock.get_provider = Mock(return_value=_party_mock())
187 mass_mock.webserver.base_url = "http://ma.local:8095"
188 release = asyncio.Event()
189 mass_mock.http_session = _slow_http_session_mock(_black_cover_png(), release)
190 stamp_calls: list[int] = []
191
192 def _tracking_stamp(cover_bytes: bytes, qr_bytes: bytes) -> bytes:
193 stamp_calls.append(1)
194 return _stamp_qr_on_cover(cover_bytes, qr_bytes)
195
196 monkeypatch.setattr(http_server_module, "_stamp_qr_on_cover", _tracking_stamp)
197 server = MSXHTTPServer(provider, 0)
198 client = AiohttpTestClient(TestServer(server.app))
199 await client.start_server()
200 try:
201 requests = [
202 asyncio.ensure_future(
203 client.get(
204 "/api/party/qr-cover.png", params={"image": COVER_URL}, allow_redirects=False
205 )
206 )
207 for _ in range(5)
208 ]
209 await asyncio.sleep(0.05)
210 release.set()
211 responses = await asyncio.gather(*requests)
212 assert all(r.status == 200 for r in responses)
213 assert mass_mock.http_session.get.call_count == 1
214 assert len(stamp_calls) == 1
215 finally:
216 await client.close()
217
218
219async def test_qr_cover_render_survives_requester_cancellation(
220 provider: MSXBridgeProvider, mass_mock: Mock
221) -> None:
222 """A disconnected TV must not cancel the shared render; the cache still fills."""
223 release = asyncio.Event()
224 mass_mock.http_session = _slow_http_session_mock(_black_cover_png(), release)
225 server = MSXHTTPServer(provider, 0)
226 cache_key = (COVER_URL, "v1")
227
228 task = server._qr_cover_task(cache_key, COVER_URL, JOIN_URL)
229 assert server._qr_cover_task(cache_key, COVER_URL, JOIN_URL) is task
230 waiter = asyncio.ensure_future(join_task(task))
231 await asyncio.sleep(0)
232 waiter.cancel()
233 release.set()
234 rendered = await task
235
236 assert server._qr_cover_cache[cache_key] == rendered
237 assert cache_key not in server._qr_cover_inflight
238
239
240async def test_qr_cover_render_failure_after_cancellation_logs_no_loop_error(
241 provider: MSXBridgeProvider, mass_mock: Mock
242) -> None:
243 """A render failing after one TV gave up reaches the waiting TV only, not the log."""
244 release = asyncio.Event()
245 mass_mock.get_provider = Mock(return_value=_party_mock())
246 mass_mock.webserver.base_url = "http://ma.local:8095"
247 mass_mock.http_session = _failing_http_session_mock(release)
248 server = MSXHTTPServer(provider, 0)
249 path = f"/api/party/qr-cover.png?{urlencode({'image': COVER_URL})}"
250
251 with collect_loop_errors() as reported:
252 gave_up = asyncio.create_task(
253 server._handle_party_qr_cover(make_mocked_request("GET", path))
254 )
255 waiting = asyncio.create_task(
256 server._handle_party_qr_cover(make_mocked_request("GET", path))
257 )
258 while not server._qr_cover_inflight and not gave_up.done():
259 await asyncio.sleep(0)
260 await asyncio.sleep(0) # let the second TV join the same render
261 gave_up.cancel()
262 with pytest.raises(asyncio.CancelledError):
263 await gave_up
264 # release the fetch only once the cancellation is fully processed, so the failure
265 # reliably lands after the TV that gave up is gone
266 release.set()
267 with pytest.raises(web.HTTPFound) as redirect:
268 await waiting
269
270 assert str(redirect.value.location) == COVER_URL
271 assert mass_mock.http_session.get.call_count == 1
272 assert reported == []
273
274
275async def test_qr_cover_no_party_redirects_to_original(
276 provider: MSXBridgeProvider, mass_mock: Mock
277) -> None:
278 """Without an active party the endpoint redirects to the (allowed) original image."""
279 mass_mock.webserver.base_url = "http://ma.local:8095"
280 server = MSXHTTPServer(provider, 0)
281 client = AiohttpTestClient(TestServer(server.app))
282 await client.start_server()
283 try:
284 resp = await client.get(
285 "/api/party/qr-cover.png", params={"image": COVER_URL}, allow_redirects=False
286 )
287 assert resp.status == 302
288 assert resp.headers["Location"] == COVER_URL
289 finally:
290 await client.close()
291
292
293async def test_qr_cover_disallowed_source_rejected(
294 provider: MSXBridgeProvider, mass_mock: Mock
295) -> None:
296 """External-host image URLs are never fetched NOR redirected to (open redirect)."""
297 mass_mock.get_provider = Mock(return_value=_party_mock())
298 mass_mock.webserver.base_url = "http://ma.local:8095"
299 mass_mock.http_session = _http_session_mock(_black_cover_png())
300 server = MSXHTTPServer(provider, 0)
301 client = AiohttpTestClient(TestServer(server.app))
302 await client.start_server()
303 try:
304 evil = "http://evil.example/img.png"
305 resp = await client.get(
306 "/api/party/qr-cover.png", params={"image": evil}, allow_redirects=False
307 )
308 assert resp.status == 400
309 mass_mock.http_session.get.assert_not_called()
310 finally:
311 await client.close()
312
313
314async def test_qr_cover_prefix_bypass_rejected(
315 provider: MSXBridgeProvider, mass_mock: Mock
316) -> None:
317 """A host that merely starts with an allowed base must be rejected (SSRF bypass)."""
318 mass_mock.get_provider = Mock(return_value=_party_mock())
319 mass_mock.webserver.base_url = "http://ma.local:8095"
320 mass_mock.http_session = _http_session_mock(_black_cover_png())
321 server = MSXHTTPServer(provider, 0)
322 client = AiohttpTestClient(TestServer(server.app))
323 await client.start_server()
324 try:
325 bypass = "http://ma.local:8095.evil.example/img.png"
326 resp = await client.get(
327 "/api/party/qr-cover.png", params={"image": bypass}, allow_redirects=False
328 )
329 assert resp.status == 400
330 mass_mock.http_session.get.assert_not_called()
331 finally:
332 await client.close()
333
334
335async def test_qr_cover_fetch_does_not_follow_redirects(
336 provider: MSXBridgeProvider, mass_mock: Mock
337) -> None:
338 """The cover fetch must not follow redirects (allowlisted host 302 -> loopback)."""
339 mass_mock.get_provider = Mock(return_value=_party_mock())
340 mass_mock.webserver.base_url = "http://ma.local:8095"
341 mass_mock.http_session = _http_session_mock(_black_cover_png())
342 server = MSXHTTPServer(provider, 0)
343 client = AiohttpTestClient(TestServer(server.app))
344 await client.start_server()
345 try:
346 resp = await client.get(
347 "/api/party/qr-cover.png", params={"image": COVER_URL}, allow_redirects=False
348 )
349 assert resp.status == 200
350 assert mass_mock.http_session.get.call_args.kwargs.get("allow_redirects") is False
351 finally:
352 await client.close()
353
354
355async def test_qr_cover_fetch_failure_redirects(
356 provider: MSXBridgeProvider, mass_mock: Mock
357) -> None:
358 """A failing cover fetch degrades to a redirect, never a 500."""
359 mass_mock.get_provider = Mock(return_value=_party_mock())
360 mass_mock.webserver.base_url = "http://ma.local:8095"
361 mass_mock.http_session = _http_session_mock(b"", status=404)
362 server = MSXHTTPServer(provider, 0)
363 client = AiohttpTestClient(TestServer(server.app))
364 await client.start_server()
365 try:
366 resp = await client.get(
367 "/api/party/qr-cover.png", params={"image": COVER_URL}, allow_redirects=False
368 )
369 assert resp.status == 302
370 assert resp.headers["Location"] == COVER_URL
371 finally:
372 await client.close()
373
374
375# --- Playlist background wiring ---
376
377
378def _track_mock() -> Mock:
379 track = Mock()
380 track.name = "Test Track"
381 track.uri = "library://track/1"
382 track.duration = 180
383 track.artist_str = "Artist"
384 track.image = Mock()
385 return track
386
387
388def test_playlist_backgrounds_use_qr_cover_when_party_active(
389 provider: MSXBridgeProvider, mass_mock: Mock
390) -> None:
391 """With a party active, playlist item backgrounds route through the compositor."""
392 mass_mock.metadata.get_image_url = Mock(return_value=COVER_URL)
393 playlist = map_tracks_to_msx_playlist(
394 [_track_mock()],
395 0,
396 "http://tv-host:8099",
397 "msx_test",
398 provider,
399 qr_cover_base="http://tv-host:8099/api/party/qr-cover.png",
400 )
401 assert playlist.items is not None
402 item = playlist.items[0]
403 assert item.background is not None
404 assert item.background.startswith("http://tv-host:8099/api/party/qr-cover.png?image=")
405 assert "cover.jpg" in item.background
406 # the small thumbnail stays a clean cover
407 assert item.image == COVER_URL
408
409
410def test_playlist_qr_cover_uses_proxied_image_for_external_cover(
411 provider: MSXBridgeProvider, mass_mock: Mock
412) -> None:
413 """
414 QR-cover backgrounds wrap the MA-proxied cover URL, not the external one.
415
416 A remotely-accessible cover resolves to an external CDN URL the qr-cover
417 endpoint rejects (400) â the composited background must wrap the MA-proxied
418 URL instead, so the cover still loads on the TV during a party.
419 """
420
421 def _get_image(_image: object, prefer_proxy: bool = False, **_kw: object) -> str:
422 return "http://ma.local:8095/imageproxy/abc" if prefer_proxy else "https://cdn.ext/art.jpg"
423
424 mass_mock.metadata.get_image_url = Mock(side_effect=_get_image)
425 playlist = map_tracks_to_msx_playlist(
426 [_track_mock()],
427 0,
428 "http://tv-host:8099",
429 "msx_test",
430 provider,
431 qr_cover_base="http://tv-host:8099/api/party/qr-cover.png",
432 )
433 assert playlist.items is not None
434 bg = playlist.items[0].background
435 assert bg is not None
436 inner = parse_qs(urlsplit(bg).query)["image"][0]
437 assert inner == "http://ma.local:8095/imageproxy/abc" # proxied, not external CDN
438 # the small thumbnail stays the direct (non-proxied) cover
439 assert playlist.items[0].image == "https://cdn.ext/art.jpg"
440
441
442def test_playlist_backgrounds_unchanged_without_party(
443 provider: MSXBridgeProvider, mass_mock: Mock
444) -> None:
445 """Without a party, backgrounds keep the original cover URL."""
446 mass_mock.metadata.get_image_url = Mock(return_value=COVER_URL)
447 playlist = map_tracks_to_msx_playlist(
448 [_track_mock()], 0, "http://tv-host:8099", "msx_test", provider
449 )
450 assert playlist.items is not None
451 assert playlist.items[0].background == COVER_URL
452
453
454# --- WS play background wiring ---
455
456
457async def test_broadcast_play_rewrites_image_when_party_cached(
458 provider: MSXBridgeProvider, mass_mock: Mock
459) -> None:
460 """broadcast_play stamps the QR into the play background while a party is active."""
461 server = MSXHTTPServer(provider, 0)
462 server._party_cache = (
463 time.monotonic(),
464 PartyInfo(join_url=JOIN_URL, name="My Party", qr_text=None, qr_version="abc123"),
465 )
466 server._client_prefixes["msx_test"] = "http://tv-host:8099"
467 ws = AsyncMock()
468 ws.closed = False
469 server._ws_clients["msx_test"] = {ws}
470 coros: list[Any] = []
471
472 def _capture_task(coro: Any) -> Mock:
473 coros.append(coro)
474 return Mock()
475
476 mass_mock.create_task = Mock(side_effect=_capture_task)
477
478 server.broadcast_play("msx_test", image_url=COVER_URL, title="T")
479
480 await coros[0]
481 payload = json.loads(ws.send_str.call_args[0][0])
482 assert payload["image_url"].startswith("http://tv-host:8099/api/party/qr-cover.png?image=")
483
484
485async def test_broadcast_play_keeps_image_without_party(
486 provider: MSXBridgeProvider, mass_mock: Mock
487) -> None:
488 """broadcast_play keeps the original background when no party is active."""
489 server = MSXHTTPServer(provider, 0)
490 ws = AsyncMock()
491 ws.closed = False
492 server._ws_clients["msx_test"] = {ws}
493 coros: list[Any] = []
494
495 def _capture_task(coro: Any) -> Mock:
496 coros.append(coro)
497 return Mock()
498
499 mass_mock.create_task = Mock(side_effect=_capture_task)
500
501 server.broadcast_play("msx_test", image_url=COVER_URL, title="T")
502
503 await coros[0]
504 payload = json.loads(ws.send_str.call_args[0][0])
505 assert payload["image_url"] == COVER_URL
506