/
/
/
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 (
12 ManualAuthenticationHelper,
13 TidalAuthManager,
14)
15
16
17@pytest.fixture
18def http_session() -> AsyncMock:
19 """Return a mock http session."""
20 return AsyncMock(spec=ClientSession)
21
22
23@pytest.fixture
24def config_updater() -> Mock:
25 """Return a mock config updater."""
26 return Mock()
27
28
29@pytest.fixture
30def auth_manager(http_session: AsyncMock, config_updater: Mock) -> TidalAuthManager:
31 """Return a TidalAuthManager instance."""
32 logger = Mock()
33 return TidalAuthManager(http_session, config_updater, logger)
34
35
36async def test_initialize_success(auth_manager: TidalAuthManager) -> None:
37 """Test successful initialization."""
38 auth_data = json.dumps(
39 {
40 "access_token": "token",
41 "refresh_token": "refresh",
42 "expires_at": time.time() + 3600,
43 "client_id": "client_id",
44 }
45 )
46 assert await auth_manager.initialize(auth_data) is True
47 assert auth_manager.access_token == "token"
48
49
50async def test_initialize_invalid_json(auth_manager: TidalAuthManager) -> None:
51 """Test initialization with invalid JSON."""
52 assert await auth_manager.initialize("invalid") is False
53
54
55async def test_ensure_valid_token_valid(auth_manager: TidalAuthManager) -> None:
56 """Test ensure_valid_token with valid token."""
57 auth_manager._auth_info = {"expires_at": time.time() + 3600}
58 assert await auth_manager.ensure_valid_token() is True
59
60
61async def test_ensure_valid_token_expired(
62 auth_manager: TidalAuthManager, http_session: AsyncMock, config_updater: Mock
63) -> None:
64 """Test ensure_valid_token with expired token."""
65 auth_manager._auth_info = {
66 "expires_at": time.time() - 3600,
67 "refresh_token": "refresh",
68 "client_id": "client_id",
69 }
70
71 # Mock refresh response
72 response = AsyncMock()
73 response.status = 200
74 response.json.return_value = {
75 "access_token": "new_token",
76 "expires_in": 3600,
77 "refresh_token": "new_refresh",
78 }
79 http_session.post.return_value.__aenter__.return_value = response
80
81 assert await auth_manager.ensure_valid_token() is True
82 assert auth_manager.access_token == "new_token"
83 config_updater.assert_called_once()
84
85
86async def test_refresh_token_failure(
87 auth_manager: TidalAuthManager, http_session: AsyncMock
88) -> None:
89 """Test refresh_token failure."""
90 auth_manager._auth_info = {
91 "refresh_token": "refresh",
92 "client_id": "client_id",
93 }
94
95 # Mock refresh response failure
96 response = AsyncMock()
97 response.status = 400
98 response.text.return_value = "Bad Request"
99 http_session.post.return_value.__aenter__.return_value = response
100
101 assert await auth_manager.refresh_token() is False
102
103
104@patch("music_assistant.providers.tidal.auth_manager.pkce")
105@patch("music_assistant.providers.tidal.auth_manager.app_var")
106@pytest.mark.usefixtures("auth_manager")
107async def test_generate_auth_url(mock_app_var: Mock, mock_pkce: Mock) -> None:
108 """Test generate_auth_url."""
109 mock_pkce.generate_pkce_pair.return_value = ("verifier", "challenge")
110 mock_app_var.side_effect = ["client_id", "client_secret"]
111
112 mass = Mock()
113 mass.loop.call_soon_threadsafe = Mock()
114 auth_helper = ManualAuthenticationHelper(mass, "session_id")
115
116 result = await TidalAuthManager.generate_auth_url(auth_helper, "HIGH")
117
118 assert "code_verifier" in result
119 assert "client_unique_key" in result
120 mass.loop.call_soon_threadsafe.assert_called_once()
121
122
123async def test_process_pkce_login_success(http_session: AsyncMock) -> None:
124 """Test process_pkce_login success."""
125 auth_params = json.dumps(
126 {
127 "code_verifier": "verifier",
128 "client_unique_key": "key",
129 "client_id": "id",
130 "client_secret": "secret",
131 "quality": "HIGH",
132 }
133 )
134 redirect_url = "https://tidal.com/android/login/auth?code=auth_code"
135
136 # Mock token response
137 token_response = AsyncMock()
138 token_response.status = 200
139 token_response.json.return_value = {
140 "access_token": "access",
141 "refresh_token": "refresh",
142 "expires_in": 3600,
143 }
144
145 # Mock user info response
146 user_response = AsyncMock()
147 user_response.status = 200
148 user_response.json.return_value = {
149 "id": "user_id",
150 "username": "user",
151 }
152
153 http_session.post.return_value.__aenter__.return_value = token_response
154 http_session.get.return_value.__aenter__.return_value = user_response
155
156 result = await TidalAuthManager.process_pkce_login(http_session, auth_params, redirect_url)
157
158 assert result["access_token"] == "access"
159 assert result["id"] == "user_id"
160
161
162async def test_process_pkce_login_missing_code(http_session: AsyncMock) -> None:
163 """Test process_pkce_login missing code."""
164 auth_params = json.dumps(
165 {
166 "code_verifier": "verifier",
167 "client_unique_key": "key",
168 }
169 )
170 redirect_url = "https://tidal.com/android/login/auth"
171
172 with pytest.raises(LoginFailed, match="No authorization code"):
173 await TidalAuthManager.process_pkce_login(http_session, auth_params, redirect_url)
174