/
/
/
1"""Test Audiobookshelf's two-method recommendations contract."""
2
3from __future__ import annotations
4
5import asyncio
6from typing import Any
7from unittest.mock import AsyncMock, Mock
8
9import pytest
10from aioaudiobookshelf.schema.shelf import ShelfBook, ShelfLibraryItemMinified
11from aioaudiobookshelf.schema.shelf import ShelfId as AbsShelfId
12from aioaudiobookshelf.schema.shelf import ShelfType as AbsShelfType
13from music_assistant_models.media_items import BrowseFolder, UniqueList
14
15from music_assistant.providers.audiobookshelf import Audiobookshelf
16
17
18def _make_shelf() -> Mock:
19 """Create a minimal recently-added book shelf as returned by the personalized view."""
20 entity = Mock(spec=ShelfLibraryItemMinified)
21 entity.id_ = "book1"
22 shelf = Mock(spec=ShelfBook)
23 shelf.id_ = AbsShelfId.RECENTLY_ADDED
24 shelf.type_ = AbsShelfType.BOOK
25 shelf.entities = [entity]
26 return shelf
27
28
29def _stub_backend(provider: Audiobookshelf) -> AsyncMock:
30 """Stub the personalized view call and the library item lookup."""
31 view_mock = AsyncMock(return_value=[_make_shelf()])
32 provider._client.get_library_personalized_view = view_mock # type: ignore[method-assign]
33 provider.mass.music.get_library_item_by_prov_id = AsyncMock( # type: ignore[method-assign]
34 return_value=Mock()
35 )
36 return view_mock
37
38
39def _install_cache_mocks(provider: Audiobookshelf) -> list[asyncio.Future[Any]]:
40 """Back the @use_cache decorator with a dict store; return the background store tasks."""
41 store: dict[str, Any] = {}
42 tasks: list[asyncio.Future[Any]] = []
43
44 async def _cache_get(key: str, **_kwargs: Any) -> tuple[Any, bool, bool]:
45 if key in store:
46 return store[key], True, True
47 return None, False, False
48
49 async def _cache_set(key: str, data: Any, **_kwargs: Any) -> None:
50 store[key] = data
51
52 def _create_task(target: Any, *_args: Any, **_kwargs: Any) -> asyncio.Future[Any]:
53 task: asyncio.Future[Any] = asyncio.ensure_future(target)
54 tasks.append(task)
55 return task
56
57 provider.mass.cache.get_with_freshness = AsyncMock( # type: ignore[method-assign]
58 side_effect=_cache_get
59 )
60 provider.mass.cache.set = AsyncMock(side_effect=_cache_set) # type: ignore[method-assign]
61 provider.mass.create_task = Mock(side_effect=_create_task) # type: ignore[method-assign]
62 return tasks
63
64
65@pytest.mark.asyncio
66async def test_get_recommendations_returns_shelf_rows_plus_browse(
67 provider: Audiobookshelf,
68) -> None:
69 """The rows are the payload's shelf rows plus the static browse row, all without items."""
70 _install_cache_mocks(provider)
71 view_mock = _stub_backend(provider)
72
73 rows = await provider.get_recommendations()
74
75 view_mock.assert_awaited_once_with(library_id="lib1", limit=20)
76 assert [f.item_id for f in rows] == ["recently-added", "browse"]
77 assert all(len(f.items) == 0 for f in rows)
78
79
80@pytest.mark.asyncio
81async def test_get_recommendations_row_identity(provider: Audiobookshelf) -> None:
82 """Row identity fields match the previous bulk implementation exactly."""
83 _install_cache_mocks(provider)
84 _stub_backend(provider)
85
86 shelf_row, browse_row = await provider.get_recommendations()
87
88 assert shelf_row.name == "Recently added"
89 assert shelf_row.icon == "mdi-plus-box-multiple-outline"
90 assert shelf_row.translation_key == "recently_added"
91 assert shelf_row.provider == provider.instance_id
92 assert browse_row.name == "Libraries"
93 assert browse_row.icon == "mdi-bookshelf"
94 assert browse_row.translation_key == "library"
95 assert browse_row.provider == provider.instance_id
96
97
98@pytest.mark.asyncio
99async def test_get_recommendations_no_libraries_returns_no_rows(
100 provider: Audiobookshelf,
101) -> None:
102 """Without any libraries there are no rows at all, browse included."""
103 _install_cache_mocks(provider)
104 view_mock = _stub_backend(provider)
105 provider.libraries.audiobooks.clear()
106
107 assert await provider.get_recommendations() == []
108 view_mock.assert_not_awaited()
109
110
111@pytest.mark.asyncio
112async def test_get_recommendation_items_served_from_cached_payload(
113 provider: Audiobookshelf,
114) -> None:
115 """A shelf row's items call reuses the payload the rows call fetched."""
116 tasks = _install_cache_mocks(provider)
117 view_mock = _stub_backend(provider)
118
119 await provider.get_recommendations()
120 # let the background cache-store task complete before the items call
121 await asyncio.gather(*tasks)
122 items = await provider.get_recommendation_items("recently-added")
123
124 view_mock.assert_awaited_once_with(library_id="lib1", limit=20)
125 assert len(items) == 1
126
127
128@pytest.mark.asyncio
129async def test_get_recommendation_items_cold_triggers_payload_fetch(
130 provider: Audiobookshelf,
131) -> None:
132 """An items call without a warm cache performs the payload fetch itself."""
133 _install_cache_mocks(provider)
134 view_mock = _stub_backend(provider)
135
136 items = await provider.get_recommendation_items("recently-added")
137
138 view_mock.assert_awaited_once_with(library_id="lib1", limit=20)
139 assert len(items) == 1
140
141
142@pytest.mark.asyncio
143async def test_get_recommendation_items_browse_is_local(provider: Audiobookshelf) -> None:
144 """The browse row's items are built from local library state, without any backend fetch."""
145 _install_cache_mocks(provider)
146 view_mock = _stub_backend(provider)
147
148 items = await provider.get_recommendation_items("browse")
149
150 view_mock.assert_not_awaited()
151 assert len(items) > 0
152 assert all(isinstance(item, BrowseFolder) for item in items)
153
154
155@pytest.mark.asyncio
156async def test_get_recommendation_items_unknown_id_returns_empty(
157 provider: Audiobookshelf,
158) -> None:
159 """An unknown row id yields an empty result."""
160 _install_cache_mocks(provider)
161 _stub_backend(provider)
162
163 items = await provider.get_recommendation_items("bogus")
164
165 assert isinstance(items, UniqueList)
166 assert len(items) == 0
167