/
/
/
1"""Test nicovideo's two-method recommendations contract."""
2
3from __future__ import annotations
4
5import asyncio
6from typing import Any
7from unittest.mock import AsyncMock, Mock, call
8
9import pytest
10from music_assistant_models.media_items import ProviderMapping, Track
11from music_assistant_models.unique_list import UniqueList
12
13from music_assistant.providers.nicovideo import SUPPORTED_FEATURES
14from music_assistant.providers.nicovideo.provider import NicovideoMusicProvider
15from tests.common import use_real_create_task
16
17EXPECTED_ROW_IDS = [
18 "nicovideo_recommendations",
19 "nicovideo_history",
20 "nicovideo_following_activities",
21 "nicovideo_like_history",
22]
23
24
25@pytest.fixture
26def provider() -> NicovideoMusicProvider:
27 """Create a real NicovideoMusicProvider with mocked dependencies."""
28 mass = Mock()
29 manifest = Mock()
30 manifest.domain = "nicovideo"
31 config = Mock()
32 config.instance_id = "nicovideo--test123"
33 config.name = "Nicovideo Test"
34 config.enabled = True
35 config.get_value.side_effect = lambda key, default=None: {
36 "log_level": "GLOBAL",
37 }.get(key, default)
38 return NicovideoMusicProvider(mass, manifest, config, SUPPORTED_FEATURES)
39
40
41def _track(item_id: str) -> Track:
42 """Create a minimal Track for stubbed user service responses."""
43 return Track(
44 item_id=item_id,
45 provider="nicovideo--test123",
46 name=f"Track {item_id}",
47 provider_mappings={
48 ProviderMapping(
49 item_id=item_id,
50 provider_domain="nicovideo",
51 provider_instance="nicovideo--test123",
52 )
53 },
54 )
55
56
57def _stub_user_service(provider: NicovideoMusicProvider) -> dict[str, AsyncMock]:
58 """Replace the user service fetch methods with AsyncMocks returning canned tracks."""
59 user = provider.service_manager.user
60 mocks = {
61 "get_recommendations": AsyncMock(return_value=[_track("rec1")]),
62 "get_user_history": AsyncMock(return_value=[_track("hist1")]),
63 "get_following_activities": AsyncMock(return_value=[_track("follow1")]),
64 "get_like_history": AsyncMock(return_value=[_track("like1")]),
65 }
66 user.get_recommendations = mocks["get_recommendations"] # type: ignore[method-assign]
67 user.get_user_history = mocks["get_user_history"] # type: ignore[method-assign]
68 user.get_following_activities = mocks["get_following_activities"] # type: ignore[method-assign]
69 user.get_like_history = mocks["get_like_history"] # type: ignore[method-assign]
70 return mocks
71
72
73def _install_cache_mocks(provider: NicovideoMusicProvider) -> None:
74 """Make the @use_cache decorator treat every call as a cache miss."""
75 provider.mass.cache.get_with_freshness = AsyncMock( # type: ignore[method-assign]
76 return_value=(None, False, False)
77 )
78 provider.mass.cache.set = AsyncMock() # type: ignore[method-assign]
79 use_real_create_task(provider.mass)
80
81
82def _install_dict_backed_cache(provider: NicovideoMusicProvider) -> list[asyncio.Future[Any]]:
83 """
84 Install a dict-backed fake cache that serves what set() stored.
85
86 Mirrors production: entries are stored serialized (to_dict) and, when the caller
87 passes base_class, reconstructed via base_class.from_dict on a hit. Returns the
88 list of background store tasks so tests can await them between calls.
89 """
90 store: dict[str, Any] = {}
91 tasks: list[asyncio.Future[Any]] = []
92
93 async def _cache_get(key: str, **kwargs: Any) -> tuple[Any, bool, bool]:
94 if key not in store:
95 return None, False, False
96 data = store[key]
97 base_class = kwargs.get("base_class")
98 if base_class is not None and data is not None:
99 return base_class.from_dict(data), True, True
100 return data, True, True
101
102 async def _cache_set(key: str, data: Any, **_kwargs: Any) -> None:
103 store[key] = data.to_dict() if data is not None else None
104
105 def _create_task(target: Any, *_args: Any, **_kwargs: Any) -> asyncio.Future[Any]:
106 task: asyncio.Future[Any] = asyncio.ensure_future(target)
107 tasks.append(task)
108 return task
109
110 provider.mass.cache.get_with_freshness = AsyncMock( # type: ignore[method-assign]
111 side_effect=_cache_get
112 )
113 provider.mass.cache.set = AsyncMock(side_effect=_cache_set) # type: ignore[method-assign]
114 provider.mass.create_task = Mock(side_effect=_create_task) # type: ignore[method-assign]
115 return tasks
116
117
118@pytest.mark.asyncio
119async def test_get_recommendations_returns_static_rows_without_backend_calls(
120 provider: NicovideoMusicProvider,
121) -> None:
122 """get_recommendations() returns the four static rows, without items or backend calls."""
123 mocks = _stub_user_service(provider)
124
125 result = await provider.get_recommendations()
126
127 assert [folder.item_id for folder in result] == EXPECTED_ROW_IDS
128 for folder in result:
129 assert folder.provider == provider.instance_id
130 assert not folder.items
131 for mock in mocks.values():
132 mock.assert_not_awaited()
133
134
135@pytest.mark.asyncio
136@pytest.mark.parametrize(
137 ("item_id", "fetch_name", "expected_call", "expected_track_id"),
138 [
139 (
140 "nicovideo_recommendations",
141 "get_recommendations",
142 call("video_recommendation_recommend", limit=25),
143 "rec1",
144 ),
145 ("nicovideo_history", "get_user_history", call(limit=50), "hist1"),
146 (
147 "nicovideo_following_activities",
148 "get_following_activities",
149 call(limit=30),
150 "follow1",
151 ),
152 ("nicovideo_like_history", "get_like_history", call(limit=50), "like1"),
153 ],
154)
155async def test_get_recommendation_items_fetches_only_that_row(
156 provider: NicovideoMusicProvider,
157 item_id: str,
158 fetch_name: str,
159 expected_call: object,
160 expected_track_id: str,
161) -> None:
162 """get_recommendation_items(row) issues only that row's fetch and returns its items."""
163 _install_cache_mocks(provider)
164 mocks = _stub_user_service(provider)
165
166 result = await provider.get_recommendation_items(item_id)
167
168 mocks[fetch_name].assert_awaited_once()
169 assert mocks[fetch_name].await_args == expected_call
170 for name, mock in mocks.items():
171 if name != fetch_name:
172 mock.assert_not_awaited()
173 assert [item.item_id for item in result] == [expected_track_id]
174
175
176@pytest.mark.asyncio
177async def test_get_recommendation_items_unknown_id_returns_empty(
178 provider: NicovideoMusicProvider,
179) -> None:
180 """get_recommendation_items() with an unknown item_id returns empty without any fetch."""
181 _install_cache_mocks(provider)
182 mocks = _stub_user_service(provider)
183
184 result = await provider.get_recommendation_items("nicovideo_does_not_exist")
185
186 assert result == UniqueList()
187 for mock in mocks.values():
188 mock.assert_not_awaited()
189
190
191@pytest.mark.asyncio
192async def test_get_recommendation_items_warm_cache_hit_serves_stored_items(
193 provider: NicovideoMusicProvider,
194) -> None:
195 """A second call for the same row is served from the cache, without a new backend fetch."""
196 tasks = _install_dict_backed_cache(provider)
197 mocks = _stub_user_service(provider)
198
199 first = await provider.get_recommendation_items("nicovideo_history")
200 # let the background cache-store task complete before the warm call
201 await asyncio.gather(*tasks)
202 second = await provider.get_recommendation_items("nicovideo_history")
203
204 mocks["get_user_history"].assert_awaited_once()
205 assert [item.item_id for item in first] == ["hist1"]
206 assert [item.item_id for item in second] == ["hist1"]
207