/
/
/
1"""Test Audible Provider."""
2
3from typing import Any
4from unittest.mock import AsyncMock, MagicMock, patch
5
6import pytest
7from music_assistant_models.enums import MediaType
8from music_assistant_models.media_items import PodcastEpisode
9
10from music_assistant.providers.audible import Audibleprovider
11from music_assistant.providers.audible.audible_helper import AudibleHelper
12
13
14@pytest.fixture
15def mass_mock() -> AsyncMock:
16 """Return a mock MusicAssistant instance."""
17 mass = AsyncMock()
18 mass.http_session = AsyncMock()
19 mass.cache.get = AsyncMock(return_value=None)
20 mass.cache.set = AsyncMock()
21 return mass
22
23
24@pytest.fixture
25def audible_client_mock() -> AsyncMock:
26 """Return a mock Audible AsyncClient."""
27 client = AsyncMock()
28 client.post = AsyncMock()
29 client.put = AsyncMock()
30 return client
31
32
33@pytest.fixture
34def helper(mass_mock: AsyncMock, audible_client_mock: AsyncMock) -> AudibleHelper:
35 """Return an AudibleHelper instance."""
36 return AudibleHelper(
37 mass=mass_mock,
38 client=audible_client_mock,
39 provider_domain="audible",
40 provider_instance="audible_test",
41 )
42
43
44@pytest.fixture
45def provider(mass_mock: AsyncMock) -> Audibleprovider:
46 """Return an Audibleprovider instance."""
47 manifest = MagicMock()
48 manifest.domain = "audible"
49 config = MagicMock()
50
51 def get_value(key: str) -> str | None:
52 if key == "locale":
53 return "us"
54 if key == "auth_file":
55 return "mock_auth_file"
56 return None
57
58 config.get_value.side_effect = get_value
59 config.get_value.return_value = None # Default
60
61 # Patch logger setLevel to avoid ValueError with 'us'
62 with patch("music_assistant.models.provider.logging.Logger.setLevel"):
63 prov = Audibleprovider(mass_mock, manifest, config)
64
65 prov.helper = MagicMock(spec=AudibleHelper)
66 return prov
67
68
69async def test_pagination_get_library(helper: AudibleHelper) -> None:
70 """Test get_library uses pagination correctly."""
71 # To trigger pagination, the first page must have 50 items (page_size)
72 # We generate 50 dummy items for page 1
73 page1_items = [
74 {
75 "asin": f"1_{i}",
76 "title": f"Book 1_{i}",
77 "content_delivery_type": "SinglePartBook",
78 "authors": [],
79 }
80 for i in range(50)
81 ]
82 page2_items = [
83 {
84 "asin": "2_1",
85 "title": "Book 2_1",
86 "content_delivery_type": "SinglePartBook",
87 "authors": [],
88 },
89 ]
90
91 # Mock side_effect for _call_api
92 async def side_effect(_: str, **kwargs: Any) -> dict[str, Any]:
93 if kwargs.get("page") == 1:
94 return {"items": page1_items, "total_results": 51}
95 if kwargs.get("page") == 2:
96 return {"items": page2_items, "total_results": 51}
97 return {"items": [], "total_results": 51}
98
99 with patch.object(helper, "_call_api", side_effect=side_effect) as mock_call:
100 books = []
101 async for book in helper.get_library():
102 books.append(book)
103
104 # 50 from page 1 + 1 from page 2 = 51
105 assert len(books) == 51
106 assert books[0].item_id == "1_0"
107 assert books[50].item_id == "2_1"
108
109 # Verify pagination calls
110 assert mock_call.call_count >= 2
111 calls = mock_call.call_args_list
112 assert calls[0].kwargs["page"] == 1
113 assert calls[1].kwargs["page"] == 2
114
115
116async def test_pagination_browse_helpers(helper: AudibleHelper) -> None:
117 """Test browse helpers (like get_authors) use pagination."""
118 # Mock _call_api to return items across pages
119 # Page 1 must be full (50 items) to trigger next page
120 page1_items = [
121 {
122 "asin": f"1_{i}",
123 "content_delivery_type": "SinglePartBook",
124 "authors": [{"asin": f"A1_{i}", "name": f"Author 1_{i}"}],
125 }
126 for i in range(50)
127 ]
128 page2_items = [
129 {
130 "asin": "2_1",
131 "content_delivery_type": "SinglePartBook",
132 "authors": [{"asin": "A2_1", "name": "Author 2_1"}],
133 },
134 ]
135
136 async def side_effect(_: str, **kwargs: Any) -> dict[str, Any]:
137 if kwargs.get("page") == 1:
138 return {"items": page1_items}
139 if kwargs.get("page") == 2:
140 return {"items": page2_items}
141 return {"items": []}
142
143 with patch.object(helper, "_call_api", side_effect=side_effect):
144 authors = await helper.get_authors()
145
146 # 50 authors from page 1 + 1 from page 2 = 51
147 assert len(authors) == 51
148 assert authors["A1_0"] == "Author 1_0"
149 assert authors["A2_1"] == "Author 2_1"
150
151
152async def test_acr_caching(helper: AudibleHelper, audible_client_mock: AsyncMock) -> None:
153 """Test ACR is cached and used for set_last_position."""
154 asin = "B001"
155
156 # Mock get_stream response
157 audible_client_mock.post.return_value = {
158 "content_license": {
159 "acr": "test_acr_value",
160 "license_response": "http://stream.url",
161 "content_metadata": {"content_reference": {"content_size_in_bytes": 1000}},
162 }
163 }
164
165 # 1. Call get_stream to populate cache
166 await helper.get_stream(asin, MediaType.AUDIOBOOK)
167 assert (asin, MediaType.AUDIOBOOK) in helper._acr_cache
168 assert helper._acr_cache[(asin, MediaType.AUDIOBOOK)] == "test_acr_value"
169
170 # Reset mock to ensure it's not called again if we were to call get_stream
171 # (but we check cache usage in set_last_position)
172 audible_client_mock.post.reset_mock()
173
174 # 2. Call set_last_position -> should use cache and NOT call get_stream
175 # (which calls client.post)
176 # We patch get_stream to verify it's NOT called
177 with patch.object(helper, "get_stream") as mock_get_stream:
178 await helper.set_last_position(asin, 10, MediaType.AUDIOBOOK)
179
180 mock_get_stream.assert_not_called()
181 audible_client_mock.put.assert_called_once()
182 call_args = audible_client_mock.put.call_args[1]
183 assert call_args["body"]["acr"] == "test_acr_value"
184
185
186async def test_set_last_position_without_cache(
187 helper: AudibleHelper, audible_client_mock: AsyncMock
188) -> None:
189 """Test set_last_position fetches ACR if not in cache."""
190 asin = "B002"
191
192 # Mock get_stream internal call
193 with patch.object(helper, "get_stream") as mock_get_stream:
194 mock_get_stream.return_value.data = {"acr": "fetched_acr"}
195
196 await helper.set_last_position(asin, 10, MediaType.AUDIOBOOK)
197
198 mock_get_stream.assert_called_once_with(asin=asin, media_type=MediaType.AUDIOBOOK)
199 audible_client_mock.put.assert_called_once()
200 call_args = audible_client_mock.put.call_args[1]
201 assert call_args["body"]["acr"] == "fetched_acr"
202
203
204async def test_podcast_parent_fallback(helper: AudibleHelper) -> None:
205 """Test podcast episode parsing handles missing parent ASIN."""
206 episode_data = {
207 "asin": "ep1",
208 "title": "Episode 1",
209 "relationships": [], # No parent relationship
210 }
211
212 # Should not raise error, but log warning and use empty/self ASIN for parent
213 episode = helper._parse_podcast_episode(episode_data, None, 0)
214
215 assert isinstance(episode, PodcastEpisode)
216 assert episode.podcast.item_id == ""
217
218
219async def test_browse_decoding(provider: Audibleprovider) -> None:
220 """Test browse path decoding."""
221 # We need to test the provider's browse method, not the helper's.
222 # We mocked the helper in the provider fixture.
223
224 # Mock helper methods to return empty lists/dicts so we just check calls
225 provider.helper.get_audiobooks_by_author = AsyncMock(return_value=[]) # type: ignore[method-assign]
226 provider.helper.get_audiobooks_by_genre = AsyncMock(return_value=[]) # type: ignore[method-assign]
227
228 # Test Author with special chars
229 await provider.browse("audible://authors/Author%20Name")
230 provider.helper.get_audiobooks_by_author.assert_called_with("Author Name")
231
232 # Test Genre with slash (encoded)
233 await provider.browse("audible://genres/Sci-Fi%2FFantasy")
234 provider.helper.get_audiobooks_by_genre.assert_called_with("Sci-Fi/Fantasy")
235