music-assistant-server

1"""Test Tidal Auth Manager."""
2
3import json
4import time
5from unittest.mock import AsyncMock, Mock, patch
6
7import pytest
8from aiohttp import ClientSession
9from music_assistant_models.errors import LoginFailed
10
11from music_assistant.providers.tidal.auth_manager import TidalAuthManager
12
13
14@pytest.fixture
15def http_session() -> AsyncMock:
16    """Return a mock http session."""
17    return AsyncMock(spec=ClientSession)
18
19
20@pytest.fixture
21def config_updater() -> Mock:
22    """Return a mock config updater."""
23    return Mock()
24
25
26@pytest.fixture
27def auth_manager(http_session: AsyncMock, config_updater: Mock) -> TidalAuthManager:
28    """Return a TidalAuthManager instance."""
29    logger = Mock()
30    return TidalAuthManager(http_session, config_updater, logger)
31
32
33async def test_initialize_success(auth_manager: TidalAuthManager) -> None:
34    """Test successful initialization."""
35    auth_data = json.dumps(
36        {
37            "access_token": "token",
38            "refresh_token": "refresh",
39            "expires_at": time.time() + 3600,
40        }
41    )
42    assert await auth_manager.initialize(auth_data) is True
43    assert auth_manager.access_token == "token"
44
45
46async def test_initialize_invalid_json(auth_manager: TidalAuthManager) -> None:
47    """Test initialization with invalid JSON."""
48    assert await auth_manager.initialize("invalid") is False
49
50
51async def test_ensure_valid_token_valid(auth_manager: TidalAuthManager) -> None:
52    """Test ensure_valid_token with valid token."""
53    auth_manager._auth_info = {"expires_at": time.time() + 3600}
54    assert await auth_manager.ensure_valid_token() is True
55
56
57async def test_ensure_valid_token_expired(
58    auth_manager: TidalAuthManager, http_session: AsyncMock, config_updater: Mock
59) -> None:
60    """Test ensure_valid_token with expired token."""
61    auth_manager._auth_info = {
62        "expires_at": time.time() - 3600,
63        "refresh_token": "refresh",
64    }
65
66    # Mock refresh response
67    response = AsyncMock()
68    response.status = 200
69    response.json.return_value = {
70        "access_token": "new_token",
71        "expires_in": 3600,
72        "refresh_token": "new_refresh",
73    }
74    http_session.post.return_value.__aenter__.return_value = response
75
76    assert await auth_manager.ensure_valid_token() is True
77    assert auth_manager.access_token == "new_token"
78    config_updater.assert_called_once()
79
80
81async def test_refresh_token_failure(
82    auth_manager: TidalAuthManager, http_session: AsyncMock
83) -> None:
84    """Test refresh_token failure."""
85    auth_manager._auth_info = {
86        "refresh_token": "refresh",
87    }
88
89    # Mock refresh response failure
90    response = AsyncMock()
91    response.status = 400
92    response.text.return_value = "Bad Request"
93    http_session.post.return_value.__aenter__.return_value = response
94
95    assert await auth_manager.refresh_token() is False
96
97
98async def test_refresh_token_cooldown(
99    auth_manager: TidalAuthManager, http_session: AsyncMock
100) -> None:
101    """Test a refresh right after a successful one skips the token endpoint."""
102    auth_manager._auth_info = {
103        "refresh_token": "refresh",
104    }
105
106    response = AsyncMock()
107    response.status = 200
108    response.json.return_value = {
109        "access_token": "new_token",
110        "expires_in": 3600,
111    }
112    http_session.post.return_value.__aenter__.return_value = response
113
114    assert await auth_manager.refresh_token() is True
115    assert await auth_manager.refresh_token() is True
116    assert http_session.post.call_count == 1
117
118
119@patch("music_assistant.providers.tidal.auth_manager.app_var")
120async def test_start_device_login(mock_app_var: Mock, http_session: AsyncMock) -> None:
121    """start_device_login returns the device authorization response."""
122    mock_app_var.side_effect = ["client_id", "client_secret"]
123    response = AsyncMock()
124    response.status = 200
125    response.json.return_value = {
126        "deviceCode": "dev",
127        "userCode": "ABCDE",
128        "verificationUri": "link.tidal.com",
129        "interval": 2,
130        "expiresIn": 300,
131    }
132    http_session.post.return_value.__aenter__.return_value = response
133
134    device = await TidalAuthManager.start_device_login(http_session)
135
136    assert device["deviceCode"] == "dev"
137    assert device["userCode"] == "ABCDE"
138
139
140@patch("music_assistant.providers.tidal.auth_manager.app_var")
141async def test_start_device_login_failure(mock_app_var: Mock, http_session: AsyncMock) -> None:
142    """start_device_login raises when the client is rejected (e.g. wrong client type)."""
143    mock_app_var.side_effect = ["client_id", "client_secret"]
144    response = AsyncMock()
145    response.status = 400
146    response.json.return_value = {"error": "invalid_request"}
147    http_session.post.return_value.__aenter__.return_value = response
148
149    with pytest.raises(LoginFailed):
150        await TidalAuthManager.start_device_login(http_session)
151
152
153@patch("music_assistant.providers.tidal.auth_manager.app_var")
154async def test_start_device_login_non_json_body(
155    mock_app_var: Mock, http_session: AsyncMock
156) -> None:
157    """A non-JSON body (e.g. proxy HTML error page) surfaces as LoginFailed."""
158    mock_app_var.side_effect = ["client_id", "client_secret"]
159    response = AsyncMock()
160    response.status = 502
161    response.json.side_effect = json.JSONDecodeError("Expecting value", "<html>", 0)
162    http_session.post.return_value.__aenter__.return_value = response
163
164    with pytest.raises(LoginFailed, match="non-JSON"):
165        await TidalAuthManager.start_device_login(http_session)
166
167
168@patch("music_assistant.providers.tidal.auth_manager.app_var")
169async def test_poll_device_login_success(mock_app_var: Mock, http_session: AsyncMock) -> None:
170    """poll_device_login returns auth data with user info and an absolute expiry."""
171    mock_app_var.side_effect = ["client_id", "client_secret"]
172    device = {"deviceCode": "dev", "interval": 0}
173
174    token_response = AsyncMock()
175    token_response.status = 200
176    token_response.json.return_value = {
177        "access_token": "access",
178        "refresh_token": "refresh",
179        "expires_in": 3600,
180    }
181    user_response = AsyncMock()
182    user_response.status = 200
183    user_response.json.return_value = {"userId": "user_id", "countryCode": "US"}
184
185    http_session.post.return_value.__aenter__.return_value = token_response
186    http_session.get.return_value.__aenter__.return_value = user_response
187
188    result = await TidalAuthManager.poll_device_login(http_session, device)
189
190    assert result["access_token"] == "access"
191    assert result["userId"] == "user_id"
192    assert result["expires_at"] > time.time()
193
194
195@patch("music_assistant.providers.tidal.auth_manager.app_var")
196async def test_poll_device_login_waits_for_pending(
197    mock_app_var: Mock, http_session: AsyncMock
198) -> None:
199    """poll_device_login keeps polling while authorization is pending, then succeeds."""
200    mock_app_var.side_effect = ["client_id", "client_secret"]
201    device = {"deviceCode": "dev", "interval": 0}
202
203    pending = AsyncMock()
204    pending.status = 400
205    pending.json.return_value = {"error": "authorization_pending"}
206    ok = AsyncMock()
207    ok.status = 200
208    ok.json.return_value = {
209        "access_token": "access",
210        "refresh_token": "refresh",
211        "expires_in": 3600,
212    }
213    user_response = AsyncMock()
214    user_response.status = 200
215    user_response.json.return_value = {"userId": "user_id"}
216
217    http_session.post.return_value.__aenter__.side_effect = [pending, ok]
218    http_session.get.return_value.__aenter__.return_value = user_response
219
220    result = await TidalAuthManager.poll_device_login(http_session, device)
221
222    assert result["access_token"] == "access"
223    assert http_session.post.call_count == 2
224
225
226@patch("music_assistant.providers.tidal.auth_manager.app_var")
227async def test_poll_device_login_terminal_error(
228    mock_app_var: Mock, http_session: AsyncMock
229) -> None:
230    """poll_device_login raises on a terminal (non-pending) error such as denial."""
231    mock_app_var.side_effect = ["client_id", "client_secret"]
232    device = {"deviceCode": "dev", "interval": 0}
233
234    response = AsyncMock()
235    response.status = 400
236    response.json.return_value = {"error": "access_denied"}
237    http_session.post.return_value.__aenter__.return_value = response
238
239    with pytest.raises(LoginFailed):
240        await TidalAuthManager.poll_device_login(http_session, device)
241
242
243async def test_finalize_login_missing_tokens(http_session: AsyncMock) -> None:
244    """_finalize_login raises when the token response lacks the required tokens."""
245    with pytest.raises(LoginFailed):
246        await TidalAuthManager._finalize_login(http_session, {"access_token": "only_access"})
247
248
249async def test_finalize_login_sessions_error(http_session: AsyncMock) -> None:
250    """_finalize_login raises when the sessions lookup returns a non-200."""
251    response = AsyncMock()
252    response.status = 500
253    response.text.return_value = "server error"
254    http_session.get.return_value.__aenter__.return_value = response
255
256    with pytest.raises(LoginFailed):
257        await TidalAuthManager._finalize_login(
258            http_session, {"access_token": "access", "refresh_token": "refresh"}
259        )
260
261
262@patch("music_assistant.providers.tidal.auth_manager.asyncio.sleep")
263@patch("music_assistant.providers.tidal.auth_manager.app_var")
264async def test_poll_device_login_expired_token_keeps_polling(
265    mock_app_var: Mock, mock_sleep: AsyncMock, http_session: AsyncMock
266) -> None:
267    """
268    An expired_token response keeps polling; the step deadline re-mints the code.
269
270    Tidal's clock may declare the code expired moments before the setup flow's own
271    deadline fires, so treating it as terminal would abort a flow about to recover.
272    """
273    mock_app_var.side_effect = ["client_id", "client_secret"]
274    device = {"deviceCode": "dev", "interval": 0}
275
276    expired = AsyncMock()
277    expired.status = 400
278    expired.json.return_value = {"error": "expired_token"}
279    ok = AsyncMock()
280    ok.status = 200
281    ok.json.return_value = {"access_token": "a", "refresh_token": "r", "expires_in": 3600}
282    user = AsyncMock()
283    user.status = 200
284    user.json.return_value = {"userId": "u"}
285
286    http_session.post.return_value.__aenter__.side_effect = [expired, ok]
287    http_session.get.return_value.__aenter__.return_value = user
288
289    result = await TidalAuthManager.poll_device_login(http_session, device)
290
291    assert result["access_token"] == "a"
292    assert http_session.post.call_count == 2
293    assert mock_sleep.await_count == 2
294
295
296@patch("music_assistant.providers.tidal.auth_manager.asyncio.sleep")
297@patch("music_assistant.providers.tidal.auth_manager.app_var")
298async def test_poll_device_login_survives_gateway_error(
299    mock_app_var: Mock, mock_sleep: AsyncMock, http_session: AsyncMock
300) -> None:
301    """A transient 5xx mid-poll keeps polling instead of aborting the login."""
302    mock_app_var.side_effect = ["client_id", "client_secret"]
303    device = {"deviceCode": "dev", "interval": 0}
304
305    gateway_error = AsyncMock()
306    gateway_error.status = 502
307    gateway_error.json.side_effect = json.JSONDecodeError("Expecting value", "<html>", 0)
308    ok = AsyncMock()
309    ok.status = 200
310    ok.json.return_value = {"access_token": "a", "refresh_token": "r", "expires_in": 3600}
311    user = AsyncMock()
312    user.status = 200
313    user.json.return_value = {"userId": "u"}
314
315    http_session.post.return_value.__aenter__.side_effect = [gateway_error, ok]
316    http_session.get.return_value.__aenter__.return_value = user
317
318    result = await TidalAuthManager.poll_device_login(http_session, device)
319
320    assert result["access_token"] == "a"
321    assert http_session.post.call_count == 2
322    # slept before each poll: the 502 waited for the next interval, no tight retry
323    assert mock_sleep.await_count == 2
324
325
326@patch("music_assistant.providers.tidal.auth_manager.asyncio.sleep")
327@patch("music_assistant.providers.tidal.auth_manager.app_var")
328async def test_poll_device_login_slow_down(
329    mock_app_var: Mock, mock_sleep: AsyncMock, http_session: AsyncMock
330) -> None:
331    """poll_device_login backs off on slow_down and then succeeds."""
332    mock_app_var.side_effect = ["client_id", "client_secret"]
333    device = {"deviceCode": "dev", "interval": 0}
334
335    slow = AsyncMock()
336    slow.status = 400
337    slow.json.return_value = {"error": "slow_down"}
338    ok = AsyncMock()
339    ok.status = 200
340    ok.json.return_value = {"access_token": "a", "refresh_token": "r", "expires_in": 3600}
341    user = AsyncMock()
342    user.status = 200
343    user.json.return_value = {"userId": "u"}
344
345    http_session.post.return_value.__aenter__.side_effect = [slow, ok]
346    http_session.get.return_value.__aenter__.return_value = user
347
348    result = await TidalAuthManager.poll_device_login(http_session, device)
349
350    assert result["access_token"] == "a"
351    assert http_session.post.call_count == 2
352    # slept once before each poll (the second interval was bumped by slow_down)
353    assert mock_sleep.await_count == 2
354