/
/
/
1"""Tests for the Soundcloud 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.unique_list import UniqueList
11
12from music_assistant.providers.soundcloud import SoundcloudMusicProvider
13
14MIXED_DATA = {
15 "collection": [
16 {
17 "id": "trending-music",
18 "title": "Trending",
19 "items": {
20 "collection": [
21 {"kind": "system-playlist", "id": "sp1", "title": "Daily Mix"},
22 ]
23 },
24 },
25 ]
26}
27FEED_DATA = {
28 "collection": [
29 {
30 "type": "track",
31 "track": {
32 "id": 111,
33 "title": "Feed Track",
34 "duration": 180000,
35 "permalink_url": "https://soundcloud.com/artist/feed-track",
36 "user": {"id": 42},
37 },
38 },
39 ]
40}
41USER_DATA = {"id": 42, "username": "Artist", "permalink": "artist"}
42
43
44def _stub_api(provider: SoundcloudMusicProvider) -> Mock:
45 """Attach a stubbed soundcloud client with canned responses."""
46 api = Mock()
47 api.get_mixed_selection = AsyncMock(return_value=MIXED_DATA)
48 api.get_subscribe_feed = AsyncMock(return_value=FEED_DATA)
49 api.get_user_details = AsyncMock(return_value=USER_DATA)
50 provider._soundcloud = api
51 return api
52
53
54def _install_fake_cache(provider: SoundcloudMusicProvider) -> list[asyncio.Future[Any]]:
55 """Back @use_cache with a dict store; return the background store tasks."""
56 store: dict[str, Any] = {}
57 background_tasks: list[asyncio.Future[Any]] = []
58
59 async def _cache_get(key: str, **_kwargs: Any) -> tuple[Any, bool, bool]:
60 if key in store:
61 return store[key], True, True
62 return None, False, False
63
64 async def _cache_set(key: str, data: Any, **_kwargs: Any) -> None:
65 store[key] = data
66
67 def _create_task(target: Any, *_args: Any, **_kwargs: Any) -> asyncio.Future[Any]:
68 task: asyncio.Future[Any] = asyncio.ensure_future(target)
69 background_tasks.append(task)
70 return task
71
72 provider.mass.cache.get_with_freshness = AsyncMock( # type: ignore[method-assign]
73 side_effect=_cache_get
74 )
75 provider.mass.cache.set = AsyncMock(side_effect=_cache_set) # type: ignore[method-assign]
76 provider.mass.create_task = Mock(side_effect=_create_task) # type: ignore[method-assign]
77 return background_tasks
78
79
80@pytest.mark.asyncio
81async def test_get_recommendations_returns_rows_without_items(
82 provider: SoundcloudMusicProvider,
83) -> None:
84 """get_recommendations returns payload rows plus the static feed row, all without items."""
85 background_tasks = _install_fake_cache(provider)
86 api = _stub_api(provider)
87
88 rows = await provider.get_recommendations()
89
90 api.get_mixed_selection.assert_awaited_once_with(40)
91 api.get_subscribe_feed.assert_not_awaited()
92 assert [row.item_id for row in rows] == [
93 f"{provider.instance_id}_trending-music",
94 f"{provider.instance_id}_sc_subscribed_feed",
95 ]
96 assert all(len(row.items) == 0 for row in rows)
97 trending, feed = rows
98 assert trending.name == "Trending"
99 assert trending.icon == "mdi-playlist-music"
100 assert trending.provider == provider.instance_id
101 assert feed.name == "SoundCloud Feed"
102 assert feed.translation_key == "soundcloud_feed"
103 assert feed.icon == "mdi-rss"
104 assert feed.provider == provider.instance_id
105 await asyncio.gather(*background_tasks)
106
107
108@pytest.mark.asyncio
109async def test_rows_and_items_share_one_payload_fetch(
110 provider: SoundcloudMusicProvider,
111) -> None:
112 """A rows call followed by an items call is served from one cached payload fetch."""
113 background_tasks = _install_fake_cache(provider)
114 api = _stub_api(provider)
115
116 await provider.get_recommendations()
117 # let the background cache-store task complete before the items call
118 await asyncio.gather(*background_tasks)
119 items = await provider.get_recommendation_items(f"{provider.instance_id}_trending-music")
120
121 api.get_mixed_selection.assert_awaited_once_with(40)
122 assert [item.item_id for item in items] == ["sp1"]
123
124
125@pytest.mark.asyncio
126async def test_get_recommendation_items_collection_row(
127 provider: SoundcloudMusicProvider,
128) -> None:
129 """Items for a mixed-selection row trigger only the payload fetch, not the feed."""
130 background_tasks = _install_fake_cache(provider)
131 api = _stub_api(provider)
132
133 items = await provider.get_recommendation_items(f"{provider.instance_id}_trending-music")
134
135 api.get_mixed_selection.assert_awaited_once_with(40)
136 api.get_subscribe_feed.assert_not_awaited()
137 assert [item.item_id for item in items] == ["sp1"]
138 assert items[0].name == "Daily Mix"
139 await asyncio.gather(*background_tasks)
140
141
142@pytest.mark.asyncio
143async def test_get_recommendation_items_feed_row(
144 provider: SoundcloudMusicProvider,
145) -> None:
146 """Items for the feed row trigger only the subscribed-feed fetch, not the payload."""
147 background_tasks = _install_fake_cache(provider)
148 api = _stub_api(provider)
149
150 items = await provider.get_recommendation_items(f"{provider.instance_id}_sc_subscribed_feed")
151
152 api.get_subscribe_feed.assert_awaited_once_with(40)
153 api.get_mixed_selection.assert_not_awaited()
154 assert isinstance(items, UniqueList)
155 assert [item.item_id for item in items] == ["111"]
156 assert items[0].name == "Feed Track"
157 await asyncio.gather(*background_tasks)
158
159
160@pytest.mark.asyncio
161async def test_get_recommendation_items_unknown_id_returns_empty(
162 provider: SoundcloudMusicProvider,
163) -> None:
164 """An unknown item_id yields an empty result."""
165 background_tasks = _install_fake_cache(provider)
166 _stub_api(provider)
167
168 result = await provider.get_recommendation_items("bogus_row")
169
170 assert isinstance(result, UniqueList)
171 assert len(result) == 0
172 await asyncio.gather(*background_tasks)
173