music-assistant-server

1"""Test Tidal recommendations: payload manager + two-method provider contract."""
2
3from __future__ import annotations
4
5import asyncio
6from typing import Any
7from unittest.mock import AsyncMock, Mock, patch
8
9import pytest
10from music_assistant_models.enums import MediaType
11from music_assistant_models.errors import ResourceTemporarilyUnavailable
12from music_assistant_models.media_items import (
13    ItemMapping,
14    RecommendationFolder,
15    UniqueList,
16)
17
18from music_assistant.providers.tidal.provider import TidalProvider
19from music_assistant.providers.tidal.recommendations import TidalRecommendationManager
20
21INSTANCE_ID = "tidal_test"
22
23
24@pytest.fixture
25def provider_mock() -> Mock:
26    """Return a mock provider for manager-level tests."""
27    provider = Mock()
28    provider.domain = "tidal"
29    provider.instance_id = "tidal_instance"
30    provider.auth.user_id = "12345"
31    provider.auth.country_code = "US"
32    provider.api = AsyncMock()
33    provider.logger = Mock()
34
35    # Mock mass
36    provider.mass = Mock()
37    provider.mass.config.get_provider_configs = AsyncMock(return_value=[])
38    provider.mass.metadata.locale = "en_US"
39    provider.mass.cache.set = AsyncMock()
40
41    return provider
42
43
44@pytest.fixture
45def recommendation_manager(provider_mock: Mock) -> TidalRecommendationManager:
46    """Return a TidalRecommendationManager instance."""
47    return TidalRecommendationManager(provider_mock)
48
49
50@pytest.fixture
51def mass_mock() -> Mock:
52    """Return a mock MusicAssistant with a dict-backed cache and real background tasks."""
53    mass = Mock()
54    mass.http_session = AsyncMock()
55    mass.metadata.locale = "en_US"
56
57    cache_store: dict[str, Any] = {}
58    background_tasks: list[asyncio.Future[Any]] = []
59
60    async def _cache_get(key: str, **_kwargs: Any) -> tuple[Any, bool, bool]:
61        if key in cache_store:
62            return cache_store[key], True, True
63        return None, False, False
64
65    async def _cache_set(key: str, data: Any, **_kwargs: Any) -> None:
66        cache_store[key] = data
67
68    def _create_task(target: Any, *_args: Any, **_kwargs: Any) -> asyncio.Future[Any]:
69        task: asyncio.Future[Any] = asyncio.ensure_future(target)
70        background_tasks.append(task)
71        return task
72
73    mass.cache.get = AsyncMock(return_value=None)
74    mass.cache.get_with_freshness = AsyncMock(side_effect=_cache_get)
75    mass.cache.set = AsyncMock(side_effect=_cache_set)
76    mass.create_task = Mock(side_effect=_create_task)
77    mass.background_tasks = background_tasks
78    return mass
79
80
81@pytest.fixture
82def provider(mass_mock: Mock) -> TidalProvider:
83    """Return a TidalProvider instance for provider-level tests."""
84    manifest = Mock()
85    manifest.domain = "tidal"
86
87    config = Mock()
88    config.name = "Tidal Test"
89    config.instance_id = INSTANCE_ID
90    config.enabled = True
91    config.get_value.side_effect = lambda key: "INFO" if "log" in key else None
92
93    return TidalProvider(mass_mock, manifest, config)
94
95
96def _make_payload() -> list[RecommendationFolder]:
97    """Build a two-folder payload, shaped like the manager's merged page modules."""
98    return [
99        RecommendationFolder(
100            item_id="12345_for_you",
101            provider=INSTANCE_ID,
102            name="For You",
103            translation_key="12345_for_you",
104            icon="mdi-playlist-music",
105            subtitle="From Home • 1 items",
106            items=UniqueList(
107                [
108                    ItemMapping(
109                        media_type=MediaType.PLAYLIST,
110                        item_id="p1",
111                        provider=INSTANCE_ID,
112                        name="Daily Mix",
113                    )
114                ]
115            ),
116        ),
117        RecommendationFolder(
118            item_id="12345_new_albums",
119            provider=INSTANCE_ID,
120            name="New Albums",
121            translation_key="12345_new_albums",
122            icon="mdi-album",
123            subtitle="From Explore New Music • 1 items",
124            items=UniqueList(
125                [
126                    ItemMapping(
127                        media_type=MediaType.ALBUM,
128                        item_id="a1",
129                        provider=INSTANCE_ID,
130                        name="New Album",
131                    )
132                ]
133            ),
134        ),
135    ]
136
137
138async def test_get_recommendations_returns_rows_without_items(provider: TidalProvider) -> None:
139    """get_recommendations returns the payload folders as rows, stripped of items."""
140    payload = _make_payload()
141    with patch.object(
142        provider.recommendations_manager, "get_recommendations", new_callable=AsyncMock
143    ) as mock_fetch:
144        mock_fetch.return_value = payload
145
146        rows = await provider.get_recommendations()
147
148    mock_fetch.assert_awaited_once()
149    assert [row.item_id for row in rows] == ["12345_for_you", "12345_new_albums"]
150    for row, folder in zip(rows, payload, strict=True):
151        assert row.name == folder.name
152        assert row.translation_key == folder.translation_key
153        assert row.icon == folder.icon
154        assert len(row.items) == 0
155
156
157async def test_rows_and_items_share_one_payload_fetch(
158    provider: TidalProvider, mass_mock: Mock
159) -> None:
160    """Rows and both items calls are served from a single cached payload fetch."""
161    payload = _make_payload()
162    with patch.object(
163        provider.recommendations_manager, "get_recommendations", new_callable=AsyncMock
164    ) as mock_fetch:
165        mock_fetch.return_value = payload
166
167        await provider.get_recommendations()
168        # let the background cache-store task complete before the next calls
169        await asyncio.gather(*mass_mock.background_tasks)
170        items_for_you = await provider.get_recommendation_items("12345_for_you")
171        items_new_albums = await provider.get_recommendation_items("12345_new_albums")
172
173    mock_fetch.assert_awaited_once()
174    assert items_for_you == payload[0].items
175    assert items_new_albums == payload[1].items
176
177
178async def test_get_recommendation_items_unknown_id_returns_empty(
179    provider: TidalProvider,
180) -> None:
181    """An unknown row item_id yields an empty UniqueList."""
182    with patch.object(
183        provider.recommendations_manager, "get_recommendations", new_callable=AsyncMock
184    ) as mock_fetch:
185        mock_fetch.return_value = _make_payload()
186
187        result = await provider.get_recommendation_items("bogus_row")
188
189    assert isinstance(result, UniqueList)
190    assert len(result) == 0
191
192
193@pytest.mark.usefixtures("provider_mock")
194async def test_manager_builds_payload_folders(
195    recommendation_manager: TidalRecommendationManager,
196) -> None:
197    """The manager merges page modules into folders with items and stable item_ids."""
198    # Mock get_page_content to return a mock parser
199    mock_parser = Mock()
200    mock_parser.modules = [{"title": "Test Module"}]
201    mock_parser.get_module_items.return_value = (
202        [Mock(item_id="rec_1", name="Recommendation 1")],
203        MediaType.PLAYLIST,
204    )
205
206    with patch.object(
207        recommendation_manager, "get_page_content", new_callable=AsyncMock
208    ) as mock_get_page:
209        mock_get_page.return_value = mock_parser
210
211        recommendations = await recommendation_manager.get_recommendations()
212
213        assert len(recommendations) == 1
214        assert recommendations[0].name == "Test Module"
215        assert recommendations[0].item_id == "12345_test_module"
216        assert len(recommendations[0].items) == 1
217
218        # Should fetch pages
219        assert mock_get_page.call_count >= 1
220
221
222async def test_manager_strips_at_symbol_when_multiple_instances(
223    recommendation_manager: TidalRecommendationManager, provider_mock: Mock
224) -> None:
225    """Test that username is included and '@' is stripped when multiple instances exist."""
226    provider_mock.auth.user = Mock(profile_name="[email protected]", user_name=None)
227
228    provider_mock.mass.config.get_provider_configs = AsyncMock(
229        return_value=[
230            Mock(domain="tidal", instance_id="tidal_instance_1"),
231            Mock(domain="tidal", instance_id="tidal_instance_2"),
232            Mock(domain="other", instance_id="other_instance"),
233        ]
234    )
235
236    parser_with_module = Mock()
237    parser_with_module.modules = [{"title": "Test Module"}]
238    parser_with_module.get_module_items.return_value = (
239        [Mock(item_id="rec_1", name="Recommendation 1")],
240        MediaType.PLAYLIST,
241    )
242
243    parser_empty = Mock()
244    parser_empty.modules = []
245    parser_empty.get_module_items = Mock()
246
247    with patch.object(
248        recommendation_manager, "get_page_content", new_callable=AsyncMock
249    ) as mock_get_page:
250        # Only first page returns the module, remaining pages return no modules
251        mock_get_page.side_effect = [parser_with_module] + [parser_empty] * 4
252
253        recommendations = await recommendation_manager.get_recommendations()
254
255        assert len(recommendations) == 1
256        assert recommendations[0].name == "Test Module (john)"
257        assert "@" not in recommendations[0].name
258        assert len(recommendations[0].items) == 1
259
260
261async def test_get_page_content(
262    recommendation_manager: TidalRecommendationManager, provider_mock: Mock
263) -> None:
264    """Test get_page_content."""
265    with patch(
266        "music_assistant.providers.tidal.recommendations.TidalPageParser"
267    ) as mock_parser_cls:
268        # Configure from_cache to be async and return None
269        mock_parser_cls.from_cache = AsyncMock(return_value=None)
270
271        # Configure parser instance
272        mock_parser_instance = mock_parser_cls.return_value
273        mock_parser_instance.parse_page_structure = Mock()  # Ensure it's a synchronous mock
274        mock_parser_instance.to_cache = Mock(
275            return_value={"module_map": [], "content_map": {}, "parsed_at": 1234567890}
276        )
277
278        # Mock API response
279        provider_mock.api.get.return_value = {"rows": []}
280
281        parser = await recommendation_manager.get_page_content("pages/home")
282
283        assert parser == mock_parser_instance
284
285        # Should check cache
286        mock_parser_cls.from_cache.assert_called_with(provider_mock, "pages/home")
287
288        # Should fetch from API
289        provider_mock.api.get.assert_called()
290
291        # Should parse structure
292        mock_parser_instance.parse_page_structure.assert_called()
293
294        # Should cache result
295        provider_mock.mass.cache.set.assert_called()
296
297
298async def test_get_page_content_propagates_api_errors(
299    recommendation_manager: TidalRecommendationManager, provider_mock: Mock
300) -> None:
301    """Test API failures propagate so an empty result is never cached."""
302    with patch(
303        "music_assistant.providers.tidal.recommendations.TidalPageParser"
304    ) as mock_parser_cls:
305        mock_parser_cls.from_cache = AsyncMock(return_value=None)
306        provider_mock.api.get.side_effect = ResourceTemporarilyUnavailable("API error")
307
308        with pytest.raises(ResourceTemporarilyUnavailable):
309            await recommendation_manager.get_page_content("pages/home")
310
311        provider_mock.mass.cache.set.assert_not_called()
312