/
/
/
1"""Test Tidal API Client."""
2
3from typing import Any
4from unittest.mock import AsyncMock, MagicMock, Mock
5
6import pytest
7from aiohttp import ClientResponse
8from music_assistant_models.errors import (
9 LoginFailed,
10 MediaNotFoundError,
11 RetriesExhausted,
12)
13
14from music_assistant.providers.tidal.api_client import TidalAPIClient
15
16
17@pytest.fixture
18def provider_mock() -> Mock:
19 """Return a mock provider."""
20 provider = Mock()
21 provider.auth = AsyncMock()
22 provider.auth.ensure_valid_token.return_value = True
23 provider.auth.access_token = "token"
24 provider.auth.session_id = "session"
25 provider.auth.country_code = "US"
26 provider.mass = Mock()
27 provider.mass.http_session = AsyncMock()
28 provider.mass.metadata.locale = "en_US"
29 provider.logger = Mock()
30 return provider
31
32
33@pytest.fixture
34def api_client(provider_mock: Mock) -> TidalAPIClient:
35 """Return a TidalAPIClient instance."""
36 return TidalAPIClient(provider_mock)
37
38
39async def test_get_success(api_client: TidalAPIClient, provider_mock: Mock) -> None:
40 """Test successful GET request."""
41 response = AsyncMock(spec=ClientResponse)
42 response.status = 200
43 response.json.return_value = {"data": "test"}
44
45 # Create a mock that acts as an async context manager
46 request_ctx = AsyncMock()
47 request_ctx.__aenter__.return_value = response
48
49 # The request method itself should be a MagicMock (not AsyncMock)
50 # that returns the context manager
51 provider_mock.mass.http_session.request = MagicMock(return_value=request_ctx)
52
53 result = await api_client.get("test/endpoint")
54 assert result == {"data": "test"}
55
56
57async def test_get_401_error(api_client: TidalAPIClient, provider_mock: Mock) -> None:
58 """Test GET request with 401 error and a failing token refresh."""
59 response = AsyncMock(spec=ClientResponse)
60 response.status = 401
61
62 request_ctx = AsyncMock()
63 request_ctx.__aenter__.return_value = response
64 provider_mock.mass.http_session.request = MagicMock(return_value=request_ctx)
65 provider_mock.auth.refresh_token.return_value = False
66
67 with pytest.raises(LoginFailed):
68 await api_client.get("test/endpoint")
69
70 provider_mock.auth.refresh_token.assert_called_once()
71
72
73async def test_get_401_refreshes_token_and_retries(
74 api_client: TidalAPIClient, provider_mock: Mock
75) -> None:
76 """Test that a 401 response forces a token refresh and retries the request once."""
77 response_401 = AsyncMock(spec=ClientResponse)
78 response_401.status = 401
79
80 response_ok = AsyncMock(spec=ClientResponse)
81 response_ok.status = 200
82 response_ok.json.return_value = {"data": "test"}
83
84 ctx1 = AsyncMock()
85 ctx1.__aenter__.return_value = response_401
86 ctx2 = AsyncMock()
87 ctx2.__aenter__.return_value = response_ok
88 provider_mock.mass.http_session.request = MagicMock(side_effect=[ctx1, ctx2])
89 provider_mock.auth.refresh_token.return_value = True
90
91 result = await api_client.get("test/endpoint")
92
93 assert result == {"data": "test"}
94 provider_mock.auth.refresh_token.assert_called_once()
95 assert provider_mock.mass.http_session.request.call_count == 2
96
97
98async def test_get_404_error(api_client: TidalAPIClient, provider_mock: Mock) -> None:
99 """Test GET request with 404 error."""
100 response = AsyncMock(spec=ClientResponse)
101 response.status = 404
102 response.url = "http://test/endpoint"
103
104 request_ctx = AsyncMock()
105 request_ctx.__aenter__.return_value = response
106 provider_mock.mass.http_session.request = MagicMock(return_value=request_ctx)
107
108 with pytest.raises(MediaNotFoundError):
109 await api_client.get("test/endpoint")
110
111
112async def test_get_429_error(api_client: TidalAPIClient, provider_mock: Mock) -> None:
113 """Test GET request with 429 error."""
114 response = AsyncMock(spec=ClientResponse)
115 response.status = 429
116 response.headers = {"Retry-After": "10"}
117
118 request_ctx = AsyncMock()
119 request_ctx.__aenter__.return_value = response
120 provider_mock.mass.http_session.request = MagicMock(return_value=request_ctx)
121
122 with pytest.raises(RetriesExhausted):
123 await api_client.get("test/endpoint")
124
125
126async def test_post_success(api_client: TidalAPIClient, provider_mock: Mock) -> None:
127 """Test successful POST request."""
128 response = AsyncMock(spec=ClientResponse)
129 response.status = 200
130 response.json.return_value = {"success": True}
131
132 request_ctx = AsyncMock()
133 request_ctx.__aenter__.return_value = response
134 provider_mock.mass.http_session.request = MagicMock(return_value=request_ctx)
135
136 result = await api_client.post("test/endpoint", data={"key": "value"})
137 assert result == {"success": True}
138
139
140async def test_paginate(api_client: TidalAPIClient, provider_mock: Mock) -> None:
141 """Test pagination."""
142 # Mock first page response
143 response1 = AsyncMock(spec=ClientResponse)
144 response1.status = 200
145 response1.json.return_value = {"items": [{"id": 1}, {"id": 2}], "totalNumberOfItems": 4}
146
147 # Mock second page response
148 response2 = AsyncMock(spec=ClientResponse)
149 response2.status = 200
150 response2.json.return_value = {"items": [{"id": 3}, {"id": 4}], "totalNumberOfItems": 4}
151
152 # Mock empty response to stop iteration
153 response3 = AsyncMock(spec=ClientResponse)
154 response3.status = 200
155 response3.json.return_value = {"items": []}
156
157 ctx1 = AsyncMock()
158 ctx1.__aenter__.return_value = response1
159
160 ctx2 = AsyncMock()
161 ctx2.__aenter__.return_value = response2
162
163 ctx3 = AsyncMock()
164 ctx3.__aenter__.return_value = response3
165
166 provider_mock.mass.http_session.request = MagicMock(side_effect=[ctx1, ctx2, ctx3])
167
168 items: list[dict[str, Any]] = []
169 async for item in api_client.paginate("test/endpoint", limit=2):
170 items.append(item)
171
172 assert len(items) == 4
173 assert items[0]["id"] == 1
174 assert items[3]["id"] == 4
175
176
177async def test_paginate_preserves_params_across_pages(
178 api_client: TidalAPIClient, provider_mock: Mock
179) -> None:
180 """Test that caller-supplied params are sent on every page, not just the first."""
181 response1 = AsyncMock(spec=ClientResponse)
182 response1.status = 200
183 response1.json.return_value = {"items": [{"id": 1}, {"id": 2}]}
184
185 response2 = AsyncMock(spec=ClientResponse)
186 response2.status = 200
187 response2.json.return_value = {"items": []}
188
189 ctx1 = AsyncMock()
190 ctx1.__aenter__.return_value = response1
191 ctx2 = AsyncMock()
192 ctx2.__aenter__.return_value = response2
193 provider_mock.mass.http_session.request = MagicMock(side_effect=[ctx1, ctx2])
194
195 items: list[dict[str, Any]] = []
196 async for item in api_client.paginate("test/endpoint", limit=2, params={"order": "DATE"}):
197 items.append(item)
198
199 assert len(items) == 2
200 for call in provider_mock.mass.http_session.request.call_args_list:
201 assert call[1]["params"]["order"] == "DATE"
202
203
204async def test_delete_success(api_client: TidalAPIClient, provider_mock: Mock) -> None:
205 """Test successful DELETE request."""
206 response = AsyncMock(spec=ClientResponse)
207 response.status = 204
208
209 request_ctx = AsyncMock()
210 request_ctx.__aenter__.return_value = response
211 provider_mock.mass.http_session.request = MagicMock(return_value=request_ctx)
212
213 await api_client.delete("test/endpoint/123")
214
215 # Verify DELETE was called
216 provider_mock.mass.http_session.request.assert_called_once()
217 call_args = provider_mock.mass.http_session.request.call_args
218 assert call_args[0][0] == "DELETE"
219
220
221async def test_delete_with_headers(api_client: TidalAPIClient, provider_mock: Mock) -> None:
222 """Test DELETE request with custom headers."""
223 response = AsyncMock(spec=ClientResponse)
224 response.status = 204
225
226 request_ctx = AsyncMock()
227 request_ctx.__aenter__.return_value = response
228 provider_mock.mass.http_session.request = MagicMock(return_value=request_ctx)
229
230 await api_client.delete("test/endpoint/123", headers={"If-Match": "etag123"})
231
232 # Verify headers were passed
233 call_args = provider_mock.mass.http_session.request.call_args
234 assert "If-Match" in call_args[1]["headers"]
235 assert call_args[1]["headers"]["If-Match"] == "etag123"
236
237
238async def test_put_success(api_client: TidalAPIClient, provider_mock: Mock) -> None:
239 """Test successful PUT request."""
240 response = AsyncMock(spec=ClientResponse)
241 response.status = 200
242 response.json.return_value = {"updated": True}
243
244 request_ctx = AsyncMock()
245 request_ctx.__aenter__.return_value = response
246 provider_mock.mass.http_session.request = MagicMock(return_value=request_ctx)
247
248 result = await api_client.put("test/endpoint", data={"key": "value"})
249 assert result == {"updated": True}
250
251 # Verify PUT was called
252 call_args = provider_mock.mass.http_session.request.call_args
253 assert call_args[0][0] == "PUT"
254
255
256async def test_put_with_form_data(api_client: TidalAPIClient, provider_mock: Mock) -> None:
257 """Test PUT request with form data."""
258 response = AsyncMock(spec=ClientResponse)
259 response.status = 200
260 response.json.return_value = {"success": True}
261
262 request_ctx = AsyncMock()
263 request_ctx.__aenter__.return_value = response
264 provider_mock.mass.http_session.request = MagicMock(return_value=request_ctx)
265
266 result = await api_client.put("test/endpoint", data={"key": "value"}, as_form=True)
267 assert result == {"success": True}
268
269 # Verify form data was used
270 call_args = provider_mock.mass.http_session.request.call_args
271 assert "data" in call_args[1]
272