music-assistant-server

1"""Tests for the Party Mode endpoints (spec 0001)."""
2
3from __future__ import annotations
4
5import threading
6from typing import TYPE_CHECKING, Any
7from unittest.mock import AsyncMock, Mock
8
9import segno
10
11from music_assistant.providers.msx_bridge.http_server import _render_qr
12
13if TYPE_CHECKING:
14    import pytest
15    from aiohttp.test_utils import TestClient
16
17JOIN_URL = "http://ma.local:8095/?join=ABC123"
18
19
20def _party_mock(
21    url: str | None = JOIN_URL,
22    name: str | None = "My Party",
23    qr_text: str | None = "Scan to join!",
24) -> Mock:
25    """Return a mock Party plugin provider."""
26    party = Mock()
27    party.get_party_url = AsyncMock(return_value=url)
28    config = Mock()
29    config.party_name = name
30    config.qr_text = qr_text
31    party.get_party_config = AsyncMock(return_value=config)
32    return party
33
34
35# --- /api/party status endpoint ---
36
37
38async def test_party_status_no_plugin(http_client: TestClient[Any, Any]) -> None:
39    """GET /api/party should report inactive when the Party plugin is not loaded."""
40    resp = await http_client.get("/api/party")
41    assert resp.status == 200
42    data = await resp.json()
43    assert data == {"active": False}
44
45
46async def test_party_status_guest_access_off(
47    http_client: TestClient[Any, Any], mass_mock: Mock
48) -> None:
49    """GET /api/party should report inactive when guest access is disabled."""
50    mass_mock.get_provider = Mock(return_value=_party_mock(url=None))
51    resp = await http_client.get("/api/party")
52    assert resp.status == 200
53    data = await resp.json()
54    assert data == {"active": False}
55
56
57async def test_party_status_active(http_client: TestClient[Any, Any], mass_mock: Mock) -> None:
58    """GET /api/party should return party name, caption and QR URL when active."""
59    mass_mock.get_provider = Mock(return_value=_party_mock())
60    resp = await http_client.get("/api/party")
61    assert resp.status == 200
62    data = await resp.json()
63    assert data["active"] is True
64    assert data["name"] == "My Party"
65    assert data["qr_text"] == "Scan to join!"
66    # relative so the kiosk works behind reverse proxies / mapped ports
67    assert data["qr_url"] == "/api/party/qr.svg"
68    # opaque version so clients refetch the QR only when the join code rotates
69    assert data["qr_version"]
70    assert JOIN_URL not in data["qr_version"]
71    mass_mock.get_provider.assert_called_with("party")
72
73
74async def test_party_status_survives_plugin_errors(
75    http_client: TestClient[Any, Any], mass_mock: Mock
76) -> None:
77    """A raising Party plugin must degrade to inactive, not a 500."""
78    party = _party_mock()
79    party.get_party_url = AsyncMock(side_effect=RuntimeError("guest access broken"))
80    mass_mock.get_provider = Mock(return_value=party)
81    resp = await http_client.get("/api/party")
82    assert resp.status == 200
83    assert await resp.json() == {"active": False}
84
85
86async def test_party_status_is_cached(http_client: TestClient[Any, Any], mass_mock: Mock) -> None:
87    """Repeated status requests within the cache TTL must not re-query the plugin."""
88    party = _party_mock()
89    mass_mock.get_provider = Mock(return_value=party)
90    resp1 = await http_client.get("/api/party")
91    resp2 = await http_client.get("/api/party")
92    assert resp1.status == resp2.status == 200
93    party.get_party_url.assert_awaited_once()
94
95
96async def test_party_status_active_without_custom_texts(
97    http_client: TestClient[Any, Any], mass_mock: Mock
98) -> None:
99    """GET /api/party should tolerate unset party name and caption."""
100    mass_mock.get_provider = Mock(return_value=_party_mock(name=None, qr_text=None))
101    resp = await http_client.get("/api/party")
102    assert resp.status == 200
103    data = await resp.json()
104    assert data["active"] is True
105    assert data["name"] is None
106    assert data["qr_text"] is None
107
108
109async def test_party_status_does_not_leak_join_url(
110    http_client: TestClient[Any, Any], mass_mock: Mock
111) -> None:
112    """GET /api/party must not include the raw join URL in the response."""
113    mass_mock.get_provider = Mock(return_value=_party_mock())
114    resp = await http_client.get("/api/party")
115    body = await resp.text()
116    assert JOIN_URL not in body
117
118
119# --- /api/party/qr.svg image endpoint ---
120
121
122async def test_party_qr_svg_active(http_client: TestClient[Any, Any], mass_mock: Mock) -> None:
123    """GET /api/party/qr.svg should return an SVG QR code of the join URL."""
124    mass_mock.get_provider = Mock(return_value=_party_mock())
125    resp = await http_client.get("/api/party/qr.svg")
126    assert resp.status == 200
127    assert "image/svg+xml" in resp.headers["Content-Type"]
128    body = await resp.text()
129    assert "<svg" in body
130
131
132async def test_party_qr_not_cached(http_client: TestClient[Any, Any], mass_mock: Mock) -> None:
133    """The QR image must not be cacheable — the join code can rotate."""
134    mass_mock.get_provider = Mock(return_value=_party_mock())
135    resp = await http_client.get("/api/party/qr.svg")
136    assert resp.status == 200
137    assert "no-store" in resp.headers.get("Cache-Control", "")
138
139
140async def test_party_qr_404_no_plugin(http_client: TestClient[Any, Any]) -> None:
141    """GET /api/party/qr.svg should 404 when the Party plugin is not loaded."""
142    resp = await http_client.get("/api/party/qr.svg")
143    assert resp.status == 404
144
145
146async def test_party_qr_404_guest_access_off(
147    http_client: TestClient[Any, Any], mass_mock: Mock
148) -> None:
149    """GET /api/party/qr.svg should 404 when guest access is disabled."""
150    mass_mock.get_provider = Mock(return_value=_party_mock(url=None))
151    resp = await http_client.get("/api/party/qr.svg")
152    assert resp.status == 404
153
154
155async def test_party_qr_png(http_client: TestClient[Any, Any], mass_mock: Mock) -> None:
156    """GET /api/party/qr.png should return a PNG QR for MSX pages on TVs without SVG."""
157    mass_mock.get_provider = Mock(return_value=_party_mock())
158    resp = await http_client.get("/api/party/qr.png")
159    assert resp.status == 200
160    assert "image/png" in resp.headers["Content-Type"]
161    body = await resp.read()
162    assert body.startswith(b"\x89PNG")
163
164
165async def test_party_qr_png_404_inactive(http_client: TestClient[Any, Any]) -> None:
166    """GET /api/party/qr.png should 404 when no party is active."""
167    resp = await http_client.get("/api/party/qr.png")
168    assert resp.status == 404
169
170
171def test_render_qr_is_memoized(monkeypatch: pytest.MonkeyPatch) -> None:
172    """Repeated QR renders for the same join URL and kind reuse the cached bytes."""
173    calls: list[int] = []
174    real_make = segno.make
175
176    def _tracking_make(*args: Any, **kwargs: Any) -> Any:
177        calls.append(1)
178        return real_make(*args, **kwargs)
179
180    monkeypatch.setattr(segno, "make", _tracking_make)
181    _render_qr.cache_clear()
182    first = _render_qr(JOIN_URL, "png")
183    second = _render_qr(JOIN_URL, "png")
184    assert first == second
185    assert len(calls) == 1
186
187
188async def test_party_qr_render_runs_off_event_loop(
189    http_client: TestClient[Any, Any], mass_mock: Mock, monkeypatch: pytest.MonkeyPatch
190) -> None:
191    """QR generation must run in a worker thread, never on the event loop."""
192    mass_mock.get_provider = Mock(return_value=_party_mock())
193    loop_thread = threading.get_ident()
194    make_threads: list[int] = []
195    real_make = segno.make
196
197    def _tracking_make(*args: Any, **kwargs: Any) -> Any:
198        make_threads.append(threading.get_ident())
199        return real_make(*args, **kwargs)
200
201    monkeypatch.setattr(segno, "make", _tracking_make)
202    resp = await http_client.get("/api/party/qr.png")
203    assert resp.status == 200
204    assert make_threads
205    assert loop_thread not in make_threads
206
207
208# --- /msx/party.json native MSX page ---
209
210
211async def test_msx_party_page_active(http_client: TestClient[Any, Any], mass_mock: Mock) -> None:
212    """GET /msx/party.json should show the QR image item when a party is active."""
213    mass_mock.get_provider = Mock(return_value=_party_mock())
214    resp = await http_client.get("/msx/party.json")
215    assert resp.status == 200
216    data = await resp.json()
217    assert data["headline"] == "My Party"
218    items = data["items"]
219    # PNG, not SVG: MSX image slots on older TV engines cannot decode SVG
220    assert any(item.get("image", "").endswith("/api/party/qr.png") for item in items)
221    body = await resp.text()
222    assert JOIN_URL not in body
223
224
225async def test_msx_party_page_refreshes_player_activity(
226    http_client: TestClient[Any, Any], mass_mock: Mock
227) -> None:
228    """Viewing the party page must register/refresh the player like other MSX pages."""
229    resp = await http_client.get("/msx/party.json?device_id=partytv")
230    assert resp.status == 200
231    mass_mock.players.register.assert_awaited()
232
233
234async def test_msx_party_page_inactive(http_client: TestClient[Any, Any]) -> None:
235    """GET /msx/party.json should render a page without a QR when no party is active."""
236    resp = await http_client.get("/msx/party.json")
237    assert resp.status == 200
238    data = await resp.json()
239    assert not any("/api/party/qr.svg" in (item.get("image") or "") for item in data["items"])
240
241
242# --- MSX menu entry ---
243
244
245async def test_msx_menu_shows_party_when_active(
246    http_client: TestClient[Any, Any], mass_mock: Mock
247) -> None:
248    """The MSX menu should include a Party entry pointing at the party page when active."""
249    mass_mock.get_provider = Mock(return_value=_party_mock())
250    resp = await http_client.get("/msx/menu.json")
251    assert resp.status == 200
252    data = await resp.json()
253    party_items = [i for i in data["items"] if i.get("label") == "Party"]
254    assert len(party_items) == 1
255    assert "/msx/party.json" in party_items[0]["content"]
256
257
258async def test_msx_menu_hides_party_when_inactive(http_client: TestClient[Any, Any]) -> None:
259    """The MSX menu should not show a Party entry when no party is active."""
260    resp = await http_client.get("/msx/menu.json")
261    assert resp.status == 200
262    data = await resp.json()
263    assert not any(i.get("label") == "Party" for i in data["items"])
264
265
266async def test_msx_menu_survives_plugin_errors(
267    http_client: TestClient[Any, Any], mass_mock: Mock
268) -> None:
269    """A raising Party plugin must not break the main menu."""
270    party = _party_mock()
271    party.get_party_url = AsyncMock(side_effect=RuntimeError("guest access broken"))
272    mass_mock.get_provider = Mock(return_value=party)
273    resp = await http_client.get("/msx/menu.json")
274    assert resp.status == 200
275    data = await resp.json()
276    assert any(i.get("label") == "Albums" for i in data["items"])
277    assert not any(i.get("label") == "Party" for i in data["items"])
278
279
280async def test_tv_plugin_menu_has_party_entry(http_client: TestClient[Any, Any]) -> None:
281    """The TV interaction plugin's client-side menu must link to the party page."""
282    resp = await http_client.get("/msx/plugin.html")
283    assert resp.status == 200
284    body = await resp.text()
285    assert "/msx/party.json" in body
286