/
/
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 "client_id": "client_id",
41 }
42 )
43 assert await auth_manager.initialize(auth_data) is True
44 assert auth_manager.access_token == "token"
45
46
47async def test_initialize_invalid_json(auth_manager: TidalAuthManager) -> None:
48 """Test initialization with invalid JSON."""
49 assert await auth_manager.initialize("invalid") is False
50
51
52async def test_ensure_valid_token_valid(auth_manager: TidalAuthManager) -> None:
53 """Test ensure_valid_token with valid token."""
54 auth_manager._auth_info = {"expires_at": time.time() + 3600}
55 assert await auth_manager.ensure_valid_token() is True
56
57
58async def test_ensure_valid_token_expired(
59 auth_manager: TidalAuthManager, http_session: AsyncMock, config_updater: Mock
60) -> None:
61 """Test ensure_valid_token with expired token."""
62 auth_manager._auth_info = {
63 "expires_at": time.time() - 3600,
64 "refresh_token": "refresh",
65 "client_id": "client_id",
66 }
67
68 # Mock refresh response
69 response = AsyncMock()
70 response.status = 200
71 response.json.return_value = {
72 "access_token": "new_token",
73 "expires_in": 3600,
74 "refresh_token": "new_refresh",
75 }
76 http_session.post.return_value.__aenter__.return_value = response
77
78 assert await auth_manager.ensure_valid_token() is True
79 assert auth_manager.access_token == "new_token"
80 config_updater.assert_called_once()
81
82
83async def test_refresh_token_failure(
84 auth_manager: TidalAuthManager, http_session: AsyncMock
85) -> None:
86 """Test refresh_token failure."""
87 auth_manager._auth_info = {
88 "refresh_token": "refresh",
89 "client_id": "client_id",
90 }
91
92 # Mock refresh response failure
93 response = AsyncMock()
94 response.status = 400
95 response.text.return_value = "Bad Request"
96 http_session.post.return_value.__aenter__.return_value = response
97
98 assert await auth_manager.refresh_token() is False
99
100
101async def test_refresh_token_cooldown(
102 auth_manager: TidalAuthManager, http_session: AsyncMock
103) -> None:
104 """Test a refresh right after a successful one skips the token endpoint."""
105 auth_manager._auth_info = {
106 "refresh_token": "refresh",
107 "client_id": "client_id",
108 }
109
110 response = AsyncMock()
111 response.status = 200
112 response.json.return_value = {
113 "access_token": "new_token",
114 "expires_in": 3600,
115 }
116 http_session.post.return_value.__aenter__.return_value = response
117
118 assert await auth_manager.refresh_token() is True
119 assert await auth_manager.refresh_token() is True
120 assert http_session.post.call_count == 1
121
122
123@patch("music_assistant.providers.tidal.auth_manager.pkce")
124@patch("music_assistant.providers.tidal.auth_manager.app_var")
125async def test_build_pkce_login(mock_app_var: Mock, mock_pkce: Mock) -> None:
126 """The authorize URL carries the PKCE challenge and the params keep the verifier."""
127 mock_pkce.generate_pkce_pair.return_value = ("verifier", "challenge")
128 mock_app_var.side_effect = ["client_id", "client_secret"]
129
130 authorize_url, auth_params = TidalAuthManager.build_pkce_login("HIGH")
131
132 assert authorize_url.startswith("https://login.tidal.com/authorize?")
133 assert "code_challenge=challenge" in authorize_url
134 assert "code_challenge_method=S256" in authorize_url
135 assert auth_params["code_verifier"] == "verifier"
136 assert auth_params["client_unique_key"]
137 assert auth_params["quality"] == "HIGH"
138
139
140async def test_process_pkce_login_success(http_session: AsyncMock) -> None:
141 """Test process_pkce_login success."""
142 auth_params = json.dumps(
143 {
144 "code_verifier": "verifier",
145 "client_unique_key": "key",
146 "client_id": "id",
147 "client_secret": "secret",
148 "quality": "HIGH",
149 }
150 )
151 redirect_url = "https://tidal.com/android/login/auth?code=auth_code"
152
153 # Mock token response
154 token_response = AsyncMock()
155 token_response.status = 200
156 token_response.json.return_value = {
157 "access_token": "access",
158 "refresh_token": "refresh",
159 "expires_in": 3600,
160 }
161
162 # Mock user info response
163 user_response = AsyncMock()
164 user_response.status = 200
165 user_response.json.return_value = {
166 "id": "user_id",
167 "username": "user",
168 }
169
170 http_session.post.return_value.__aenter__.return_value = token_response
171 http_session.get.return_value.__aenter__.return_value = user_response
172
173 result = await TidalAuthManager.process_pkce_login(http_session, auth_params, redirect_url)
174
175 assert result["access_token"] == "access"
176 assert result["id"] == "user_id"
177
178
179async def test_process_pkce_login_missing_code(http_session: AsyncMock) -> None:
180 """Test process_pkce_login missing code."""
181 auth_params = json.dumps(
182 {
183 "code_verifier": "verifier",
184 "client_unique_key": "key",
185 }
186 )
187 redirect_url = "https://tidal.com/android/login/auth"
188
189 with pytest.raises(LoginFailed, match="No authorization code"):
190 await TidalAuthManager.process_pkce_login(http_session, auth_params, redirect_url)
191