music-assistant-server

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