music-assistant-server

1"""Tests for MSXHTTPServer routes."""
2
3from __future__ import annotations
4
5import json
6import re
7import shutil
8import subprocess
9from collections.abc import AsyncGenerator
10from typing import TYPE_CHECKING, Any
11from unittest.mock import AsyncMock, MagicMock, Mock, patch
12from urllib.parse import quote, urlsplit
13
14import pytest
15from aiohttp.test_utils import TestClient as AiohttpTestClient
16from aiohttp.test_utils import TestServer
17from music_assistant_models.enums import PlaybackState
18from music_assistant_models.player import PlayerMedia
19
20from music_assistant.providers.msx_bridge.http_server import STATIC_DIR, MSXHTTPServer
21from music_assistant.providers.msx_bridge.mappers import map_track_to_msx
22from music_assistant.providers.msx_bridge.player import MSXPlayer
23from music_assistant.providers.msx_bridge.provider import MSXBridgeProvider
24
25if TYPE_CHECKING:
26    from aiohttp.test_utils import TestClient
27
28# --- Bootstrap and CORS ---
29
30
31async def test_health(http_client: TestClient[Any, Any]) -> None:
32    """GET /health should return 200 with status ok."""
33    resp = await http_client.get("/health")
34    assert resp.status == 200
35    data = await resp.json()
36    assert data["status"] == "ok"
37    assert data["provider"] == "msx_bridge"
38
39
40async def test_root_html(http_client: TestClient[Any, Any]) -> None:
41    """GET / should return 200 with text/html content."""
42    resp = await http_client.get("/")
43    assert resp.status == 200
44    assert "text/html" in resp.headers["Content-Type"]
45    body = await resp.text()
46    assert "MSX" in body
47
48
49async def test_root_html_escapes_host_header(http_client: TestClient[Any, Any]) -> None:
50    """A crafted Host header must not be reflected unescaped (XSS)."""
51    resp = await http_client.get("/", headers={"Host": 'evil"><script>alert(1)</script>'})
52    assert resp.status == 200
53    body = await resp.text()
54    assert "<script>alert(1)</script>" not in body
55    # the quote must be escaped so the host can't break out of href attributes
56    assert 'evil">' not in body
57
58
59async def test_root_html_sendspin_urls_escaped(http_client: TestClient[Any, Any]) -> None:
60    """Generated sendspin hrefs must be HTML-escaped as a whole, including & separators."""
61    resp = await http_client.get("/")
62    assert resp.status == 200
63    body = await resp.text()
64    assert "&amp;sendspin_url=http%3A%2F%2F" in body
65    assert "&sendspin_url=http%3A%2F%2F" not in body
66
67
68async def test_root_html_uses_sendspin_server_port_constant(
69    http_client: TestClient[Any, Any],
70) -> None:
71    """Generated Sendspin links must follow the core server port constant."""
72    with patch("music_assistant.providers.msx_bridge.http_server.SENDSPIN_SERVER_PORT", 12345):
73        resp = await http_client.get("/")
74
75    assert resp.status == 200
76    body = await resp.text()
77    assert "sendspin_url=http%3A%2F%2F" in body
78    assert "%3A12345" in body
79
80
81async def test_start_json(http_client: TestClient[Any, Any]) -> None:
82    """GET /msx/start.json should return launcher menu config."""
83    resp = await http_client.get("/msx/start.json")
84    assert resp.status == 200
85    data = await resp.json()
86    assert data["name"] == "Music Assistant"
87    assert data["parameter"].startswith("content:")
88    assert "/msx/launcher.json" in data["parameter"]
89    assert "scripts" not in data
90
91
92async def test_plugin_html(http_client: TestClient[Any, Any]) -> None:
93    """GET /msx/plugin.html should return HTML with interaction plugin."""
94    resp = await http_client.get("/msx/plugin.html")
95    assert resp.status == 200
96    assert "text/html" in resp.headers["Content-Type"]
97    body = await resp.text()
98    assert "tvx.InteractionPlugin" in body
99    assert "handleRequest" in body
100    assert resp.headers.get("Cache-Control") == "no-cache, no-store, must-revalidate"
101
102
103async def test_tvx_lib(http_client: TestClient[Any, Any]) -> None:
104    """GET /msx/tvx-plugin-module.min.js should return JS library."""
105    resp = await http_client.get("/msx/tvx-plugin-module.min.js")
106    assert resp.status == 200
107    assert "javascript" in resp.headers["Content-Type"]
108
109
110async def test_cors_headers(http_client: TestClient[Any, Any]) -> None:
111    """Responses should include CORS Access-Control-Allow-Origin header."""
112    resp = await http_client.get("/health")
113    assert resp.headers.get("Access-Control-Allow-Origin") == "*"
114
115
116# --- Stream proxy ---
117
118
119async def test_stream_player_not_found(http_client: TestClient[Any, Any]) -> None:
120    """GET /stream/nonexistent should return 404."""
121    resp = await http_client.get("/stream/nonexistent")
122    assert resp.status == 404
123
124
125async def test_stream_no_media(provider: MSXBridgeProvider, mass_mock: Mock) -> None:
126    """GET /stream/{id} should return 404 when player has no current media."""
127    mock_player = Mock(spec=MSXPlayer)
128    mock_player.current_media = None
129    token = provider.get_stream_token("msx_test")
130    mass_mock.players.get.return_value = mass_mock.players.get_player.return_value = mock_player
131
132    server = MSXHTTPServer(provider, 0)
133    client = AiohttpTestClient(TestServer(server.app))
134    await client.start_server()
135    try:
136        resp = await client.get(f"/stream/msx_test?token={token}")
137        assert resp.status == 404
138        body = await resp.text()
139        assert "No active stream" in body
140    finally:
141        await client.close()
142
143
144async def test_stream_token_is_unguessable_and_per_player(
145    provider: MSXBridgeProvider,
146) -> None:
147    """Tokens must be random and distinct per player, not a shared constant."""
148    first = provider.get_stream_token("msx_a")
149    second = provider.get_stream_token("msx_b")
150    assert first
151    assert second
152    assert first != second
153    assert len(first) >= 16
154
155
156async def test_stream_token_survives_player_reregistration(
157    provider: MSXBridgeProvider,
158) -> None:
159    """
160    A token outlives the player object.
161
162    An idle TV is unregistered after the configured timeout; rotating there would
163    strand the URLs a long-running kiosk already cached.
164    """
165    token = provider.get_stream_token("msx_kiosk")
166    assert provider.get_stream_token("msx_kiosk") == token
167
168
169async def test_audio_routes_send_no_cors_header(
170    provider: MSXBridgeProvider, mass_mock: Mock
171) -> None:
172    """Audio must not be readable by a cross-origin fetch, unlike the MSX content pages."""
173    token = provider.get_stream_token("msx_test")
174    mock_player = Mock(spec=MSXPlayer)
175    mock_player.current_media = None
176    mass_mock.players.get.return_value = mass_mock.players.get_player.return_value = mock_player
177
178    server = MSXHTTPServer(provider, 0)
179    client = AiohttpTestClient(TestServer(server.app))
180    await client.start_server()
181    try:
182        resp = await client.get(f"/stream/msx_test?token={token}")
183        assert resp.headers.get("Access-Control-Allow-Origin") is None
184        resp = await client.get("/msx/audio/msx_test")
185        assert resp.headers.get("Access-Control-Allow-Origin") is None
186        # the MSX content pages still need it — the MSX app loads them cross-origin
187        resp = await client.get("/msx/menu.json")
188        assert resp.headers.get("Access-Control-Allow-Origin") == "*"
189    finally:
190        await client.close()
191
192
193async def test_stream_rejects_missing_token(provider: MSXBridgeProvider, mass_mock: Mock) -> None:
194    """GET /stream/{id} without the player's token should be refused."""
195    mock_player = Mock(spec=MSXPlayer)
196    mock_player.current_media = Mock()
197    mass_mock.players.get.return_value = mass_mock.players.get_player.return_value = mock_player
198
199    server = MSXHTTPServer(provider, 0)
200    client = AiohttpTestClient(TestServer(server.app))
201    await client.start_server()
202    try:
203        resp = await client.get("/stream/msx_test")
204        assert resp.status == 403
205        resp = await client.get("/stream/msx_test?token=wrong")
206        assert resp.status == 403
207    finally:
208        await client.close()
209
210
211async def test_stream_not_msx_player(provider: MSXBridgeProvider, mass_mock: Mock) -> None:
212    """GET /stream/{id} should return 404 for a non-MSX player."""
213    # Return a plain Mock (not spec=MSXPlayer)
214    non_msx_player = Mock()
215    mass_mock.players.get.return_value = mass_mock.players.get_player.return_value = non_msx_player
216
217    server = MSXHTTPServer(provider, 0)
218    client = AiohttpTestClient(TestServer(server.app))
219    await client.start_server()
220    try:
221        resp = await client.get("/stream/other_player")
222        assert resp.status == 404
223        body = await resp.text()
224        assert "Player not found" in body
225    finally:
226        await client.close()
227
228
229@pytest.mark.skip(reason="stream test hangs with TestClient/streaming on some platforms")
230async def test_stream_success(provider: MSXBridgeProvider, mass_mock: Mock) -> None:
231    """GET /stream/{id} should stream audio via internal API."""
232    mock_player = Mock(spec=MSXPlayer)
233    mock_media = Mock()
234    mock_media.duration = 180
235    mock_media.source_id = None
236    mock_media.queue_item_id = None
237    mock_player.current_media = mock_media
238    mock_player.output_format = "mp3"
239    token = provider.get_stream_token("msx_test")
240    mass_mock.players.get.return_value = mass_mock.players.get_player.return_value = mock_player
241
242    # Mock get_stream to return an async generator
243    mass_mock.streams = Mock()
244    mass_mock.streams.get_stream = Mock(return_value=_async_iter([b"pcm-data"]))
245
246    server = MSXHTTPServer(provider, 0)
247    client = AiohttpTestClient(TestServer(server.app))
248    await client.start_server()
249    try:
250        chunks = [b"encoded-chunk-1", b"encoded-chunk-2"]
251        with patch(
252            "music_assistant.providers.msx_bridge.http_server.get_ffmpeg_stream",
253            return_value=_async_iter(chunks),
254        ):
255            resp = await client.get(f"/stream/msx_test?token={token}")
256            assert resp.status == 200
257            assert resp.headers["Content-Type"] == "audio/mpeg"
258            body = await resp.read()
259            assert b"encoded-chunk-1" in body
260            assert b"encoded-chunk-2" in body
261    finally:
262        await client.close()
263
264
265# --- Library API ---
266
267
268async def test_albums(http_client: TestClient[Any, Any]) -> None:
269    """GET /api/albums should return items list."""
270    resp = await http_client.get("/api/albums")
271    assert resp.status == 200
272    data = await resp.json()
273    assert "items" in data
274    assert "total" in data
275
276
277async def test_albums_with_data(provider: MSXBridgeProvider, mass_mock: Mock) -> None:
278    """GET /api/albums should format album data correctly."""
279    album = Mock()
280    album.item_id = 1
281    album.name = "Test Album"
282    album.artist_str = "Test Artist"
283    album.uri = "library://album/1"
284    album.image = None
285    mock_result = Mock()
286    mock_result.__iter__ = Mock(return_value=iter([album]))
287    mock_result.total = 1
288    mass_mock.music.albums.library_items.return_value = mock_result
289
290    server = MSXHTTPServer(provider, 0)
291    client = AiohttpTestClient(TestServer(server.app))
292    await client.start_server()
293    try:
294        resp = await client.get("/api/albums")
295        assert resp.status == 200
296        data = await resp.json()
297        assert len(data["items"]) == 1
298        assert data["items"][0]["name"] == "Test Album"
299        assert data["items"][0]["artist"] == "Test Artist"
300        assert data["total"] == 1
301    finally:
302        await client.close()
303
304
305async def test_album_tracks(http_client: TestClient[Any, Any]) -> None:
306    """GET /api/albums/{id}/tracks should return items list."""
307    resp = await http_client.get("/api/albums/1/tracks")
308    assert resp.status == 200
309    data = await resp.json()
310    assert "items" in data
311
312
313async def test_artists(http_client: TestClient[Any, Any]) -> None:
314    """GET /api/artists should return items list."""
315    resp = await http_client.get("/api/artists")
316    assert resp.status == 200
317    data = await resp.json()
318    assert "items" in data
319    assert "total" in data
320
321
322async def test_playlists(http_client: TestClient[Any, Any]) -> None:
323    """GET /api/playlists should return items list."""
324    resp = await http_client.get("/api/playlists")
325    assert resp.status == 200
326    data = await resp.json()
327    assert "items" in data
328    assert "total" in data
329
330
331async def test_tracks(http_client: TestClient[Any, Any]) -> None:
332    """GET /api/tracks should return items list."""
333    resp = await http_client.get("/api/tracks")
334    assert resp.status == 200
335    data = await resp.json()
336    assert "items" in data
337    assert "total" in data
338
339
340async def test_search(provider: MSXBridgeProvider, mass_mock: Mock) -> None:
341    """GET /api/search?q=test should return search results."""
342    server = MSXHTTPServer(provider, 0)
343    client = AiohttpTestClient(TestServer(server.app))
344    await client.start_server()
345    try:
346        resp = await client.get("/api/search?q=test")
347        assert resp.status == 200
348        data = await resp.json()
349        assert "artists" in data
350        assert "albums" in data
351        assert "tracks" in data
352        assert "playlists" in data
353        mass_mock.music.search.assert_awaited_once()
354    finally:
355        await client.close()
356
357
358async def test_search_missing_query(http_client: TestClient[Any, Any]) -> None:
359    """GET /api/search without q parameter should return 400."""
360    resp = await http_client.get("/api/search")
361    assert resp.status == 400
362    data = await resp.json()
363    assert "error" in data
364
365
366# --- Playback control ---
367
368
369def _register_msx_player(mass_mock: Mock, provider: MSXBridgeProvider, player_id: str) -> MSXPlayer:
370    """Create an MSXPlayer and register it with the mass_mock so _get_msx_player passes."""
371    player = MSXPlayer(provider=provider, player_id=player_id)
372    mass_mock.players.get_player = Mock(
373        side_effect=lambda pid, **_kwargs: player if pid == player_id else None
374    )
375    return player
376
377
378async def test_play_track(provider: MSXBridgeProvider, mass_mock: Mock) -> None:
379    """POST /api/play should call player_queues.play_media."""
380    _register_msx_player(mass_mock, provider, "msx_test")
381    server = MSXHTTPServer(provider, 0)
382    client = AiohttpTestClient(TestServer(server.app))
383    await client.start_server()
384    try:
385        resp = await client.post(
386            "/api/play",
387            json={"track_uri": "library://track/1", "player_id": "msx_test"},
388        )
389        assert resp.status == 200
390        data = await resp.json()
391        assert data["status"] == "ok"
392        mass_mock.player_queues.play_media.assert_awaited_once_with("msx_test", "library://track/1")
393    finally:
394        await client.close()
395
396
397async def test_play_unknown_player(provider: MSXBridgeProvider) -> None:
398    """POST /api/play with unknown player_id should return 404."""
399    server = MSXHTTPServer(provider, 0)
400    client = AiohttpTestClient(TestServer(server.app))
401    await client.start_server()
402    try:
403        resp = await client.post(
404            "/api/play",
405            json={"track_uri": "library://track/1", "player_id": "msx_test"},
406        )
407        assert resp.status == 404
408    finally:
409        await client.close()
410
411
412async def test_play_invalid_body(http_client: TestClient[Any, Any]) -> None:
413    """POST /api/play with invalid JSON should return 400."""
414    resp = await http_client.post(
415        "/api/play",
416        data=b"not json",
417        headers={"Content-Type": "application/json"},
418    )
419    assert resp.status == 400
420
421
422async def test_pause(provider: MSXBridgeProvider, mass_mock: Mock) -> None:
423    """POST /api/pause/{id} should call cmd_pause."""
424    _register_msx_player(mass_mock, provider, "msx_test")
425    server = MSXHTTPServer(provider, 0)
426    client = AiohttpTestClient(TestServer(server.app))
427    await client.start_server()
428    try:
429        resp = await client.post("/api/pause/msx_test")
430        assert resp.status == 200
431        mass_mock.players.cmd_pause.assert_awaited_once_with("msx_test")
432    finally:
433        await client.close()
434
435
436async def test_stop(provider: MSXBridgeProvider, mass_mock: Mock) -> None:
437    """POST /api/stop/{id} should call cmd_stop."""
438    _register_msx_player(mass_mock, provider, "msx_test")
439    server = MSXHTTPServer(provider, 0)
440    client = AiohttpTestClient(TestServer(server.app))
441    await client.start_server()
442    try:
443        resp = await client.post("/api/stop/msx_test")
444        assert resp.status == 200
445        mass_mock.players.cmd_stop.assert_awaited_once_with("msx_test")
446    finally:
447        await client.close()
448
449
450async def test_quick_stop(provider: MSXBridgeProvider, mass_mock: Mock) -> None:
451    """POST /api/quick-stop/{id} should call cmd_stop and notify_play_stopped."""
452    _register_msx_player(mass_mock, provider, "msx_test")
453    server = MSXHTTPServer(provider, 0)
454    client = AiohttpTestClient(TestServer(server.app))
455    await client.start_server()
456    try:
457        with patch.object(provider, "notify_play_stopped", Mock()) as mock_notify:
458            resp = await client.post("/api/quick-stop/msx_test")
459        assert resp.status == 200
460        mass_mock.players.cmd_stop.assert_awaited_once_with("msx_test")
461        mock_notify.assert_called_once_with("msx_test")
462    finally:
463        await client.close()
464
465
466async def test_control_unknown_player(provider: MSXBridgeProvider) -> None:
467    """Control endpoints with unknown player_id should return 404."""
468    server = MSXHTTPServer(provider, 0)
469    client = AiohttpTestClient(TestServer(server.app))
470    await client.start_server()
471    try:
472        for path in ("/api/pause/unknown", "/api/stop/unknown", "/api/quick-stop/unknown"):
473            resp = await client.post(path)
474            assert resp.status == 404, f"{path} should return 404 for unknown player"
475    finally:
476        await client.close()
477
478
479async def test_control_endpoints_reject_cross_site(http_client: TestClient[Any, Any]) -> None:
480    """
481    State-changing endpoints must reject browser cross-site requests (CSRF).
482
483    Any web page can fire a GET via an img/script tag; modern browsers stamp
484    such requests with Sec-Fetch-Site: cross-site. The rejection must happen
485    before the player lookup so probing is impossible too.
486    """
487    headers = {"Sec-Fetch-Site": "cross-site"}
488    for path in (
489        "/api/pause/msx_x",
490        "/api/stop/msx_x",
491        "/api/quick-stop/msx_x",
492        "/api/next/msx_x",
493        "/api/previous/msx_x",
494    ):
495        resp = await http_client.get(path, headers=headers)
496        assert resp.status == 403, f"{path} must reject cross-site GET"
497
498    resp = await http_client.post(
499        "/api/play",
500        json={"track_uri": "library://track/1", "player_id": "msx_x"},
501        headers=headers,
502    )
503    assert resp.status == 403, "/api/play must reject cross-site POST"
504
505
506async def test_control_endpoints_allow_same_origin(
507    provider: MSXBridgeProvider, mass_mock: Mock
508) -> None:
509    """Same-origin browser requests (web player, MSX plugin) must still work."""
510    _register_msx_player(mass_mock, provider, "msx_test")
511    server = MSXHTTPServer(provider, 0)
512    client = AiohttpTestClient(TestServer(server.app))
513    await client.start_server()
514    try:
515        resp = await client.get("/api/pause/msx_test", headers={"Sec-Fetch-Site": "same-origin"})
516        assert resp.status == 200
517        mass_mock.players.cmd_pause.assert_awaited_once_with("msx_test")
518    finally:
519        await client.close()
520
521
522async def test_control_endpoints_reject_unexpected_methods(
523    http_client: TestClient[Any, Any],
524) -> None:
525    """Control endpoints accept only GET and POST — no wildcard methods."""
526    resp = await http_client.delete("/api/pause/msx_x")
527    assert resp.status == 405
528
529
530# --- MSX content page actions ---
531
532
533def _make_album_mock(item_id: int = 1, name: str = "Test Album") -> Mock:
534    """Create a mock album object."""
535    album = Mock()
536    album.item_id = item_id
537    album.name = name
538    album.artist_str = "Test Artist"
539    album.uri = f"library://album/{item_id}"
540    album.image = None
541    return album
542
543
544def _make_track_mock(item_id: int = 1, name: str = "Test Track") -> Mock:
545    """Create a mock track object."""
546    track = Mock()
547    track.item_id = item_id
548    track.name = name
549    track.artist_str = "Test Artist"
550    track.uri = f"library://track/{item_id}"
551    track.image = None
552    track.album = Mock(name="Test Album")
553    track.duration = 180
554    return track
555
556
557def _make_artist_mock(item_id: int = 1, name: str = "Test Artist") -> Mock:
558    """Create a mock artist object."""
559    artist = Mock()
560    artist.item_id = item_id
561    artist.name = name
562    artist.uri = f"library://artist/{item_id}"
563    artist.image = None
564    return artist
565
566
567def _make_playlist_mock(item_id: int = 1, name: str = "Test Playlist") -> Mock:
568    """Create a mock playlist object."""
569    playlist = Mock()
570    playlist.item_id = item_id
571    playlist.name = name
572    playlist.uri = f"library://playlist/{item_id}"
573    playlist.image = None
574    playlist.owner = "test_user"
575    playlist.provider = "library"
576    return playlist
577
578
579def _make_audio_player(mass_mock: Mock) -> tuple[MagicMock, PlayerMedia]:
580    """Wire a MagicMock MSXPlayer with queue-backed media into mass_mock for audio tests."""
581    player = MagicMock(spec=MSXPlayer)
582    player.player_id = "msx_test"
583    player.output_format = "mp3"
584    player._skip_ws_notify = False
585    media = PlayerMedia(
586        uri="library://track/1",
587        title=None,
588        artist=None,
589        album=None,
590        image_url=None,
591        duration=180,
592    )
593    player.current_media = media
594    player.wait_for_media = AsyncMock(return_value=media)
595    mass_mock.players.get.return_value = mass_mock.players.get_player.return_value = player
596    return player, media
597
598
599async def test_msx_albums_have_action(provider: MSXBridgeProvider, mass_mock: Mock) -> None:
600    """GET /msx/albums.json items should have content: action for drill-down."""
601    album = _make_album_mock()
602    mock_result = Mock()
603    mock_result.__iter__ = Mock(return_value=iter([album]))
604    mass_mock.music.albums.library_items.return_value = mock_result
605
606    server = MSXHTTPServer(provider, 0)
607    client = AiohttpTestClient(TestServer(server.app))
608    await client.start_server()
609    try:
610        resp = await client.get("/msx/albums.json")
611        assert resp.status == 200
612        data = await resp.json()
613        item = data["items"][0]
614        assert "action" in item
615        assert item["action"].startswith("content:")
616        assert "/msx/albums/1/tracks.json" in item["action"]
617    finally:
618        await client.close()
619
620
621async def test_msx_artists_have_action(provider: MSXBridgeProvider, mass_mock: Mock) -> None:
622    """GET /msx/artists.json items should have content: action for drill-down."""
623    artist = _make_artist_mock()
624    mock_result = Mock()
625    mock_result.__iter__ = Mock(return_value=iter([artist]))
626    mass_mock.music.artists.library_items.return_value = mock_result
627
628    server = MSXHTTPServer(provider, 0)
629    client = AiohttpTestClient(TestServer(server.app))
630    await client.start_server()
631    try:
632        resp = await client.get("/msx/artists.json")
633        assert resp.status == 200
634        data = await resp.json()
635        item = data["items"][0]
636        assert "action" in item
637        assert item["action"].startswith("content:")
638        assert "/msx/artists/1/albums.json" in item["action"]
639    finally:
640        await client.close()
641
642
643async def test_msx_playlists_have_action(provider: MSXBridgeProvider, mass_mock: Mock) -> None:
644    """GET /msx/playlists.json items should have content: action for drill-down."""
645    playlist = _make_playlist_mock()
646    mock_result = Mock()
647    mock_result.__iter__ = Mock(return_value=iter([playlist]))
648    mass_mock.music.playlists.library_items.return_value = mock_result
649
650    server = MSXHTTPServer(provider, 0)
651    client = AiohttpTestClient(TestServer(server.app))
652    await client.start_server()
653    try:
654        resp = await client.get("/msx/playlists.json")
655        assert resp.status == 200
656        data = await resp.json()
657        item = data["items"][0]
658        assert "action" in item
659        assert item["action"].startswith("content:")
660        assert "/msx/playlists/1/tracks.json" in item["action"]
661    finally:
662        await client.close()
663
664
665async def test_msx_tracks_have_action(provider: MSXBridgeProvider, mass_mock: Mock) -> None:
666    """GET /msx/tracks.json items should have playlist: action for playback."""
667    track = _make_track_mock()
668    mock_result = Mock()
669    mock_result.__iter__ = Mock(return_value=iter([track]))
670    mass_mock.music.tracks.library_items.return_value = mock_result
671
672    server = MSXHTTPServer(provider, 0)
673    client = AiohttpTestClient(TestServer(server.app))
674    await client.start_server()
675    try:
676        resp = await client.get("/msx/tracks.json")
677        assert resp.status == 200
678        data = await resp.json()
679        item = data["items"][0]
680        assert "action" in item
681        assert item["action"].startswith("playlist:")
682        assert "/msx/playlist/tracks.json" in item["action"]
683        assert item["titleHeader"] == "{txt:msx-white:Test Track}"
684        assert "playerLabel" in item
685        assert item["playerLabel"] == "Test Track"
686    finally:
687        await client.close()
688
689
690# --- MSX detail pages ---
691
692
693async def test_msx_album_tracks(provider: MSXBridgeProvider, mass_mock: Mock) -> None:
694    """GET /msx/albums/{id}/tracks.json should return tracks with audio actions."""
695    track = _make_track_mock()
696    mass_mock.music.albums.tracks.return_value = [track]
697
698    server = MSXHTTPServer(provider, 0)
699    client = AiohttpTestClient(TestServer(server.app))
700    await client.start_server()
701    try:
702        resp = await client.get("/msx/albums/1/tracks.json")
703        assert resp.status == 200
704        data = await resp.json()
705        assert data["headline"] == "Album Tracks"
706        assert len(data["items"]) == 1
707        item = data["items"][0]
708        assert item["titleHeader"] == "{txt:msx-white:Test Track}"
709        assert item["action"].startswith("playlist:")
710        assert "/msx/playlist/album/" in item["action"]
711    finally:
712        await client.close()
713
714
715async def test_msx_artist_albums(provider: MSXBridgeProvider, mass_mock: Mock) -> None:
716    """GET /msx/artists/{id}/albums.json should return albums with content actions."""
717    album = _make_album_mock()
718    mass_mock.music.artists.albums.return_value = [album]
719
720    server = MSXHTTPServer(provider, 0)
721    client = AiohttpTestClient(TestServer(server.app))
722    await client.start_server()
723    try:
724        resp = await client.get("/msx/artists/1/albums.json")
725        assert resp.status == 200
726        data = await resp.json()
727        assert data["headline"] == "Artist Albums"
728        assert len(data["items"]) == 1
729        item = data["items"][0]
730        assert item["title"] == "Test Album"
731        assert item["action"].startswith("content:")
732        assert "/msx/albums/1/tracks.json" in item["action"]
733    finally:
734        await client.close()
735
736
737async def test_msx_playlist_tracks(provider: MSXBridgeProvider, mass_mock: Mock) -> None:
738    """GET /msx/playlists/{id}/tracks.json should return tracks with audio actions."""
739    track = _make_track_mock()
740
741    async def _mock_playlist_tracks(*_args: object, **_kwargs: object) -> AsyncGenerator[Any]:
742        yield track
743
744    mass_mock.music.playlists.tracks = Mock(side_effect=lambda *_a, **_k: _mock_playlist_tracks())
745
746    server = MSXHTTPServer(provider, 0)
747    client = AiohttpTestClient(TestServer(server.app))
748    await client.start_server()
749    try:
750        resp = await client.get("/msx/playlists/1/tracks.json")
751        assert resp.status == 200
752        data = await resp.json()
753        assert data["headline"] == "Playlist Tracks"
754        assert len(data["items"]) == 1
755        item = data["items"][0]
756        assert item["titleHeader"] == "{txt:msx-white:Test Track}"
757        assert item["action"].startswith("playlist:")
758        assert "/msx/playlist/playlist/" in item["action"]
759    finally:
760        await client.close()
761
762
763async def test_broadcast_play_path_carries_token(
764    provider: MSXBridgeProvider, mass_mock: Mock
765) -> None:
766    """The pushed stream path must carry the token the /stream route now requires."""
767    token = provider.get_stream_token("msx_test")
768
769    server = MSXHTTPServer(provider, 0)
770    ws = AsyncMock()
771    ws.closed = False
772    server._ws_clients["msx_test"] = {ws}
773    coros: list[Any] = []
774
775    def _capture_task(coro: Any) -> Mock:
776        coros.append(coro)
777        return Mock()
778
779    mass_mock.create_task = Mock(side_effect=_capture_task)
780
781    server.broadcast_play("msx_test", title="T")
782
783    await coros[0]
784    payload = json.loads(ws.send_str.call_args[0][0])
785    assert payload["path"] == f"/stream/msx_test?token={token}"
786
787
788# --- MSX audio endpoint ---
789
790
791async def test_msx_audio_missing_uri(http_client: TestClient[Any, Any]) -> None:
792    """GET /msx/audio/msx_default without ?uri= should return 400."""
793    resp = await http_client.get("/msx/audio/msx_default")
794    assert resp.status == 400
795    body = await resp.text()
796    assert "uri" in body.lower()  # "Missing uri" or "Invalid uri parameter"
797
798
799@pytest.mark.parametrize(
800    "uri",
801    [
802        "http://evil.example/payload.mp3",
803        "https://evil.example/x",
804        "rtsp://evil.example/x",
805        # the same destination wrapped in a builtin uri — parse_uri resolves both to
806        # ('builtin', 'http://evil.example/…'), so the guard must reject both
807        "builtin://track/http://evil.example/payload.mp3",
808        "builtin://radio/http://evil.example/x",
809        "builtin://unknown/https://evil.example/x",
810        "not-a-uri",
811    ],
812)
813async def test_msx_audio_rejects_raw_stream_url(
814    provider: MSXBridgeProvider, mass_mock: Mock, uri: str
815) -> None:
816    """A bare stream URL resolves to the builtin provider and must never be enqueued."""
817    server = MSXHTTPServer(provider, 0)
818    client = AiohttpTestClient(TestServer(server.app))
819    await client.start_server()
820    try:
821        _make_audio_player(mass_mock)
822        token = provider.get_stream_token("msx_test")
823        resp = await client.get(f"/msx/audio/msx_test?uri={quote(uri, safe='')}&token={token}")
824        assert resp.status == 400
825        mass_mock.player_queues.play_media.assert_not_called()
826    finally:
827        await client.close()
828
829
830async def test_api_play_rejects_non_string_body_values(
831    provider: MSXBridgeProvider, mass_mock: Mock
832) -> None:
833    """A malformed body must be a 400, not a 500 from the uri guard."""
834    server = MSXHTTPServer(provider, 0)
835    client = AiohttpTestClient(TestServer(server.app))
836    await client.start_server()
837    try:
838        _make_audio_player(mass_mock)
839        for body in (
840            {"track_uri": 123, "player_id": "msx_test"},
841            {"track_uri": True, "player_id": "msx_test"},
842            {"track_uri": "library://track/1", "player_id": 42},
843        ):
844            resp = await client.post("/api/play", json=body)
845            assert resp.status == 400
846        mass_mock.player_queues.play_media.assert_not_called()
847    finally:
848        await client.close()
849
850
851async def test_api_play_rejects_raw_stream_url(
852    provider: MSXBridgeProvider, mass_mock: Mock
853) -> None:
854    """POST /api/play must apply the same guard as the MSX audio route."""
855    server = MSXHTTPServer(provider, 0)
856    client = AiohttpTestClient(TestServer(server.app))
857    await client.start_server()
858    try:
859        _make_audio_player(mass_mock)
860        for track_uri in (
861            "http://evil.example/payload.mp3",
862            "builtin://track/http://evil.example/payload.mp3",
863        ):
864            resp = await client.post(
865                "/api/play",
866                json={"track_uri": track_uri, "player_id": "msx_test"},
867            )
868            assert resp.status == 400
869        mass_mock.player_queues.play_media.assert_not_called()
870    finally:
871        await client.close()
872
873
874async def test_msx_audio_rejects_missing_token(
875    provider: MSXBridgeProvider, mass_mock: Mock
876) -> None:
877    """A caller that was never handed a URL cannot start playback."""
878    server = MSXHTTPServer(provider, 0)
879    client = AiohttpTestClient(TestServer(server.app))
880    await client.start_server()
881    try:
882        _make_audio_player(mass_mock)
883        resp = await client.get("/msx/audio/msx_test?uri=library://track/1")
884        assert resp.status == 403
885        resp = await client.get("/msx/audio/msx_test?uri=library://track/1&token=wrong")
886        assert resp.status == 403
887        mass_mock.player_queues.play_media.assert_not_called()
888    finally:
889        await client.close()
890
891
892async def test_msx_audio_player_not_found(http_client: TestClient[Any, Any]) -> None:
893    """GET /msx/audio/nonexistent?uri=x should return 404."""
894    resp = await http_client.get("/msx/audio/nonexistent?uri=library://track/1")
895    assert resp.status == 404
896
897
898async def test_msx_audio_not_msx_player(provider: MSXBridgeProvider, mass_mock: Mock) -> None:
899    """GET /msx/audio/{id}?uri=x should return 404 for non-MSX player."""
900    non_msx_player = Mock()
901    mass_mock.players.get.return_value = mass_mock.players.get_player.return_value = non_msx_player
902
903    server = MSXHTTPServer(provider, 0)
904    client = AiohttpTestClient(TestServer(server.app))
905    await client.start_server()
906    try:
907        resp = await client.get("/msx/audio/other?uri=library://track/1")
908        assert resp.status == 404
909        body = await resp.text()
910        assert "Player not found" in body
911    finally:
912        await client.close()
913
914
915async def test_msx_audio_per_track_mode(provider: MSXBridgeProvider, mass_mock: Mock) -> None:
916    """GET /msx/audio should always use force_flow_mode=False (per-track)."""
917    server = MSXHTTPServer(provider, 0)
918    client = AiohttpTestClient(TestServer(server.app))
919    await client.start_server()
920    try:
921        _make_audio_player(mass_mock)
922        token = provider.get_stream_token("msx_test")
923
924        mass_mock.streams = Mock()
925        mass_mock.streams.get_stream = Mock(return_value=_async_iter([b"pcm"]))
926
927        chunks = [b"encoded-chunk-1"]
928        with patch(
929            "music_assistant.providers.msx_bridge.http_server.get_ffmpeg_stream",
930            return_value=_async_iter(chunks),
931        ):
932            resp = await client.get(f"/msx/audio/msx_test?uri=library://track/1&token={token}")
933            assert resp.status == 200
934
935        mass_mock.streams.get_stream.assert_called_once()
936        _args, _pos, kwargs = mass_mock.streams.get_stream.mock_calls[0]
937        assert kwargs.get("force_flow_mode") is False
938
939    finally:
940        await client.close()
941
942
943async def test_msx_audio_proxy_paces_output(provider: MSXBridgeProvider, mass_mock: Mock) -> None:
944    """The local proxy must carry the core streamserver's pacing ceiling."""
945    server = MSXHTTPServer(provider, 0)
946    client = AiohttpTestClient(TestServer(server.app))
947    await client.start_server()
948    try:
949        _make_audio_player(mass_mock)
950        token = provider.get_stream_token("msx_test")
951        mass_mock.streams = Mock()
952        mass_mock.streams.get_stream = Mock(return_value=_async_iter([b"pcm"]))
953
954        with patch(
955            "music_assistant.providers.msx_bridge.http_server.get_ffmpeg_stream",
956            return_value=_async_iter([b"encoded"]),
957        ) as ffmpeg_mock:
958            resp = await client.get(f"/msx/audio/msx_test?uri=library://track/1&token={token}")
959            assert resp.status == 200
960
961        extra_args = ffmpeg_mock.call_args.kwargs["extra_input_args"]
962        assert "-readrate" in extra_args
963        assert "-readrate_initial_burst" in extra_args
964    finally:
965        await client.close()
966
967
968async def test_msx_audio_from_playlist_skips_ws(
969    provider: MSXBridgeProvider, mass_mock: Mock
970) -> None:
971    """GET /msx/audio with from_playlist=1 should set _skip_ws_notify on the player."""
972    server = MSXHTTPServer(provider, 0)
973    client = AiohttpTestClient(TestServer(server.app))
974    await client.start_server()
975    try:
976        player, _media = _make_audio_player(mass_mock)
977        token = provider.get_stream_token("msx_test")
978
979        mass_mock.streams = Mock()
980        mass_mock.streams.get_stream = Mock(return_value=_async_iter([b"pcm"]))
981
982        # Track that _skip_ws_notify was set to True during play_media
983        notify_states: list[bool] = []
984
985        async def _capture_play_media(*_a: object, **_k: object) -> None:
986            notify_states.append(player._skip_ws_notify)
987
988        mass_mock.player_queues.play_media = _capture_play_media
989
990        chunks = [b"encoded-chunk-1"]
991        with patch(
992            "music_assistant.providers.msx_bridge.http_server.get_ffmpeg_stream",
993            return_value=_async_iter(chunks),
994        ):
995            resp = await client.get(
996                f"/msx/audio/msx_test?uri=library://track/1&from_playlist=1&token={token}"
997            )
998            assert resp.status == 200
999
1000        # _skip_ws_notify should have been True during play_media call
1001        assert notify_states == [True]
1002        # And reset to False after
1003        assert player._skip_ws_notify is False
1004
1005    finally:
1006        await client.close()
1007
1008
1009async def test_msx_audio_arms_wait_before_enqueue(
1010    provider: MSXBridgeProvider, mass_mock: Mock
1011) -> None:
1012    """GET /msx/audio must arm expect_new_media() BEFORE enqueuing new playback."""
1013    server = MSXHTTPServer(provider, 0)
1014    client = AiohttpTestClient(TestServer(server.app))
1015    await client.start_server()
1016    try:
1017        player, _media = _make_audio_player(mass_mock)
1018        token = provider.get_stream_token("msx_test")
1019
1020        mass_mock.streams = Mock()
1021        mass_mock.streams.get_stream = Mock(return_value=_async_iter([b"pcm"]))
1022
1023        call_order: list[str] = []
1024        player.expect_new_media = Mock(side_effect=lambda: call_order.append("arm"))
1025
1026        async def _record_enqueue(*_a: object, **_k: object) -> None:
1027            call_order.append("enqueue")
1028
1029        mass_mock.player_queues.play_media = _record_enqueue
1030
1031        with patch(
1032            "music_assistant.providers.msx_bridge.http_server.get_ffmpeg_stream",
1033            return_value=_async_iter([b"encoded-chunk-1"]),
1034        ):
1035            resp = await client.get(f"/msx/audio/msx_test?uri=library://track/1&token={token}")
1036            assert resp.status == 200
1037
1038        assert call_order == ["arm", "enqueue"]
1039    finally:
1040        await client.close()
1041
1042
1043# --- Served audio length (Content-Length) ---
1044
1045
1046def test_served_duration_uses_media_duration(provider: MSXBridgeProvider) -> None:
1047    """Without a seek the served audio is the whole media item."""
1048    server = MSXHTTPServer(provider, 0)
1049    media = PlayerMedia(uri="library://track/1", duration=180)
1050
1051    assert server._resolve_served_duration(media) == 180
1052
1053
1054def test_served_duration_prefers_stream_duration(provider: MSXBridgeProvider) -> None:
1055    """Starting mid-track serves less audio than the media item is long."""
1056    server = MSXHTTPServer(provider, 0)
1057    media = PlayerMedia(uri="library://track/1", duration=180, stream_duration=60)
1058
1059    assert server._resolve_served_duration(media) == 60
1060
1061
1062def test_served_duration_falls_back_to_queue_item(
1063    provider: MSXBridgeProvider, mass_mock: Mock
1064) -> None:
1065    """A media item of unknown length is resolved through its queue item."""
1066    server = MSXHTTPServer(provider, 0)
1067    queue_item = MagicMock()
1068    queue_item.media_item = None
1069    queue_item.duration = 240
1070    mass_mock.player_queues.get_item.return_value = queue_item
1071    media = PlayerMedia(uri="library://track/1", source_id="q1", queue_item_id="item1")
1072
1073    assert server._resolve_served_duration(media) == 240
1074
1075
1076# --- MSX playlist endpoints ---
1077
1078
1079async def test_msx_album_playlist_endpoint(provider: MSXBridgeProvider, mass_mock: Mock) -> None:
1080    """GET /msx/playlist/album/{id}.json should return playlist JSON."""
1081    track = _make_track_mock()
1082    mass_mock.music.albums.tracks.return_value = [track]
1083
1084    server = MSXHTTPServer(provider, 0)
1085    client = AiohttpTestClient(TestServer(server.app))
1086    await client.start_server()
1087    try:
1088        resp = await client.get("/msx/playlist/album/42.json?start=0")
1089        assert resp.status == 200
1090        data = await resp.json()
1091        assert data["type"] == "list"
1092        assert data["action"] == "player:play"
1093        assert len(data["items"]) == 1
1094        item = data["items"][0]
1095        assert item["action"].startswith("audio:")
1096        assert "/msx/audio/" in item["action"]
1097        assert "from_playlist=1" in item["action"]
1098    finally:
1099        await client.close()
1100
1101
1102async def test_msx_playlist_playlist_endpoint(provider: MSXBridgeProvider, mass_mock: Mock) -> None:
1103    """GET /msx/playlist/playlist/{id}.json should return playlist JSON."""
1104    track = _make_track_mock()
1105
1106    async def _mock_playlist_tracks(*_args: object, **_kwargs: object) -> AsyncGenerator[Any]:
1107        yield track
1108
1109    mass_mock.music.playlists.tracks = Mock(side_effect=lambda *_a, **_k: _mock_playlist_tracks())
1110
1111    server = MSXHTTPServer(provider, 0)
1112    client = AiohttpTestClient(TestServer(server.app))
1113    await client.start_server()
1114    try:
1115        resp = await client.get("/msx/playlist/playlist/5.json?start=1")
1116        assert resp.status == 200
1117        data = await resp.json()
1118        assert data["type"] == "list"
1119        assert data["action"] == "player:play"
1120        assert len(data["items"]) == 1
1121    finally:
1122        await client.close()
1123
1124
1125async def test_msx_tracks_playlist_endpoint(provider: MSXBridgeProvider, mass_mock: Mock) -> None:
1126    """GET /msx/playlist/tracks.json should return playlist JSON."""
1127    track = _make_track_mock()
1128    mock_result = Mock()
1129    mock_result.__iter__ = Mock(return_value=iter([track]))
1130    mass_mock.music.tracks.library_items.return_value = mock_result
1131
1132    server = MSXHTTPServer(provider, 0)
1133    client = AiohttpTestClient(TestServer(server.app))
1134    await client.start_server()
1135    try:
1136        resp = await client.get("/msx/playlist/tracks.json?start=0")
1137        assert resp.status == 200
1138        data = await resp.json()
1139        assert data["type"] == "list"
1140        assert len(data["items"]) == 1
1141    finally:
1142        await client.close()
1143
1144
1145# --- Duration in track formatting ---
1146
1147
1148def test_format_msx_track_includes_duration(provider: MSXBridgeProvider) -> None:
1149    """map_track_to_msx should include artist and duration in titleFooter."""
1150    track = _make_track_mock()  # duration=180
1151    item = map_track_to_msx(track, "http://localhost", "msx_test", provider)
1152    assert item.title_header == "{txt:msx-white:Test Track}"
1153    assert item.title_footer == "Test Artist · 3:00"
1154    assert item.background == item.image
1155
1156
1157def test_format_msx_track_no_duration(provider: MSXBridgeProvider) -> None:
1158    """map_track_to_msx should handle zero/missing duration gracefully."""
1159    track = _make_track_mock()
1160    track.duration = 0
1161    item = map_track_to_msx(track, "http://localhost", "msx_test", provider)
1162    assert item.title_header == "{txt:msx-white:Test Track}"
1163    assert item.title_footer == "Test Artist"
1164
1165
1166def test_format_msx_track_duration_only(provider: MSXBridgeProvider) -> None:
1167    """map_track_to_msx should show only duration when no artist."""
1168    track = _make_track_mock()
1169    track.artist_str = ""
1170    item = map_track_to_msx(track, "http://localhost", "msx_test", provider)
1171    assert item.title_header == "{txt:msx-white:Test Track}"
1172    assert item.title_footer == "3:00"
1173
1174
1175# --- Async iteration helpers for stream mocking ---
1176
1177
1178async def _async_iter(items: list[Any]) -> AsyncGenerator[Any]:
1179    """Async generator helper for mocking iter_chunked."""
1180    for item in items:
1181        yield item
1182
1183
1184# --- MSX queue-playlist endpoint ---
1185
1186
1187async def test_msx_queue_playlist_endpoint(provider: MSXBridgeProvider, mass_mock: Mock) -> None:
1188    """GET /msx/queue-playlist/{player_id}.json should return MSX playlist from MA queue."""
1189    qi1 = Mock()
1190    qi1.name = "Track 1"
1191    qi1.media_item = Mock()
1192    qi1.media_item.name = "Track 1"
1193    qi1.media_item.uri = "library://track/1"
1194    qi1.media_item.duration = 180
1195    qi1.media_item.artist_str = "Artist 1"
1196    qi1.duration = 180
1197    qi1.image = None
1198
1199    qi2 = Mock()
1200    qi2.name = "Track 2"
1201    qi2.media_item = Mock()
1202    qi2.media_item.name = "Track 2"
1203    qi2.media_item.uri = "library://track/2"
1204    qi2.media_item.duration = 200
1205    qi2.media_item.artist_str = "Artist 2"
1206    qi2.duration = 200
1207    qi2.image = None
1208
1209    mass_mock.player_queues.items = Mock(return_value=[qi1, qi2])
1210
1211    server = MSXHTTPServer(provider, 0)
1212    client = AiohttpTestClient(TestServer(server.app))
1213    await client.start_server()
1214    try:
1215        resp = await client.get("/msx/queue-playlist/msx_test.json?start=0")
1216        assert resp.status == 200
1217        data = await resp.json()
1218        assert data["type"] == "list"
1219        assert data["action"] == "player:play"
1220        assert len(data["items"]) == 2
1221        assert data["items"][0]["title"] == "Track 1"
1222        assert data["items"][1]["title"] == "Track 2"
1223        assert "from_playlist=1" in data["items"][0]["action"]
1224    finally:
1225        await client.close()
1226
1227
1228async def test_msx_queue_playlist_with_start_index(
1229    provider: MSXBridgeProvider, mass_mock: Mock
1230) -> None:
1231    """GET /msx/queue-playlist with start=1 should use player:play action."""
1232    qi = Mock()
1233    qi.name = "Track 1"
1234    qi.media_item = Mock()
1235    qi.media_item.name = "Track 1"
1236    qi.media_item.uri = "library://track/1"
1237    qi.media_item.duration = 180
1238    qi.media_item.artist_str = "Artist 1"
1239    qi.duration = 180
1240    qi.image = None
1241
1242    mass_mock.player_queues.items = Mock(return_value=[qi])
1243
1244    server = MSXHTTPServer(provider, 0)
1245    client = AiohttpTestClient(TestServer(server.app))
1246    await client.start_server()
1247    try:
1248        resp = await client.get("/msx/queue-playlist/msx_test.json?start=1")
1249        assert resp.status == 200
1250        data = await resp.json()
1251        assert data["action"] == "player:play"
1252    finally:
1253        await client.close()
1254
1255
1256async def test_msx_queue_playlist_empty_queue(provider: MSXBridgeProvider, mass_mock: Mock) -> None:
1257    """GET /msx/queue-playlist with empty queue should return empty playlist."""
1258    mass_mock.player_queues.items = Mock(return_value=[])
1259
1260    server = MSXHTTPServer(provider, 0)
1261    client = AiohttpTestClient(TestServer(server.app))
1262    await client.start_server()
1263    try:
1264        resp = await client.get("/msx/queue-playlist/msx_test.json?start=0")
1265        assert resp.status == 200
1266        data = await resp.json()
1267        assert data["type"] == "list"
1268        assert data["items"] == []
1269    finally:
1270        await client.close()
1271
1272
1273# --- WebSocket inbound message handling ---
1274
1275
1276async def test_ws_position_message(provider: MSXBridgeProvider, mass_mock: Mock) -> None:
1277    """WS position message should update player's elapsed time."""
1278    player = MSXPlayer(provider, "msx_test", name="Test TV", output_format="mp3")
1279    player.update_state = Mock()  # type: ignore[misc,method-assign]
1280    player._attr_playback_state = PlaybackState.PLAYING
1281    mass_mock.players.get.return_value = mass_mock.players.get_player.return_value = player
1282    provider.http_server = MSXHTTPServer(provider, 0)
1283
1284    server_obj = provider.http_server
1285    server_obj._handle_ws_message("msx_test", '{"type": "position", "position": 42.5}')
1286
1287    assert player._attr_elapsed_time == 42.5
1288    assert player._last_ws_position is not None
1289
1290
1291async def test_ws_position_message_unknown_player(
1292    provider: MSXBridgeProvider, mass_mock: Mock
1293) -> None:
1294    """WS position message for unknown player should not crash."""
1295    mass_mock.players.get.return_value = mass_mock.players.get_player.return_value = None
1296    provider.http_server = MSXHTTPServer(provider, 0)
1297
1298    # Should not raise
1299    provider.http_server._handle_ws_message("msx_unknown", '{"type": "position", "position": 10}')
1300
1301
1302async def test_ws_invalid_json(provider: MSXBridgeProvider) -> None:
1303    """WS invalid JSON should not crash."""
1304    provider.http_server = MSXHTTPServer(provider, 0)
1305    # Should not raise
1306    provider.http_server._handle_ws_message("msx_test", "not json")
1307
1308
1309async def test_ws_pause_message(provider: MSXBridgeProvider, mass_mock: Mock) -> None:
1310    """WS pause message should update position and call cmd_pause."""
1311    player = MSXPlayer(provider, "msx_test", name="Test TV", output_format="mp3")
1312    player.update_state = Mock()  # type: ignore[misc,method-assign]
1313    player._attr_playback_state = PlaybackState.PLAYING
1314    player._attr_elapsed_time = 10.0
1315    mass_mock.players.get.return_value = mass_mock.players.get_player.return_value = player
1316    provider.http_server = MSXHTTPServer(provider, 0)
1317
1318    provider.http_server._handle_ws_message("msx_test", '{"type": "pause", "position": 30.5}')
1319
1320    assert player._attr_elapsed_time == 30.5
1321    # Flag is now managed inside _cmd_pause_no_echo; verify the task was scheduled
1322    mass_mock.create_task.assert_called_once()
1323
1324
1325async def test_ws_resume_message(provider: MSXBridgeProvider, mass_mock: Mock) -> None:
1326    """WS resume message should call cmd_play."""
1327    player = MSXPlayer(provider, "msx_test", name="Test TV", output_format="mp3")
1328    player.update_state = Mock()  # type: ignore[misc,method-assign]
1329    player._attr_playback_state = PlaybackState.PAUSED
1330    mass_mock.players.get.return_value = mass_mock.players.get_player.return_value = player
1331    provider.http_server = MSXHTTPServer(provider, 0)
1332
1333    provider.http_server._handle_ws_message("msx_test", '{"type": "resume"}')
1334
1335    # Flag is now managed inside _cmd_play_no_echo; verify the task was scheduled
1336    mass_mock.create_task.assert_called_once()
1337
1338
1339async def test_ws_unknown_message_type(provider: MSXBridgeProvider) -> None:
1340    """WS unknown message type should not crash."""
1341    provider.http_server = MSXHTTPServer(provider, 0)
1342    # Should not raise
1343    provider.http_server._handle_ws_message("msx_test", '{"type": "unknown_cmd"}')
1344
1345
1346# --- Removed kiosk/sendspin routes ---
1347
1348
1349async def test_removed_kiosk_and_sendspin_routes_404(
1350    http_client: TestClient[Any, Any],
1351) -> None:
1352    """Removed kiosk and sendspin routes should return 404."""
1353    for path in [
1354        "/msx/kiosk-plugin.html",
1355        "/msx/kiosk.html",
1356        "/msx/kiosk-content.json",
1357        "/msx/kiosk-page.json",
1358        "/msx/kiosk-album.json",
1359        "/msx/sendspin-plugin.html",
1360        "/msx/sendspin-standalone.html",
1361        "/msx/sendspin-bundle.js",
1362    ]:
1363        resp = await http_client.get(path)
1364        assert resp.status == 404, f"Expected 404 for {path}, got {resp.status}"
1365
1366
1367# --- Server shutdown ---
1368
1369
1370async def test_server_stop_survives_ws_self_deregistration(
1371    provider: MSXBridgeProvider,
1372) -> None:
1373    """Closing a WS during stop() triggers its cleanup discard; stop() must survive it."""
1374    server = MSXHTTPServer(provider, 0)
1375
1376    class _SelfRemovingWS:
1377        closed = False
1378
1379        async def close(self) -> None:
1380            server._ws_clients["msx_test"].discard(self)
1381
1382    fake_clients: set[Any] = {_SelfRemovingWS(), _SelfRemovingWS()}
1383    server._ws_clients["msx_test"] = fake_clients
1384
1385    await server.stop()
1386
1387    assert server._ws_clients == {}
1388
1389
1390# --- Redirect stream mode (MA streamserver) ---
1391
1392
1393async def test_msx_audio_redirect_mode(provider: MSXBridgeProvider, mass_mock: Mock) -> None:
1394    """In redirect mode /msx/audio must 302-redirect to the MA streamserver stream."""
1395    provider.group_stream_mode = "redirect"
1396    server = MSXHTTPServer(provider, 0)
1397    client = AiohttpTestClient(TestServer(server.app))
1398    await client.start_server()
1399    try:
1400        _make_audio_player(mass_mock)
1401        token = provider.get_stream_token("msx_test")
1402
1403        stream_url = "http://ma:8097/single/s1/q1/i1/msx_test.mp3"
1404        mass_mock.streams = Mock()
1405        mass_mock.streams.resolve_stream_url = AsyncMock(return_value=stream_url)
1406
1407        resp = await client.get(
1408            f"/msx/audio/msx_test?uri=library://track/1&token={token}",
1409            allow_redirects=False,
1410        )
1411        assert resp.status == 302
1412        # the URL host is rewritten to the client-reachable one (see the
1413        # dedicated rewrite test); the streamserver path must be intact
1414        assert resp.headers["Location"].endswith(":8097/single/s1/q1/i1/msx_test.mp3")
1415    finally:
1416        await client.close()
1417
1418
1419async def test_msx_audio_redirect_rewrites_host_for_client(
1420    provider: MSXBridgeProvider, mass_mock: Mock
1421) -> None:
1422    """
1423    The redirect must target the host the TV already uses to reach the provider.
1424
1425    Behind Docker/NAT the MA streamserver advertises its container IP, which
1426    the TV cannot reach; only the host of the URL is rewritten — port, path
1427    and query of the streamserver URL must survive.
1428    """
1429    provider.group_stream_mode = "redirect"
1430    server = MSXHTTPServer(provider, 0)
1431    client = AiohttpTestClient(TestServer(server.app))
1432    await client.start_server()
1433    try:
1434        _make_audio_player(mass_mock)
1435        token = provider.get_stream_token("msx_test")
1436
1437        stream_url = "http://172.18.0.2:8097/single/s1/q1/i1/msx_test.mp3?flow=1"
1438        mass_mock.streams = Mock()
1439        mass_mock.streams.resolve_stream_url = AsyncMock(return_value=stream_url)
1440
1441        resp = await client.get(
1442            f"/msx/audio/msx_test?uri=library://track/1&token={token}",
1443            allow_redirects=False,
1444        )
1445        assert resp.status == 302
1446        location = urlsplit(resp.headers["Location"])
1447        assert location.hostname == "127.0.0.1"  # host the TestClient connects to
1448        assert location.port == 8097
1449        assert location.path == "/single/s1/q1/i1/msx_test.mp3"
1450        assert location.query == "flow=1"
1451    finally:
1452        await client.close()
1453
1454
1455async def test_msx_audio_redirect_mode_falls_back_to_proxy(
1456    provider: MSXBridgeProvider, mass_mock: Mock
1457) -> None:
1458    """When URL resolution fails, redirect mode must fall back to the local proxy."""
1459    provider.group_stream_mode = "redirect"
1460    server = MSXHTTPServer(provider, 0)
1461    client = AiohttpTestClient(TestServer(server.app))
1462    await client.start_server()
1463    try:
1464        _make_audio_player(mass_mock)
1465        token = provider.get_stream_token("msx_test")
1466
1467        mass_mock.streams = Mock()
1468        mass_mock.streams.resolve_stream_url = AsyncMock(side_effect=RuntimeError("boom"))
1469        mass_mock.streams.get_stream = Mock(return_value=_async_iter([b"pcm"]))
1470        filter_params = ["volume=0.5"]
1471        mass_mock.streams.audio.get_player_output_plan.return_value = Mock(
1472            filter_params=filter_params
1473        )
1474
1475        with patch(
1476            "music_assistant.providers.msx_bridge.http_server.get_ffmpeg_stream",
1477            return_value=_async_iter([b"encoded-chunk-1"]),
1478        ) as ffmpeg_stream:
1479            resp = await client.get(
1480                f"/msx/audio/msx_test?uri=library://track/1&token={token}",
1481                allow_redirects=False,
1482            )
1483            assert resp.status == 200
1484            body = await resp.read()
1485            assert b"encoded-chunk-1" in body
1486            assert ffmpeg_stream.call_args.kwargs["filter_params"] == filter_params
1487    finally:
1488        await client.close()
1489
1490
1491# --- Vendored Sendspin JS client ---
1492
1493
1494async def test_vendored_sendspin_js_served(http_client: TestClient[Any, Any]) -> None:
1495    """The Sendspin JS client must be served locally — TVs on LAN-only setups have no CDN."""
1496    resp = await http_client.get("/web/sendspin-js/index.js")
1497    assert resp.status == 200
1498    body = await resp.text()
1499    assert "SendspinPlayer" in body
1500
1501
1502def test_web_player_has_no_cdn_dependency() -> None:
1503    """web.js must import the Sendspin SDK from the vendored copy, not from a CDN."""
1504    web_js = (STATIC_DIR / "web" / "web.js").read_text(encoding="utf-8")
1505    assert "unpkg.com" not in web_js
1506    assert "jsdelivr" not in web_js
1507    assert "sendspin-js/index.js" in web_js
1508
1509
1510async def test_root_has_kiosk_url_builder(http_client: TestClient[Any, Any]) -> None:
1511    """The status page offers a kiosk URL builder with the four display toggles."""
1512    resp = await http_client.get("/")
1513    assert resp.status == 200
1514    body = await resp.text()
1515    assert 'id="kiosk-builder"' in body
1516    for name in ("controls", "party", "viz", "lyrics"):
1517        assert f'data-kiosk-param="{name}"' in body
1518    assert 'id="kiosk-builder-link"' in body
1519    assert 'id="kiosk-builder-url"' in body
1520
1521
1522def test_web_player_reads_kiosk_display_params() -> None:
1523    """web.js must read the four kiosk display params (URL contract)."""
1524    web_js = (STATIC_DIR / "web" / "web.js").read_text(encoding="utf-8")
1525    for name in ("controls", "party", "viz", "lyrics"):
1526        assert f"kioskFlag('{name}')" in web_js, f"missing kiosk param: {name}"
1527
1528
1529def test_web_player_js_parses() -> None:
1530    """web.js must be syntactically valid (guards against edits breaking the kiosk)."""
1531    node = shutil.which("node")
1532    if node is None:
1533        pytest.skip("node not available")
1534    result = subprocess.run(  # noqa: S603
1535        [node, "--check", str(STATIC_DIR / "web" / "web.js")],
1536        capture_output=True,
1537        text=True,
1538        check=False,
1539    )
1540    assert result.returncode == 0, result.stderr
1541
1542
1543def test_vendored_sendspin_js_imports_are_browser_loadable() -> None:
1544    """
1545    Every import specifier in the vendored SDK must be loadable by a browser.
1546
1547    Relative specifiers must resolve to a vendored file (the upstream dist is
1548    TypeScript output with extensionless specifiers that only work through CDN
1549    rewriting). Bare specifiers cannot resolve in browsers at all; the two
1550    opus-encdec ones are a known opus-decode fallback that stays unreachable
1551    because the web player stops advertising opus when WebCodecs is missing.
1552    """
1553    vendor_dir = STATIC_DIR / "web" / "sendspin-js"
1554    js_files = list(vendor_dir.rglob("*.js"))
1555    assert js_files, "vendored sendspin-js dist is missing"
1556    known_bare_specifiers = {
1557        "opus-encdec/dist/libopus-decoder.js",
1558        "opus-encdec/src/oggOpusDecoder.js",
1559    }
1560    # static imports/re-exports, side-effect imports, and dynamic import()
1561    pattern = re.compile(
1562        r"""(?:import\s*\(\s*|import[^'"()]*?from\s+|import\s+|export[^'"()]*?from\s+)"""
1563        r"""['"]([^'"]+)['"]"""
1564    )
1565    bad: list[str] = []
1566    for js_file in js_files:
1567        for specifier in pattern.findall(js_file.read_text(encoding="utf-8")):
1568            if specifier.startswith("."):
1569                target = (js_file.parent / specifier).resolve()
1570                if not (specifier.endswith(".js") and target.is_file()):
1571                    bad.append(f"{js_file.name}: {specifier}")
1572            elif specifier not in known_bare_specifiers:
1573                bad.append(f"{js_file.name}: bare specifier {specifier}")
1574    assert not bad, f"browser-unloadable import specifiers: {bad}"
1575
1576
1577# --- Queue API ---
1578
1579
1580async def test_queue_unknown_player(http_client: TestClient[Any, Any]) -> None:
1581    """GET /api/queue/{player_id} for unknown player returns empty items."""
1582    resp = await http_client.get("/api/queue/unknown_player")
1583    assert resp.status == 200
1584    data = await resp.json()
1585    assert data["items"] == []
1586    assert data["current_index"] == -1
1587
1588
1589async def test_queue_with_items(
1590    http_client: TestClient[Any, Any],
1591    provider: MSXBridgeProvider,
1592    mass_mock: Mock,
1593) -> None:
1594    """GET /api/queue/{player_id} returns queue items with current_index."""
1595    # Create and register a player
1596    player = MSXPlayer(provider, "msx_queue_test", name="Queue TV", output_format="mp3")
1597    player.update_state = Mock()  # type: ignore[misc,method-assign]
1598
1599    # Set current media
1600    media = MagicMock(spec=PlayerMedia)
1601    media.source_id = "msx_queue_test"
1602    media.queue_item_id = "qi_2"
1603    player._attr_current_media = media
1604
1605    # Mock get_player to return our player
1606    mass_mock.players.get_player = Mock(
1607        side_effect=lambda pid, **_kw: player if pid == "msx_queue_test" else None
1608    )
1609
1610    # Create mock queue items
1611    qi1 = Mock()
1612    qi1.name = "Track 1"
1613    qi1.duration = 180
1614    qi1.image = None
1615    mi1 = Mock()
1616    mi1.name = "Track 1"
1617    mi1.uri = "library://track/1"
1618    mi1.artist_str = "Artist A"
1619    mi1.duration = 180
1620    qi1.media_item = mi1
1621
1622    qi2 = Mock()
1623    qi2.name = "Track 2"
1624    qi2.duration = 240
1625    qi2.image = None
1626    mi2 = Mock()
1627    mi2.name = "Track 2"
1628    mi2.uri = "library://track/2"
1629    mi2.artist_str = "Artist B"
1630    mi2.duration = 240
1631    qi2.media_item = mi2
1632
1633    mass_mock.player_queues.items = Mock(return_value=[qi1, qi2])
1634
1635    # Mock get_item to resolve current track
1636    current_qi = Mock()
1637    current_qi.media_item = mi2
1638    mass_mock.player_queues.get_item = Mock(return_value=current_qi)
1639
1640    resp = await http_client.get("/api/queue/msx_queue_test")
1641    assert resp.status == 200
1642    data = await resp.json()
1643    assert len(data["items"]) == 2
1644    assert data["items"][0]["title"] == "Track 1"
1645    assert data["items"][0]["artist"] == "Artist A"
1646    assert data["items"][1]["title"] == "Track 2"
1647    assert data["current_index"] == 1  # Track 2 is current
1648
1649
1650class _AsyncCtx:
1651    """Async context manager helper for mocking session.get()."""
1652
1653    def __init__(self, obj: object) -> None:
1654        self._obj = obj
1655
1656    async def __aenter__(self) -> object:
1657        return self._obj
1658
1659    async def __aexit__(self, *args: object) -> None:
1660        pass
1661