/
/
/
1"""Test BBC Sounds get_recommendations / get_recommendation_items via the payload mixin."""
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.enums import MediaType
11from music_assistant_models.media_items import ItemMapping, RecommendationFolder
12from music_assistant_models.unique_list import UniqueList
13from sounds import MenuRecommendationOptions, SoundsClient
14
15from music_assistant.providers.bbc_sounds import BBCSoundsProvider
16
17MUSIC_ROW_ID = "listen_to_music"
18PODCAST_ROW_ID = "recommended_podcasts"
19
20
21def _folder(item_id: str, name: str) -> RecommendationFolder:
22 """Build a payload folder with one item, as the adaptor would produce it."""
23 return RecommendationFolder(
24 item_id=item_id,
25 provider="bbc_sounds",
26 name=name,
27 items=UniqueList(
28 [
29 ItemMapping(
30 media_type=MediaType.TRACK,
31 item_id=f"{item_id}_item",
32 provider="bbc_sounds",
33 name=f"{name} item",
34 )
35 ]
36 ),
37 )
38
39
40def _make_payload() -> list[RecommendationFolder]:
41 """Build the two-folder payload used by all tests."""
42 return [
43 _folder(MUSIC_ROW_ID, "Listen to Music"),
44 _folder(PODCAST_ROW_ID, "Recommended Podcasts"),
45 ]
46
47
48def _stub_api(provider: BBCSoundsProvider, payload: list[RecommendationFolder]) -> Mock:
49 """Attach a stubbed sounds client + adaptor serving the given payload folders."""
50 sub_items = [Mock() for _ in payload]
51 menu = Mock()
52 menu.sub_items = sub_items
53 # spec_set so a renamed/removed client method fails here instead of at runtime
54 client = Mock(spec_set=SoundsClient)
55 client.get_menu.return_value = menu
56 provider.client = client
57 conversions = dict(zip(sub_items, payload, strict=True))
58 adaptor = Mock()
59 adaptor.new_object = AsyncMock(side_effect=lambda item, **_kwargs: conversions[item])
60 provider.adaptor = adaptor
61 return client
62
63
64def _install_cache_mocks(provider: BBCSoundsProvider) -> list[asyncio.Future[Any]]:
65 """Back the mixin's @use_cache with a dict store; return the background store tasks."""
66 store: dict[str, Any] = {}
67 background_tasks: list[asyncio.Future[Any]] = []
68
69 async def _cache_get(key: str, **_kwargs: Any) -> tuple[Any, bool, bool]:
70 if key in store:
71 return store[key], True, True
72 return None, False, False
73
74 async def _cache_set(key: str, data: Any, **_kwargs: Any) -> None:
75 store[key] = data
76
77 def _create_task(target: Any, *_args: Any, **_kwargs: Any) -> asyncio.Future[Any]:
78 task: asyncio.Future[Any] = asyncio.ensure_future(target)
79 background_tasks.append(task)
80 return task
81
82 provider.mass.cache.get_with_freshness = AsyncMock( # type: ignore[method-assign]
83 side_effect=_cache_get
84 )
85 provider.mass.cache.set = AsyncMock(side_effect=_cache_set) # type: ignore[method-assign]
86 provider.mass.create_task = Mock(side_effect=_create_task) # type: ignore[method-assign]
87 return background_tasks
88
89
90@pytest.mark.asyncio
91async def test_get_recommendations_returns_rows_without_items(
92 provider: BBCSoundsProvider,
93) -> None:
94 """get_recommendations returns the payload folders as rows, stripped of items."""
95 _install_cache_mocks(provider)
96 client = _stub_api(provider, _make_payload())
97
98 rows = await provider.get_recommendations()
99
100 client.get_menu.assert_awaited_once_with(recommendations=MenuRecommendationOptions.ONLY)
101 assert [row.item_id for row in rows] == [MUSIC_ROW_ID, PODCAST_ROW_ID]
102 assert all(len(row.items) == 0 for row in rows)
103
104
105@pytest.mark.asyncio
106async def test_rows_and_items_share_one_payload_fetch(
107 provider: BBCSoundsProvider,
108) -> None:
109 """Items calls for both rows are served from the payload cached by the rows call."""
110 background_tasks = _install_cache_mocks(provider)
111 payload = _make_payload()
112 client = _stub_api(provider, payload)
113
114 rows = await provider.get_recommendations()
115 # let the background cache-store task complete before the items calls
116 await asyncio.gather(*background_tasks)
117 music_items = await provider.get_recommendation_items(MUSIC_ROW_ID)
118 podcast_items = await provider.get_recommendation_items(PODCAST_ROW_ID)
119
120 client.get_menu.assert_awaited_once()
121 assert len(rows) == 2
122 assert music_items == payload[0].items
123 assert podcast_items == payload[1].items
124
125
126@pytest.mark.asyncio
127async def test_get_recommendation_items_unknown_id_returns_empty(
128 provider: BBCSoundsProvider,
129) -> None:
130 """An item_id not present in the payload yields an empty UniqueList."""
131 _install_cache_mocks(provider)
132 _stub_api(provider, _make_payload())
133
134 result = await provider.get_recommendation_items("bogus_row")
135
136 assert isinstance(result, UniqueList)
137 assert len(result) == 0
138
139
140@pytest.mark.asyncio
141async def test_not_logged_in_returns_empty_without_backend_calls(
142 provider: BBCSoundsProvider,
143) -> None:
144 """Without a login, both methods return empty and never touch the backend."""
145 _install_cache_mocks(provider)
146 client = _stub_api(provider, _make_payload())
147 provider.logged_in = False
148
149 rows = await provider.get_recommendations()
150 items = await provider.get_recommendation_items(MUSIC_ROW_ID)
151
152 client.get_menu.assert_not_awaited()
153 assert rows == []
154 assert isinstance(items, UniqueList)
155 assert len(items) == 0
156