/
/
/
1"""Test Nugs.net two-method recommendations: static rows + per-row item fetching."""
2
3from __future__ import annotations
4
5import asyncio
6from typing import Any
7from unittest.mock import AsyncMock, Mock
8
9import pytest
10
11from music_assistant.providers.nugs import NugsProvider
12from tests.common import use_real_create_task
13
14POPULAR_DATA = {"items": [{"id": "100"}]}
15SHOW_DATA = {
16 "Response": {"id": "100", "title": "Popular Show", "artist": {"id": "9", "name": "Artist"}}
17}
18RECOMMENDED_DATA = {
19 "items": [{"id": "200", "title": "Recommended Show", "artist": {"id": "9", "name": "Artist"}}]
20}
21RECENT_DATA = {
22 "items": [{"id": "300", "title": "Recent Show", "artist": {"id": "9", "name": "Artist"}}]
23}
24
25
26def _stub_get_data(provider: NugsProvider) -> AsyncMock:
27 """Attach a _get_data stub that returns canned data keyed by endpoint."""
28
29 async def _fake(_nugs_api: str, endpoint: str, **_kwargs: Any) -> Any:
30 if endpoint == "releases/popular":
31 return POPULAR_DATA
32 if endpoint.startswith("shows/"):
33 return SHOW_DATA
34 if endpoint == "me/releases/recommendations":
35 return RECOMMENDED_DATA
36 if endpoint == "releases/recent":
37 return RECENT_DATA
38 return {"items": []}
39
40 mock = AsyncMock(side_effect=_fake)
41 provider._get_data = mock # type: ignore[method-assign]
42 return mock
43
44
45def _install_cache_mocks(provider: NugsProvider) -> None:
46 """Make the @use_cache decorator treat every call as a cache miss."""
47 provider.mass.cache.get_with_freshness = AsyncMock( # type: ignore[method-assign]
48 return_value=(None, False, False)
49 )
50 provider.mass.cache.set = AsyncMock() # type: ignore[method-assign]
51 use_real_create_task(provider.mass)
52
53
54@pytest.mark.asyncio
55async def test_get_recommendations_static_rows_no_backend_calls(provider: NugsProvider) -> None:
56 """get_recommendations returns the three static rows without any backend call."""
57 api_mock = _stub_get_data(provider)
58
59 result = await provider.get_recommendations()
60
61 api_mock.assert_not_awaited()
62 assert [folder.item_id for folder in result] == [
63 "nugs_popular_shows",
64 "nugs_recommended_shows",
65 "nugs_recent_shows",
66 ]
67 assert [folder.name for folder in result] == [
68 "Most Popular",
69 "Recommended Shows",
70 "Recent Shows",
71 ]
72 assert [folder.translation_key for folder in result] == [
73 "nugs_popular_shows",
74 "nugs_recommended_shows",
75 "nugs_recent_shows",
76 ]
77 assert all(folder.provider == provider.instance_id for folder in result)
78 assert all(len(folder.items) == 0 for folder in result)
79
80
81@pytest.mark.asyncio
82async def test_get_recommendation_items_popular(provider: NugsProvider) -> None:
83 """The popular row fetches the popular listing plus per-show details, nothing else."""
84 _install_cache_mocks(provider)
85 api_mock = _stub_get_data(provider)
86
87 result = await provider.get_recommendation_items("nugs_popular_shows")
88
89 called_endpoints = {call.args[1] for call in api_mock.call_args_list}
90 assert called_endpoints == {"releases/popular", "shows/100"}
91 assert [item.item_id for item in result] == ["100"]
92
93
94@pytest.mark.asyncio
95async def test_get_recommendation_items_recommended(provider: NugsProvider) -> None:
96 """The recommended row fetches only its own endpoint."""
97 _install_cache_mocks(provider)
98 api_mock = _stub_get_data(provider)
99
100 result = await provider.get_recommendation_items("nugs_recommended_shows")
101
102 api_mock.assert_awaited_once()
103 assert api_mock.call_args.args == ("catalog", "me/releases/recommendations")
104 assert [item.item_id for item in result] == ["200"]
105
106
107@pytest.mark.asyncio
108async def test_get_recommendation_items_recent(provider: NugsProvider) -> None:
109 """The recent row fetches only releases/recent, skipping the popular N+1 loop."""
110 _install_cache_mocks(provider)
111 api_mock = _stub_get_data(provider)
112
113 result = await provider.get_recommendation_items("nugs_recent_shows")
114
115 api_mock.assert_awaited_once()
116 assert api_mock.call_args.args == ("catalog", "releases/recent")
117 assert [item.item_id for item in result] == ["300"]
118
119
120@pytest.mark.asyncio
121async def test_get_recommendation_items_warm_cache_hit(provider: NugsProvider) -> None:
122 """A second items call is served from the cache: backend not fetched, items intact."""
123 store: dict[str, Any] = {}
124 tasks: list[asyncio.Future[Any]] = []
125
126 async def _cache_get(key: str, **kwargs: Any) -> tuple[Any, bool, bool]:
127 if key not in store:
128 return None, False, False
129 data = store[key]
130 base_class = kwargs.get("base_class")
131 if base_class is not None and isinstance(data, dict):
132 # mimic production get_with_freshness reconstructing via base_class
133 data = base_class.from_dict(data)
134 return data, True, True
135
136 async def _cache_set(key: str, data: Any, **_kwargs: Any) -> None:
137 # mimic production cache storing serialized data
138 store[key] = data.to_dict() if hasattr(data, "to_dict") else data
139
140 def _create_task(coro: Any, **_kwargs: Any) -> Any:
141 task = asyncio.ensure_future(coro)
142 tasks.append(task)
143 return task
144
145 provider.mass.cache.get_with_freshness = AsyncMock( # type: ignore[method-assign]
146 side_effect=_cache_get
147 )
148 provider.mass.cache.set = AsyncMock(side_effect=_cache_set) # type: ignore[method-assign]
149 provider.mass.create_task = Mock(side_effect=_create_task) # type: ignore[method-assign]
150 api_mock = _stub_get_data(provider)
151
152 first = await provider.get_recommendation_items("nugs_recommended_shows")
153 # let the background cache-store task complete before the second call
154 await asyncio.gather(*tasks)
155 api_mock.assert_awaited_once()
156 api_mock.reset_mock()
157
158 second = await provider.get_recommendation_items("nugs_recommended_shows")
159
160 api_mock.assert_not_awaited()
161 assert [item.item_id for item in first] == ["200"]
162 assert [item.item_id for item in second] == ["200"]
163
164
165@pytest.mark.asyncio
166async def test_get_recommendation_items_unknown_id_returns_empty(provider: NugsProvider) -> None:
167 """An unknown row item_id yields an empty result without any backend call."""
168 _install_cache_mocks(provider)
169 api_mock = _stub_get_data(provider)
170
171 result = await provider.get_recommendation_items("bogus_row")
172
173 api_mock.assert_not_awaited()
174 assert len(result) == 0
175