/
/
/
1"""Test YouTube Music's two-method recommendations contract."""
2
3from __future__ import annotations
4
5import asyncio
6from typing import Any
7from unittest.mock import AsyncMock, MagicMock, patch
8
9import pytest
10from music_assistant_models.enums import MediaType
11from music_assistant_models.media_items import ItemMapping, RecommendationFolder, UniqueList
12
13from music_assistant.providers.ytmusic import YoutubeMusicProvider
14
15GET_HOME_PATH = "music_assistant.providers.ytmusic.get_home"
16
17
18def _make_home_data() -> list[dict[str, Any]]:
19 """
20 Build two home sections: one with a parseable playlist item, one empty.
21
22 Enough to assert the server-derived folder item_ids and per-row item extraction.
23 Built fresh per call because the provider's parser mutates the raw item dicts.
24 """
25 return [
26 {
27 "title": "Listen again",
28 "contents": [{"playlistId": "PL123", "title": "Morning Mix"}],
29 },
30 {"title": "Quick picks", "contents": []},
31 ]
32
33
34class _FakeCache:
35 """Dict-backed stand-in for the mass cache and task scheduling used by @use_cache."""
36
37 def __init__(self) -> None:
38 self.store: dict[str, Any] = {}
39 self.background_tasks: list[asyncio.Future[Any]] = []
40
41 async def get_with_freshness(self, key: str, **kwargs: Any) -> tuple[Any, bool, bool]:
42 if key in self.store:
43 return self.store[key], True, True
44 return None, False, False
45
46 async def set(self, key: str, data: Any, **kwargs: Any) -> None:
47 self.store[key] = data
48
49 def create_task(self, target: Any, *args: Any, **kwargs: Any) -> asyncio.Future[Any]:
50 task: asyncio.Future[Any] = asyncio.ensure_future(target)
51 self.background_tasks.append(task)
52 return task
53
54 async def flush(self) -> None:
55 """Let scheduled background cache-store tasks complete."""
56 await asyncio.gather(*self.background_tasks)
57
58
59@pytest.fixture
60def fake_cache() -> _FakeCache:
61 """Return a fresh fake cache."""
62 return _FakeCache()
63
64
65@pytest.fixture
66def provider(fake_cache: _FakeCache) -> YoutubeMusicProvider:
67 """Return a YoutubeMusicProvider instance with mocked dependencies."""
68 mass = MagicMock()
69 manifest = MagicMock()
70 manifest.domain = "ytmusic"
71 config = MagicMock()
72 config.instance_id = "ytmusic--test"
73 config.get_value.return_value = "GLOBAL"
74 prov = YoutubeMusicProvider(mass, manifest, config)
75 prov._headers = {}
76 prov._yt_user = None
77 prov.language = "en"
78 mass.cache.get_with_freshness = fake_cache.get_with_freshness
79 mass.cache.set = fake_cache.set
80 mass.create_task = fake_cache.create_task
81 return prov
82
83
84def _stub_mixed_for_you(provider: YoutubeMusicProvider) -> AsyncMock:
85 """Attach a _get_mixed_for_you_folder stub returning a folder with one item."""
86 folder = RecommendationFolder(
87 name="Mixed for you",
88 item_id=f"{provider.instance_id}_mixed_for_you",
89 provider=provider.instance_id,
90 icon="mdi:shuffle-variant",
91 )
92 folder.items.append(
93 ItemMapping(
94 media_type=MediaType.PLAYLIST,
95 item_id="RDTMAK5uy_mix1",
96 provider=provider.instance_id,
97 name="My Mix 1",
98 )
99 )
100 mock = AsyncMock(return_value=folder)
101 provider._get_mixed_for_you_folder = mock # type: ignore[method-assign]
102 return mock
103
104
105async def test_get_recommendations_returns_rows_without_items(
106 provider: YoutubeMusicProvider,
107 fake_cache: _FakeCache,
108) -> None:
109 """The rows call returns home section rows plus the static mixed_for_you descriptor."""
110 mixed_mock = _stub_mixed_for_you(provider)
111 with patch(
112 GET_HOME_PATH, new_callable=AsyncMock, return_value=_make_home_data()
113 ) as get_home_mock:
114 result = await provider.get_recommendations()
115 await fake_cache.flush()
116
117 get_home_mock.assert_awaited_once()
118 # the mixed_for_you row is a static descriptor: its dedicated fetch must not run
119 mixed_mock.assert_not_awaited()
120 assert [f.item_id for f in result] == [
121 f"{provider.instance_id}_Listen again",
122 f"{provider.instance_id}_Quick picks",
123 f"{provider.instance_id}_mixed_for_you",
124 ]
125 assert all(len(f.items) == 0 for f in result)
126 assert result[0].name == "Listen again"
127 mixed_row = result[2]
128 assert mixed_row.name == "Mixed for you"
129 assert mixed_row.translation_key == "mixed_for_you"
130 assert mixed_row.icon == "mdi:shuffle-variant"
131 assert mixed_row.provider == provider.instance_id
132
133
134async def test_rows_and_items_share_one_cached_payload_fetch(
135 provider: YoutubeMusicProvider,
136 fake_cache: _FakeCache,
137) -> None:
138 """The rows call and a later items call are served from one cached payload fetch."""
139 with patch(
140 GET_HOME_PATH, new_callable=AsyncMock, return_value=_make_home_data()
141 ) as get_home_mock:
142 rows = await provider.get_recommendations()
143 # let the background cache-store task complete before the next call
144 await fake_cache.flush()
145 items = await provider.get_recommendation_items(f"{provider.instance_id}_Listen again")
146
147 assert get_home_mock.await_count == 1
148 assert len(rows) == 3
149 assert [item.item_id for item in items] == ["PL123"]
150
151
152async def test_get_recommendation_items_section_row(
153 provider: YoutubeMusicProvider,
154 fake_cache: _FakeCache,
155) -> None:
156 """An items call for a home section row fetches the payload, not the mixed_for_you chain."""
157 mixed_mock = _stub_mixed_for_you(provider)
158 with patch(
159 GET_HOME_PATH, new_callable=AsyncMock, return_value=_make_home_data()
160 ) as get_home_mock:
161 items = await provider.get_recommendation_items(f"{provider.instance_id}_Listen again")
162 await fake_cache.flush()
163
164 get_home_mock.assert_awaited_once()
165 mixed_mock.assert_not_awaited()
166 assert [item.item_id for item in items] == ["PL123"]
167 assert items[0].name == "Morning Mix"
168
169
170async def test_get_recommendation_items_mixed_for_you_row(
171 provider: YoutubeMusicProvider,
172) -> None:
173 """An items call for the mixed_for_you row uses its dedicated fetch, not get_home."""
174 mixed_mock = _stub_mixed_for_you(provider)
175 with patch(
176 GET_HOME_PATH, new_callable=AsyncMock, return_value=_make_home_data()
177 ) as get_home_mock:
178 items = await provider.get_recommendation_items(f"{provider.instance_id}_mixed_for_you")
179
180 get_home_mock.assert_not_awaited()
181 mixed_mock.assert_awaited_once()
182 assert [item.item_id for item in items] == ["RDTMAK5uy_mix1"]
183
184
185async def test_get_recommendation_items_unknown_id_returns_empty(
186 provider: YoutubeMusicProvider,
187 fake_cache: _FakeCache,
188) -> None:
189 """An items call for an unknown row id returns an empty UniqueList."""
190 with patch(GET_HOME_PATH, new_callable=AsyncMock, return_value=_make_home_data()):
191 result = await provider.get_recommendation_items(f"{provider.instance_id}_bogus")
192 await fake_cache.flush()
193
194 assert isinstance(result, UniqueList)
195 assert len(result) == 0
196