/
/
/
1"""Test QQ Music two-method recommendations: static rows + per-row item fetches."""
2
3# mypy: ignore-errors
4
5from __future__ import annotations
6
7from types import SimpleNamespace
8from unittest.mock import AsyncMock, Mock
9
10import pytest
11from music_assistant_models.unique_list import UniqueList
12
13from music_assistant.providers.qqmusic import QQMusicProvider
14
15GUESS_PAYLOAD = {
16 "songs": [{"mid": "t1", "title": "Guess Song", "singer": [{"mid": "a1", "name": "Artist One"}]}]
17}
18RADAR_PAYLOAD = {
19 "songs": [{"mid": "t2", "title": "Radar Song", "singer": [{"mid": "a2", "name": "Artist Two"}]}]
20}
21NEWSONG_PAYLOAD = {
22 "songs": [{"mid": "t3", "title": "New Song", "singer": [{"mid": "a3", "name": "Artist Three"}]}]
23}
24SONGLIST_PAYLOAD = {"songlists": [{"id": 123, "title": "Recommended List"}]}
25
26
27def _make_provider() -> QQMusicProvider:
28 """Create a QQMusicProvider with stubbed recommendation backend calls."""
29 provider = QQMusicProvider.__new__(QQMusicProvider)
30 provider.manifest = SimpleNamespace(domain="qqmusic")
31 provider.config = SimpleNamespace(instance_id="qqmusic_instance")
32 provider.logger = Mock()
33 provider._credential = Mock()
34 provider._recommend_payload_cache = {}
35 provider._qq_recommend = SimpleNamespace(
36 get_guess_recommend=AsyncMock(return_value=GUESS_PAYLOAD),
37 get_radar_recommend=AsyncMock(return_value=RADAR_PAYLOAD),
38 get_recommend_newsong=AsyncMock(return_value=NEWSONG_PAYLOAD),
39 get_recommend_songlist=AsyncMock(return_value=SONGLIST_PAYLOAD),
40 )
41
42 async def _run_with_session(coro):
43 return await coro
44
45 provider._run_with_session = _run_with_session
46 return provider
47
48
49def _assert_no_backend_calls(provider: QQMusicProvider) -> None:
50 """Assert none of the stubbed recommendation endpoints were awaited."""
51 provider._qq_recommend.get_guess_recommend.assert_not_awaited()
52 provider._qq_recommend.get_radar_recommend.assert_not_awaited()
53 provider._qq_recommend.get_recommend_newsong.assert_not_awaited()
54 provider._qq_recommend.get_recommend_songlist.assert_not_awaited()
55
56
57@pytest.mark.asyncio
58async def test_get_recommendations_static_rows_zero_backend_calls() -> None:
59 """get_recommendations returns the three static rows without any backend I/O."""
60 provider = _make_provider()
61
62 rows = await provider.get_recommendations()
63
64 _assert_no_backend_calls(provider)
65 assert [row.item_id for row in rows] == [
66 "guess_recommend",
67 "new_songs",
68 "recommended_playlists",
69 ]
70 assert all(row.provider == "qqmusic_instance" for row in rows)
71 assert all(len(row.items) == 0 for row in rows)
72 assert [row.translation_key for row in rows] == [
73 "recommended_tracks",
74 "recommended_new_tracks",
75 "recommended_playlists",
76 ]
77 assert [row.icon for row in rows] == [
78 "mdi-lightbulb-on-outline",
79 "mdi-music-note-plus",
80 "mdi-playlist-music",
81 ]
82
83
84@pytest.mark.asyncio
85async def test_get_recommendation_items_guess_recommend() -> None:
86 """guess_recommend triggers only the guess fetch and returns its tracks."""
87 provider = _make_provider()
88
89 items = await provider.get_recommendation_items("guess_recommend")
90
91 provider._qq_recommend.get_guess_recommend.assert_awaited_once()
92 provider._qq_recommend.get_radar_recommend.assert_not_awaited()
93 provider._qq_recommend.get_recommend_newsong.assert_not_awaited()
94 provider._qq_recommend.get_recommend_songlist.assert_not_awaited()
95 assert [item.item_id for item in items] == ["t1"]
96
97
98@pytest.mark.asyncio
99async def test_get_recommendation_items_guess_falls_back_to_radar() -> None:
100 """An empty guess payload falls back to the radar fetch within the same row."""
101 provider = _make_provider()
102 provider._qq_recommend.get_guess_recommend.return_value = {"songs": []}
103
104 items = await provider.get_recommendation_items("guess_recommend")
105
106 provider._qq_recommend.get_guess_recommend.assert_awaited_once()
107 provider._qq_recommend.get_radar_recommend.assert_awaited_once()
108 provider._qq_recommend.get_recommend_newsong.assert_not_awaited()
109 provider._qq_recommend.get_recommend_songlist.assert_not_awaited()
110 assert [item.item_id for item in items] == ["t2"]
111
112
113@pytest.mark.asyncio
114async def test_get_recommendation_items_new_songs() -> None:
115 """new_songs triggers only the newsong fetch and returns its tracks."""
116 provider = _make_provider()
117
118 items = await provider.get_recommendation_items("new_songs")
119
120 provider._qq_recommend.get_recommend_newsong.assert_awaited_once()
121 provider._qq_recommend.get_guess_recommend.assert_not_awaited()
122 provider._qq_recommend.get_radar_recommend.assert_not_awaited()
123 provider._qq_recommend.get_recommend_songlist.assert_not_awaited()
124 assert [item.item_id for item in items] == ["t3"]
125
126
127@pytest.mark.asyncio
128async def test_get_recommendation_items_recommended_playlists() -> None:
129 """recommended_playlists triggers only the songlist fetch and returns its playlists."""
130 provider = _make_provider()
131
132 items = await provider.get_recommendation_items("recommended_playlists")
133
134 provider._qq_recommend.get_recommend_songlist.assert_awaited_once()
135 provider._qq_recommend.get_guess_recommend.assert_not_awaited()
136 provider._qq_recommend.get_radar_recommend.assert_not_awaited()
137 provider._qq_recommend.get_recommend_newsong.assert_not_awaited()
138 assert len(items) == 1
139 assert items[0].name == "Recommended List"
140
141
142@pytest.mark.asyncio
143async def test_get_recommendation_items_unknown_id_returns_empty() -> None:
144 """An unknown row item_id returns an empty UniqueList without backend calls."""
145 provider = _make_provider()
146
147 items = await provider.get_recommendation_items("bogus_row")
148
149 _assert_no_backend_calls(provider)
150 assert isinstance(items, UniqueList)
151 assert len(items) == 0
152
153
154@pytest.mark.asyncio
155async def test_get_recommendation_items_uses_payload_cache() -> None:
156 """A repeated items call within the TTL is served from the payload cache."""
157 provider = _make_provider()
158
159 first = await provider.get_recommendation_items("new_songs")
160 second = await provider.get_recommendation_items("new_songs")
161
162 provider._qq_recommend.get_recommend_newsong.assert_awaited_once()
163 assert [item.item_id for item in first] == [item.item_id for item in second] == ["t3"]
164