music-assistant-server

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