/
/
/
1"""Test the TuneIn two-method recommendations contract."""
2
3from __future__ import annotations
4
5import asyncio
6from typing import Any
7from unittest.mock import AsyncMock, Mock
8
9import pytest
10from music_assistant_models.media_items import UniqueList
11
12from music_assistant.providers.tunein import TuneInProvider
13from tests.common import use_real_create_task
14
15TRENDING_DATA = {
16 "body": [
17 {
18 "type": "audio",
19 "preset_id": "s100",
20 "text": "Station One",
21 "image": "http://cdn.example/1.png",
22 },
23 {"type": "audio", "preset_id": "s200", "text": "Station Two"},
24 {"type": "link", "text": "Some Category", "URL": "http://cdn.example/category"},
25 ]
26}
27
28
29def _stub_get_data(provider: TuneInProvider) -> AsyncMock:
30 """Attach a __get_data stub that returns canned trending data."""
31 mock = AsyncMock(return_value=TRENDING_DATA)
32 provider._TuneInProvider__get_data = mock # type: ignore[attr-defined]
33 return mock
34
35
36def _install_cache_mocks(provider: TuneInProvider) -> None:
37 """Make the @use_cache decorator treat every call as a cache miss."""
38 provider.mass.cache.get_with_freshness = AsyncMock( # type: ignore[method-assign]
39 return_value=(None, False, False)
40 )
41 provider.mass.cache.set = AsyncMock() # type: ignore[method-assign]
42 use_real_create_task(provider.mass)
43
44
45def _install_fake_cache(provider: TuneInProvider) -> list[asyncio.Future[Any]]:
46 """
47 Back the real @use_cache decorator with a dict-backed fake cache.
48
49 Mirrors production: set() stores the serialized (to_dict) folder and, since the
50 caller passes base_class=RecommendationFolder, get_with_freshness reconstructs it
51 via RecommendationFolder.from_dict on a hit.
52 """
53 store: dict[str, Any] = {}
54 background_tasks: list[asyncio.Future[Any]] = []
55
56 async def _cache_get(key: str, **kwargs: Any) -> tuple[Any, bool, bool]:
57 if key not in store:
58 return None, False, False
59 data = store[key]
60 base_class = kwargs.get("base_class")
61 if base_class is not None and data is not None:
62 return base_class.from_dict(data), True, True
63 return data, True, True
64
65 async def _cache_set(key: str, data: Any, **_kwargs: Any) -> None:
66 store[key] = data.to_dict() if data is not None else None
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 provider.mass.cache.get_with_freshness = AsyncMock( # type: ignore[method-assign]
74 side_effect=_cache_get
75 )
76 provider.mass.cache.set = AsyncMock(side_effect=_cache_set) # type: ignore[method-assign]
77 provider.mass.create_task = Mock(side_effect=_create_task) # type: ignore[method-assign]
78 return background_tasks
79
80
81@pytest.mark.asyncio
82async def test_get_recommendations_static_row_without_backend_calls(
83 provider: TuneInProvider,
84) -> None:
85 """get_recommendations returns the static trending row without any backend call."""
86 api_mock = _stub_get_data(provider)
87
88 result = await provider.get_recommendations()
89
90 api_mock.assert_not_awaited()
91 assert len(result) == 1
92 row = result[0]
93 assert row.item_id == "trending"
94 assert row.provider == provider.instance_id
95 assert row.name == "Trending"
96 assert row.translation_key == "trending_stations"
97 assert len(row.items) == 0
98
99
100@pytest.mark.asyncio
101async def test_get_recommendation_items_trending_fetches_stations(
102 provider: TuneInProvider,
103) -> None:
104 """get_recommendation_items("trending") fetches the trending feed and parses its stations."""
105 _install_cache_mocks(provider)
106 api_mock = _stub_get_data(provider)
107
108 result = await provider.get_recommendation_items("trending")
109
110 api_mock.assert_awaited_once()
111 assert api_mock.call_args.args == ("Browse.ashx",)
112 assert api_mock.call_args.kwargs == {"c": "trending"}
113 # only the audio entries with a preset_id become stations (order is randomized)
114 assert {item.item_id for item in result} == {"s100", "s200"}
115
116
117@pytest.mark.asyncio
118async def test_get_recommendation_items_served_from_warm_cache(
119 provider: TuneInProvider,
120) -> None:
121 """A second items call is served from the cache and still returns the items."""
122 background_tasks = _install_fake_cache(provider)
123 api_mock = _stub_get_data(provider)
124
125 first = await provider.get_recommendation_items("trending")
126 # let the background cache-store task complete before the second call
127 await asyncio.gather(*background_tasks)
128 second = await provider.get_recommendation_items("trending")
129
130 # the backend was fetched exactly once: the second call was a cache hit
131 api_mock.assert_awaited_once()
132 assert {item.item_id for item in first} == {"s100", "s200"}
133 assert {item.item_id for item in second} == {"s100", "s200"}
134
135
136@pytest.mark.asyncio
137async def test_get_recommendation_items_unknown_id_returns_empty(
138 provider: TuneInProvider,
139) -> None:
140 """An unknown row item_id yields an empty list without any backend call."""
141 _install_cache_mocks(provider)
142 api_mock = _stub_get_data(provider)
143
144 result = await provider.get_recommendation_items("bogus_row")
145
146 api_mock.assert_not_awaited()
147 assert isinstance(result, UniqueList)
148 assert len(result) == 0
149