/
/
/
1"""Test Audible Provider."""
2
3import json
4from pathlib import Path
5from typing import Any
6from unittest.mock import AsyncMock, MagicMock, patch
7
8import audible
9import pytest
10from music_assistant_models.enums import MediaType
11from music_assistant_models.media_items import PodcastEpisode
12
13from music_assistant.providers.audible import Audibleprovider
14from music_assistant.providers.audible.audible_helper import (
15 AudibleHelper,
16 cached_authenticator_from_file,
17 evict_cached_authenticator,
18)
19
20
21@pytest.fixture
22def mass_mock() -> AsyncMock:
23 """Return a mock MusicAssistant instance."""
24 mass = AsyncMock()
25 mass.http_session = AsyncMock()
26 mass.cache.get = AsyncMock(return_value=None)
27 mass.cache.set = AsyncMock()
28 return mass
29
30
31@pytest.fixture
32def audible_client_mock() -> AsyncMock:
33 """Return a mock Audible AsyncClient."""
34 client = AsyncMock()
35 client.post = AsyncMock()
36 client.put = AsyncMock()
37 return client
38
39
40@pytest.fixture
41def helper(mass_mock: AsyncMock, audible_client_mock: AsyncMock) -> AudibleHelper:
42 """Return an AudibleHelper instance."""
43 return AudibleHelper(
44 mass=mass_mock,
45 client=audible_client_mock,
46 provider_domain="audible",
47 provider_instance="audible_test",
48 provider=MagicMock(),
49 )
50
51
52@pytest.fixture
53def provider(mass_mock: AsyncMock) -> Audibleprovider:
54 """Return an Audibleprovider instance."""
55 manifest = MagicMock()
56 manifest.domain = "audible"
57 config = MagicMock()
58
59 def get_value(key: str) -> str | None:
60 if key == "locale":
61 return "us"
62 if key == "auth_file":
63 return "mock_auth_file"
64 return None
65
66 config.get_value.side_effect = get_value
67 config.get_value.return_value = None # Default
68
69 # Patch logger setLevel to avoid ValueError with 'us'
70 with patch("music_assistant.models.provider.logging.Logger.setLevel"):
71 prov = Audibleprovider(mass_mock, manifest, config)
72
73 prov.helper = MagicMock(spec=AudibleHelper)
74 return prov
75
76
77async def test_pagination_get_library(helper: AudibleHelper) -> None:
78 """Test get_library uses pagination correctly."""
79 # To trigger pagination, the first page must have 50 items (page_size)
80 # We generate 50 dummy items for page 1
81 page1_items = [
82 {
83 "asin": f"1_{i}",
84 "title": f"Book 1_{i}",
85 "content_delivery_type": "SinglePartBook",
86 "authors": [],
87 }
88 for i in range(50)
89 ]
90 page2_items = [
91 {
92 "asin": "2_1",
93 "title": "Book 2_1",
94 "content_delivery_type": "SinglePartBook",
95 "authors": [],
96 },
97 ]
98
99 # Mock side_effect for _call_api
100 async def side_effect(_: str, **kwargs: Any) -> dict[str, Any]:
101 if kwargs.get("page") == 1:
102 return {"items": page1_items, "total_results": 51}
103 if kwargs.get("page") == 2:
104 return {"items": page2_items, "total_results": 51}
105 return {"items": [], "total_results": 51}
106
107 with patch.object(helper, "_call_api", side_effect=side_effect) as mock_call:
108 books = []
109 async for book in helper.get_library():
110 books.append(book)
111
112 # 50 from page 1 + 1 from page 2 = 51
113 assert len(books) == 51
114 assert books[0].item_id == "1_0"
115 assert books[50].item_id == "2_1"
116
117 # Verify pagination calls
118 assert mock_call.call_count >= 2
119 calls = mock_call.call_args_list
120 assert calls[0].kwargs["page"] == 1
121 assert calls[1].kwargs["page"] == 2
122
123
124async def test_pagination_browse_helpers(helper: AudibleHelper) -> None:
125 """Test browse helpers (like get_authors) use pagination."""
126 # Mock _call_api to return items across pages
127 # Page 1 must be full (50 items) to trigger next page
128 page1_items = [
129 {
130 "asin": f"1_{i}",
131 "content_delivery_type": "SinglePartBook",
132 "authors": [{"asin": f"A1_{i}", "name": f"Author 1_{i}"}],
133 }
134 for i in range(50)
135 ]
136 page2_items = [
137 {
138 "asin": "2_1",
139 "content_delivery_type": "SinglePartBook",
140 "authors": [{"asin": "A2_1", "name": "Author 2_1"}],
141 },
142 ]
143
144 async def side_effect(_: str, **kwargs: Any) -> dict[str, Any]:
145 if kwargs.get("page") == 1:
146 return {"items": page1_items}
147 if kwargs.get("page") == 2:
148 return {"items": page2_items}
149 return {"items": []}
150
151 with patch.object(helper, "_call_api", side_effect=side_effect):
152 authors = await helper.get_authors()
153
154 # 50 authors from page 1 + 1 from page 2 = 51
155 assert len(authors) == 51
156 assert authors["A1_0"] == "Author 1_0"
157 assert authors["A2_1"] == "Author 2_1"
158
159
160async def test_acr_caching(helper: AudibleHelper, audible_client_mock: AsyncMock) -> None:
161 """Test ACR is cached and used for set_last_position."""
162 asin = "B001"
163
164 # Mock get_stream response
165 audible_client_mock.post.return_value = {
166 "content_license": {
167 "acr": "test_acr_value",
168 "license_response": "http://stream.url",
169 "content_metadata": {"content_reference": {"content_size_in_bytes": 1000}},
170 }
171 }
172
173 # 1. Call get_stream to populate cache
174 await helper.get_stream(asin, MediaType.AUDIOBOOK)
175 assert (asin, MediaType.AUDIOBOOK) in helper._acr_cache
176 assert helper._acr_cache[(asin, MediaType.AUDIOBOOK)] == "test_acr_value"
177
178 # Reset mock to ensure it's not called again if we were to call get_stream
179 # (but we check cache usage in set_last_position)
180 audible_client_mock.post.reset_mock()
181
182 # 2. Call set_last_position -> should use cache and NOT call get_stream
183 # (which calls client.post)
184 # We patch get_stream to verify it's NOT called
185 with patch.object(helper, "get_stream") as mock_get_stream:
186 await helper.set_last_position(asin, 10, MediaType.AUDIOBOOK)
187
188 mock_get_stream.assert_not_called()
189 audible_client_mock.put.assert_called_once()
190 call_args = audible_client_mock.put.call_args[1]
191 assert call_args["body"]["acr"] == "test_acr_value"
192
193
194async def test_set_last_position_without_cache(
195 helper: AudibleHelper, audible_client_mock: AsyncMock
196) -> None:
197 """Test set_last_position fetches ACR if not in cache."""
198 asin = "B002"
199
200 # Mock get_stream internal call
201 with patch.object(helper, "get_stream") as mock_get_stream:
202 mock_get_stream.return_value.data = {"acr": "fetched_acr"}
203
204 await helper.set_last_position(asin, 10, MediaType.AUDIOBOOK)
205
206 mock_get_stream.assert_called_once_with(asin=asin, media_type=MediaType.AUDIOBOOK)
207 audible_client_mock.put.assert_called_once()
208 call_args = audible_client_mock.put.call_args[1]
209 assert call_args["body"]["acr"] == "fetched_acr"
210
211
212async def test_podcast_parent_fallback(helper: AudibleHelper) -> None:
213 """Test podcast episode parsing handles missing parent ASIN."""
214 episode_data = {
215 "asin": "ep1",
216 "title": "Episode 1",
217 "relationships": [], # No parent relationship
218 }
219
220 # Should not raise error, but log warning and use empty/self ASIN for parent
221 episode = helper._parse_podcast_episode(episode_data, None, 0)
222
223 assert isinstance(episode, PodcastEpisode)
224 assert episode.podcast.item_id == ""
225
226
227def _mock_auth(locale: str) -> MagicMock:
228 """Return a mock Authenticator with signing auth and the given locale."""
229 auth = MagicMock()
230 auth.adp_token = "adp_token"
231 auth.device_private_key = "private_key"
232 auth.locale = audible.localization.Locale(locale)
233 return auth
234
235
236def _write_auth_file(path: Path, locale_code: str) -> None:
237 """Write a syntactically valid auth file with dummy tokens and the given locale."""
238 # assembled at runtime so the detect-private-key hook does not flag the dummy value
239 pem = "RSA PRIVATE " + "KEY-----"
240 path.write_text(
241 json.dumps(
242 {
243 "website_cookies": {"session-id": "dummy"},
244 "adp_token": "{enc:x}{key:x}{iv:x}{name:x}{serial:Mg==}",
245 "access_token": "Atna|dummy",
246 "refresh_token": "Atnr|dummy",
247 "device_private_key": f"-----BEGIN {pem}\ndummy\n-----END {pem}\n",
248 "expires": 9999999999.0,
249 "locale_code": locale_code,
250 "device_info": {"device_serial_number": "dummy"},
251 "customer_info": {"name": "dummy"},
252 "with_username": False,
253 }
254 )
255 )
256
257
258async def test_cached_authenticator_corrects_locale_mismatch(tmp_path: Path) -> None:
259 """An auth file holding a stale marketplace is corrected to the configured locale."""
260 path = tmp_path / "auth.json"
261 _write_auth_file(path, "us")
262
263 auth = await cached_authenticator_from_file(str(path), "de")
264
265 assert auth.locale is not None
266 assert auth.locale.country_code == "de"
267 assert auth.locale.domain == "de"
268 assert json.loads(path.read_text())["locale_code"] == "de"
269 evict_cached_authenticator(str(path))
270
271
272async def test_cached_authenticator_keeps_matching_locale(tmp_path: Path) -> None:
273 """An auth file matching the configured locale is left untouched."""
274 path = str(tmp_path / "auth.json")
275 auth = _mock_auth("de")
276 with patch("audible.Authenticator.from_file", return_value=auth):
277 result = await cached_authenticator_from_file(path, "de")
278
279 assert result is auth
280 auth.to_file.assert_not_called()
281 evict_cached_authenticator(path)
282
283
284async def test_cached_authenticator_loads_new_file_after_reauth(tmp_path: Path) -> None:
285 """A new auth file written by a reconfigure is loaded instead of a stale authenticator."""
286 old_path = str(tmp_path / "old.json")
287 new_path = str(tmp_path / "new.json")
288 old_auth = _mock_auth("us")
289 new_auth = _mock_auth("de")
290 with patch("audible.Authenticator.from_file", side_effect=[old_auth, new_auth]):
291 assert await cached_authenticator_from_file(old_path, "us") is old_auth
292 assert await cached_authenticator_from_file(new_path, "de") is new_auth
293
294 evict_cached_authenticator(old_path)
295 evict_cached_authenticator(new_path)
296
297
298async def test_browse_decoding(provider: Audibleprovider) -> None:
299 """Test browse path decoding."""
300 # We need to test the provider's browse method, not the helper's.
301 # We mocked the helper in the provider fixture.
302
303 # Mock helper methods to return empty lists/dicts so we just check calls
304 provider.helper.get_audiobooks_by_author = AsyncMock(return_value=[]) # type: ignore[method-assign]
305 provider.helper.get_audiobooks_by_genre = AsyncMock(return_value=[]) # type: ignore[method-assign]
306
307 # Test Author with special chars
308 await provider.browse("audible://authors/Author%20Name")
309 provider.helper.get_audiobooks_by_author.assert_called_with("Author Name")
310
311 # Test Genre with slash (encoded)
312 await provider.browse("audible://genres/Sci-Fi%2FFantasy")
313 provider.helper.get_audiobooks_by_genre.assert_called_with("Sci-Fi/Fantasy")
314
315
316async def test_get_library_podcasts_includes_legacy_periodicals(helper: AudibleHelper) -> None:
317 """Test podcast sync also picks up series with the legacy Periodical delivery type."""
318 library_items = [
319 {
320 "asin": "P1",
321 "title": "Modern Podcast",
322 "content_delivery_type": "PodcastParent",
323 },
324 {
325 "asin": "P2",
326 "title": "Audible Original Show",
327 "content_delivery_type": "Periodical",
328 },
329 {
330 "asin": "B1",
331 "title": "Some Book",
332 "content_delivery_type": "SinglePartBook",
333 },
334 ]
335
336 async def side_effect(_: str, **kwargs: Any) -> dict[str, Any]:
337 if kwargs.get("page") == 1:
338 return {"items": library_items}
339 return {"items": []}
340
341 with patch.object(helper, "_call_api", side_effect=side_effect):
342 podcasts = [podcast async for podcast in helper.get_library_podcasts()]
343
344 assert [podcast.item_id for podcast in podcasts] == ["P1", "P2"]
345
346
347async def test_podcast_episodes_ranked_by_publication_datetime(helper: AudibleHelper) -> None:
348 """Episodes are positioned by publication time, not the order the API returns them."""
349 # a serialised show: listed oldest-first, with two episodes sharing a release_date
350 episodes = [
351 {
352 "asin": "trailer",
353 "title": "Trailer",
354 "relationships": [],
355 "release_date": "2022-08-17",
356 "publication_datetime": "2022-08-17T04:27:10Z",
357 },
358 {
359 "asin": "ep2",
360 "title": "Ep2",
361 "relationships": [],
362 "release_date": "2022-08-21",
363 "publication_datetime": "2022-08-21T15:02:44Z",
364 },
365 {
366 "asin": "ep3",
367 "title": "Ep3",
368 "relationships": [],
369 "release_date": "2022-08-21",
370 "publication_datetime": "2022-08-21T15:04:34Z",
371 },
372 ]
373
374 async def side_effect(_: str, **kwargs: Any) -> dict[str, Any]:
375 return {"items": episodes} if kwargs.get("page") == 1 else {"items": []}
376
377 helper._call_api = AsyncMock(side_effect=side_effect) # type: ignore[method-assign]
378 helper.get_podcast = AsyncMock(return_value=None) # type: ignore[method-assign]
379
380 parsed = [ep async for ep in helper.get_podcast_episodes("parent")]
381
382 assert {ep.item_id: ep.position for ep in parsed} == {"trailer": 1, "ep2": 2, "ep3": 3}
383
384
385async def test_podcast_episodes_reverse_a_newest_first_listing(helper: AudibleHelper) -> None:
386 """A show listed newest-first gets its newest episode the highest position."""
387 episodes = [
388 {
389 "asin": "new",
390 "title": "Newest",
391 "relationships": [],
392 "publication_datetime": "2026-08-24T04:01:00Z",
393 },
394 {
395 "asin": "mid",
396 "title": "Middle",
397 "relationships": [],
398 "publication_datetime": "2026-08-10T04:01:00Z",
399 },
400 {
401 "asin": "old",
402 "title": "Oldest",
403 "relationships": [],
404 "publication_datetime": "2026-07-27T04:01:00Z",
405 },
406 ]
407
408 async def side_effect(_: str, **kwargs: Any) -> dict[str, Any]:
409 return {"items": episodes} if kwargs.get("page") == 1 else {"items": []}
410
411 helper._call_api = AsyncMock(side_effect=side_effect) # type: ignore[method-assign]
412 helper.get_podcast = AsyncMock(return_value=None) # type: ignore[method-assign]
413
414 parsed = [ep async for ep in helper.get_podcast_episodes("parent")]
415
416 assert {ep.item_id: ep.position for ep in parsed} == {"new": 3, "mid": 2, "old": 1}
417
418
419async def test_legacy_show_episodes_use_the_listing_order(helper: AudibleHelper) -> None:
420 """A legacy series is released in one go, so its listing order decides the position."""
421 # every episode shares one publication timestamp, so the dates rank nothing
422 episodes = [
423 {
424 "asin": "ep5",
425 "title": "Ep 5",
426 "relationships": [],
427 "content_type": "Show",
428 "publication_datetime": "2021-02-09T00:00:00Z",
429 },
430 {
431 "asin": "ep4",
432 "title": "Ep 4",
433 "relationships": [],
434 "content_type": "Show",
435 "publication_datetime": "2021-02-09T00:00:00Z",
436 },
437 {
438 "asin": "ep3",
439 "title": "Ep 3",
440 "relationships": [],
441 "content_type": "Show",
442 "publication_datetime": "2021-02-09T00:00:00Z",
443 },
444 ]
445
446 async def side_effect(_: str, **kwargs: Any) -> dict[str, Any]:
447 return {"items": episodes} if kwargs.get("page") == 1 else {"items": []}
448
449 helper._call_api = AsyncMock(side_effect=side_effect) # type: ignore[method-assign]
450 helper.get_podcast = AsyncMock(return_value=None) # type: ignore[method-assign]
451
452 parsed = [ep async for ep in helper.get_podcast_episodes("parent")]
453
454 assert {ep.item_id: ep.position for ep in parsed} == {"ep5": 3, "ep4": 2, "ep3": 1}
455