/
/
/
1"""Tests for the short-lived tokens that address the preview endpoint."""
2
3from collections.abc import AsyncGenerator
4from typing import TYPE_CHECKING, cast
5from unittest.mock import AsyncMock, MagicMock, patch
6
7import pytest
8from aiohttp import web
9
10from music_assistant.controllers.webserver.controller import (
11 MAX_PREVIEW_TOKENS,
12 PREVIEW_TOKEN_TTL,
13 WebserverController,
14)
15
16if TYPE_CHECKING:
17 from music_assistant_models.config_entries import CoreConfig
18
19
20@pytest.fixture
21def clock(monkeypatch: pytest.MonkeyPatch) -> list[float]:
22 """Yield a single-element list holding the monotonic time the controller reads."""
23 now = [1000.0]
24 monkeypatch.setattr(
25 "music_assistant.controllers.webserver.controller.time.monotonic",
26 lambda: now[0],
27 )
28 return now
29
30
31@pytest.fixture
32def webserver(mock_mass: MagicMock) -> WebserverController:
33 """Create a WebserverController with a resolved base url."""
34 webserver = WebserverController(mock_mass)
35 config = MagicMock()
36 config.get_value.return_value = False
37 webserver.config = cast("CoreConfig", config)
38 webserver.publish_port = 8095
39 webserver._auto_base_url = "http://192.168.1.5:8095"
40 return webserver
41
42
43def test_preview_url_carries_a_token_and_not_the_item(webserver: WebserverController) -> None:
44 """The url identifies the clip by token, so it reveals no provider or item."""
45 url = webserver.create_preview_url("spotify--abc", "track123")
46 assert "/preview?token=" in url
47 assert "spotify--abc" not in url
48 assert "track123" not in url
49
50
51def test_preview_url_is_relative(webserver: WebserverController) -> None:
52 """Clients reach this server on their own address, which we cannot guess for them."""
53 url = webserver.create_preview_url("spotify--abc", "track123")
54 assert url.startswith("/preview?")
55 assert "://" not in url
56
57
58def test_each_mint_is_a_fresh_token(webserver: WebserverController) -> None:
59 """Handing out a new url must not invalidate one already in use."""
60 first = webserver.create_preview_url("spotify--abc", "track123").split("token=")[1]
61 second = webserver.create_preview_url("spotify--abc", "track456").split("token=")[1]
62 assert first != second
63 assert webserver._resolve_preview_token(first) == ("spotify--abc", "track123")
64 assert webserver._resolve_preview_token(second) == ("spotify--abc", "track456")
65
66
67def test_store_is_capped(webserver: WebserverController) -> None:
68 """A chatty client cannot grow the store without bound."""
69 for index in range(MAX_PREVIEW_TOKENS + 50):
70 webserver.create_preview_url("spotify--abc", f"track{index}")
71 assert len(webserver._preview_tokens) <= MAX_PREVIEW_TOKENS
72
73
74def test_sweep_keeps_tokens_that_are_still_valid(
75 webserver: WebserverController, clock: list[float]
76) -> None:
77 """Clearing out expired tokens must not take live ones with it."""
78 stale = webserver.create_preview_url("spotify--abc", "old").split("token=")[1]
79 clock[0] += PREVIEW_TOKEN_TTL - 1
80 live = webserver.create_preview_url("spotify--abc", "new").split("token=")[1]
81 clock[0] += 2
82 webserver.create_preview_url("spotify--abc", "newest")
83 assert stale not in webserver._preview_tokens
84 assert webserver._resolve_preview_token(live) == ("spotify--abc", "new")
85
86
87def test_token_resolves_to_the_item_it_was_minted_for(webserver: WebserverController) -> None:
88 """A freshly minted token grants exactly the provider and item it was created with."""
89 url = webserver.create_preview_url("spotify--abc", "track123")
90 token = url.split("token=")[1]
91 assert webserver._resolve_preview_token(token) == ("spotify--abc", "track123")
92
93
94def test_token_stays_valid_for_repeated_requests(webserver: WebserverController) -> None:
95 """Players re-request a media url they already opened, so a token is not single-use."""
96 token = webserver.create_preview_url("spotify--abc", "track123").split("token=")[1]
97 assert webserver._resolve_preview_token(token) is not None
98 assert webserver._resolve_preview_token(token) is not None
99
100
101@pytest.mark.parametrize("token", ["", "not-a-real-token"])
102def test_unknown_token_is_refused(webserver: WebserverController, token: str) -> None:
103 """A token that was never minted grants nothing."""
104 assert webserver._resolve_preview_token(token) is None
105
106
107def test_expired_token_is_refused_and_dropped(
108 webserver: WebserverController, clock: list[float]
109) -> None:
110 """A token past its lifetime no longer resolves and leaves no entry behind."""
111 token = webserver.create_preview_url("spotify--abc", "track123").split("token=")[1]
112 clock[0] += PREVIEW_TOKEN_TTL + 1
113 assert webserver._resolve_preview_token(token) is None
114 assert token not in webserver._preview_tokens
115
116
117def test_minting_sweeps_expired_tokens(webserver: WebserverController, clock: list[float]) -> None:
118 """Expired tokens are cleared out as new ones are handed out."""
119 stale = webserver.create_preview_url("spotify--abc", "track123").split("token=")[1]
120 clock[0] += PREVIEW_TOKEN_TTL + 1
121 webserver.create_preview_url("spotify--abc", "track456")
122 assert stale not in webserver._preview_tokens
123 assert len(webserver._preview_tokens) == 1
124
125
126async def test_serve_preview_stream_refuses_an_unknown_token(
127 webserver: WebserverController,
128) -> None:
129 """The endpoint no longer streams for a caller that just names a provider and item."""
130 request = MagicMock()
131 request.query = {"provider": "spotify--abc", "item_id": "track123"}
132 with pytest.raises(web.HTTPNotFound):
133 await webserver.serve_preview_stream(request)
134
135
136async def test_serve_preview_stream_serves_the_item_the_token_was_minted_for(
137 webserver: WebserverController, monkeypatch: pytest.MonkeyPatch
138) -> None:
139 """A valid token streams that item, and nothing from the query string is consulted."""
140
141 async def _clip() -> AsyncGenerator[bytes]:
142 yield b"audio"
143
144 get_preview_stream = MagicMock(return_value=_clip())
145 monkeypatch.setattr(webserver.mass.streams, "get_preview_stream", get_preview_stream)
146 token = webserver.create_preview_url("spotify--abc", "track123").split("token=")[1]
147 request = MagicMock()
148 # a mismatching provider/item in the query must be ignored entirely
149 request.query = {"token": token, "provider": "tidal--xyz", "item_id": "other"}
150 response = MagicMock()
151 response.prepare = AsyncMock()
152 response.write = AsyncMock()
153 with patch(
154 "music_assistant.controllers.webserver.controller.web.StreamResponse",
155 MagicMock(return_value=response),
156 ):
157 await webserver.serve_preview_stream(request)
158 get_preview_stream.assert_called_once_with("spotify--abc", "track123")
159 response.write.assert_awaited_once_with(b"audio")
160