/
/
/
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
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 client = Mock()
54 client.personal.get_experience_menu = AsyncMock(return_value=menu)
55 provider.client = client
56 conversions = dict(zip(sub_items, payload, strict=True))
57 adaptor = Mock()
58 adaptor.new_object = AsyncMock(side_effect=lambda item, **_kwargs: conversions[item])
59 provider.adaptor = adaptor
60 return client
61
62
63def _install_cache_mocks(provider: BBCSoundsProvider) -> list[asyncio.Future[Any]]:
64 """Back the mixin's @use_cache with a dict store; return the background store tasks."""
65 store: dict[str, Any] = {}
66 background_tasks: list[asyncio.Future[Any]] = []
67
68 async def _cache_get(key: str, **_kwargs: Any) -> tuple[Any, bool, bool]:
69 if key in store:
70 return store[key], True, True
71 return None, False, False
72
73 async def _cache_set(key: str, data: Any, **_kwargs: Any) -> None:
74 store[key] = data
75
76 def _create_task(target: Any, *_args: Any, **_kwargs: Any) -> asyncio.Future[Any]:
77 task: asyncio.Future[Any] = asyncio.ensure_future(target)
78 background_tasks.append(task)
79 return task
80
81 provider.mass.cache.get_with_freshness = AsyncMock( # type: ignore[method-assign]
82 side_effect=_cache_get
83 )
84 provider.mass.cache.set = AsyncMock(side_effect=_cache_set) # type: ignore[method-assign]
85 provider.mass.create_task = Mock(side_effect=_create_task) # type: ignore[method-assign]
86 return background_tasks
87
88
89@pytest.mark.asyncio
90async def test_get_recommendations_returns_rows_without_items(
91 provider: BBCSoundsProvider,
92) -> None:
93 """get_recommendations returns the payload folders as rows, stripped of items."""
94 _install_cache_mocks(provider)
95 client = _stub_api(provider, _make_payload())
96
97 rows = await provider.get_recommendations()
98
99 client.personal.get_experience_menu.assert_awaited_once_with(
100 recommendations=MenuRecommendationOptions.ONLY
101 )
102 assert [row.item_id for row in rows] == [MUSIC_ROW_ID, PODCAST_ROW_ID]
103 assert all(len(row.items) == 0 for row in rows)
104
105
106@pytest.mark.asyncio
107async def test_rows_and_items_share_one_payload_fetch(
108 provider: BBCSoundsProvider,
109) -> None:
110 """Items calls for both rows are served from the payload cached by the rows call."""
111 background_tasks = _install_cache_mocks(provider)
112 payload = _make_payload()
113 client = _stub_api(provider, payload)
114
115 rows = await provider.get_recommendations()
116 # let the background cache-store task complete before the items calls
117 await asyncio.gather(*background_tasks)
118 music_items = await provider.get_recommendation_items(MUSIC_ROW_ID)
119 podcast_items = await provider.get_recommendation_items(PODCAST_ROW_ID)
120
121 client.personal.get_experience_menu.assert_awaited_once()
122 assert len(rows) == 2
123 assert music_items == payload[0].items
124 assert podcast_items == payload[1].items
125
126
127@pytest.mark.asyncio
128async def test_get_recommendation_items_unknown_id_returns_empty(
129 provider: BBCSoundsProvider,
130) -> None:
131 """An item_id not present in the payload yields an empty UniqueList."""
132 _install_cache_mocks(provider)
133 _stub_api(provider, _make_payload())
134
135 result = await provider.get_recommendation_items("bogus_row")
136
137 assert isinstance(result, UniqueList)
138 assert len(result) == 0
139
140
141@pytest.mark.asyncio
142async def test_not_logged_in_returns_empty_without_backend_calls(
143 provider: BBCSoundsProvider,
144) -> None:
145 """Without a login, both methods return empty and never touch the backend."""
146 _install_cache_mocks(provider)
147 client = _stub_api(provider, _make_payload())
148 provider.logged_in = False
149
150 rows = await provider.get_recommendations()
151 items = await provider.get_recommendation_items(MUSIC_ROW_ID)
152
153 client.personal.get_experience_menu.assert_not_awaited()
154 assert rows == []
155 assert isinstance(items, UniqueList)
156 assert len(items) == 0
157