/
/
/
1"""Test the YouSee Musik two-method recommendations contract."""
2
3from __future__ import annotations
4
5import asyncio
6from typing import TYPE_CHECKING
7from unittest.mock import AsyncMock, Mock
8
9import pytest
10from music_assistant_models.media_items import UniqueList
11
12from music_assistant.providers.yousee.provider import YouSeeMusikProvider
13from music_assistant.providers.yousee.recommendations import YouSeeRecommendationsManager
14
15if TYPE_CHECKING:
16 from typing import Any
17
18GRAPHQL_RESULT = {
19 "data": {
20 "me": {
21 "recommendations": {
22 "albumRecommendations": {
23 "id": "discoveralbums",
24 "title": "Discover albums",
25 "subtitle": "Albums picked for you",
26 "albums": {
27 "items": [
28 {
29 "id": "album1",
30 "title": "Album One",
31 "artist": {"id": "artist1", "title": "Artist One"},
32 }
33 ]
34 },
35 },
36 "trackRecommendations": {
37 "id": "discovertracks",
38 "title": "Discover tracks",
39 "subtitle": "Tracks picked for you",
40 "tracks": {
41 "items": [
42 {
43 "id": "track1",
44 "title": "Track One",
45 "duration": 200,
46 "artist": {"id": "artist1", "title": "Artist One"},
47 }
48 ]
49 },
50 },
51 "weeklyDiscoveries": {
52 "id": "weeklyDiscoveries",
53 "title": "Weekly discoveries",
54 "subtitle": "Fresh finds",
55 "tracks": {
56 "items": [
57 {
58 "id": "track2",
59 "title": "Track Two",
60 "duration": 180,
61 "artist": {"id": "artist2", "title": "Artist Two"},
62 }
63 ]
64 },
65 },
66 }
67 }
68 }
69}
70
71
72def _stub_api(provider: YouSeeMusikProvider) -> Mock:
73 """Attach a stubbed API client and a real recommendations manager to the provider."""
74 api = Mock()
75 api.post_graphql = AsyncMock(return_value=GRAPHQL_RESULT)
76 provider.api = api
77 provider.auth = Mock()
78 provider.recommendations_manager = YouSeeRecommendationsManager(provider)
79 return api
80
81
82@pytest.mark.asyncio
83async def test_get_recommendations_returns_rows_without_items(
84 provider: YouSeeMusikProvider,
85) -> None:
86 """get_recommendations returns the payload's rows, stripped of items."""
87 api = _stub_api(provider)
88
89 rows = await provider.get_recommendations()
90
91 api.post_graphql.assert_awaited_once()
92 assert [row.item_id for row in rows] == [
93 "discoveralbums",
94 "discovertracks",
95 "weeklyDiscoveries",
96 ]
97 assert [row.name for row in rows] == [
98 "Discover albums",
99 "Discover tracks",
100 "Weekly discoveries",
101 ]
102 assert [row.subtitle for row in rows] == [
103 "Albums picked for you",
104 "Tracks picked for you",
105 "Fresh finds",
106 ]
107 assert all(row.provider == provider.instance_id for row in rows)
108 assert all(not row.items for row in rows)
109
110
111@pytest.mark.asyncio
112async def test_rows_and_items_share_one_payload_fetch(
113 provider: YouSeeMusikProvider,
114 background_tasks: list[asyncio.Future[Any]],
115) -> None:
116 """The rows call and subsequent per-row items calls trigger exactly one GraphQL fetch."""
117 api = _stub_api(provider)
118
119 rows = await provider.get_recommendations()
120 # let the background cache-store task complete before the items calls
121 await asyncio.gather(*background_tasks)
122 album_items = await provider.get_recommendation_items("discoveralbums")
123 track_items = await provider.get_recommendation_items("discovertracks")
124
125 api.post_graphql.assert_awaited_once()
126 assert len(rows) == 3
127 assert [item.item_id for item in album_items] == ["album1"]
128 assert [item.name for item in album_items] == ["Album One"]
129 assert [item.item_id for item in track_items] == ["track1"]
130 assert [item.name for item in track_items] == ["Track One"]
131
132
133@pytest.mark.asyncio
134async def test_get_recommendation_items_returns_row_items(
135 provider: YouSeeMusikProvider,
136) -> None:
137 """A cold items call fetches the payload and returns only the requested row's items."""
138 api = _stub_api(provider)
139
140 items = await provider.get_recommendation_items("weeklyDiscoveries")
141
142 api.post_graphql.assert_awaited_once()
143 assert [item.item_id for item in items] == ["track2"]
144 assert [item.name for item in items] == ["Track Two"]
145
146
147@pytest.mark.asyncio
148async def test_get_recommendation_items_unknown_id_returns_empty(
149 provider: YouSeeMusikProvider,
150) -> None:
151 """An unknown row item_id yields an empty UniqueList."""
152 _stub_api(provider)
153
154 result = await provider.get_recommendation_items("bogus_row")
155
156 assert isinstance(result, UniqueList)
157 assert len(result) == 0
158