/
/
/
1"""Tests for the Plex provider's two-method recommendations contract."""
2
3from __future__ import annotations
4
5import asyncio
6from typing import Any
7from unittest.mock import AsyncMock, MagicMock, Mock
8
9import pytest
10from music_assistant_models.enums import MediaType
11from music_assistant_models.media_items import ItemMapping, UniqueList
12
13from music_assistant.providers.plex import PlexProvider
14from music_assistant.providers.plex.helpers import SUPPORTED_FEATURES
15
16INSTANCE_ID = "plex_instance_1"
17HUB_RECENT = "hub.music.recentlyplayed"
18HUB_NEW = "hub.music.newreleases"
19
20
21class _FakeHubItem:
22 """Minimal Plex hub item stub."""
23
24 def __init__(self, type_: str, key: str, title: str) -> None:
25 self.type = type_
26 self.key = key
27 self.title = title
28
29
30class _FakeHub:
31 """Minimal Plex hub stub with preloaded partial items."""
32
33 def __init__(self, identifier: str, title: str, items: list[Any]) -> None:
34 self.hubIdentifier = identifier
35 self.title = title
36 self._partialItems = items
37
38
39def _make_hubs() -> list[_FakeHub]:
40 """Build two hubs with one parseable item each."""
41 return [
42 _FakeHub(
43 HUB_RECENT,
44 "Recently Played",
45 [_FakeHubItem("track", "/library/metadata/1", "Recent Track")],
46 ),
47 _FakeHub(
48 HUB_NEW,
49 "New Releases",
50 [_FakeHubItem("track", "/library/metadata/2", "New Track")],
51 ),
52 ]
53
54
55def _make_provider(hubs: list[_FakeHub]) -> tuple[Any, list[asyncio.Future[Any]]]:
56 """Create a PlexProvider with stubbed backend, dict-backed cache and canned hubs."""
57 mock_mass = MagicMock()
58 mock_config = MagicMock()
59 mock_config.instance_id = INSTANCE_ID
60 config_values: dict[str, Any] = {
61 "library_type": "music",
62 "log_level": "INFO",
63 "token": "local_auth",
64 "hub_items_limit": 10,
65 "extended_recommendations": False,
66 }
67 mock_config.get_value = lambda key: config_values.get(key)
68 mock_manifest = MagicMock()
69 mock_manifest.type = "music"
70 mock_manifest.domain = "plex"
71
72 provider = PlexProvider(mock_mass, mock_manifest, mock_config, SUPPORTED_FEATURES)
73 provider._plex_server = MagicMock()
74 provider._plex_library = MagicMock()
75 provider._plex_library.key = "10"
76 provider._plex_library.fetchItems = Mock(return_value=hubs)
77 provider._myplex_account = MagicMock()
78
79 async def _run_async(call: Any, *args: Any, **kwargs: Any) -> Any:
80 return call(*args, **kwargs)
81
82 provider._run_async = _run_async # type: ignore[method-assign]
83
84 async def _parse(item: Any) -> ItemMapping:
85 return ItemMapping(
86 media_type=MediaType.TRACK,
87 item_id=item.key,
88 provider=INSTANCE_ID,
89 name=item.title,
90 )
91
92 provider._parse = AsyncMock(side_effect=_parse) # type: ignore[method-assign]
93
94 # dict-backed fake cache so the mixin's @use_cache sees real hits/misses
95 cache_store: dict[str, Any] = {}
96 background_tasks: list[asyncio.Future[Any]] = []
97
98 async def _cache_get(key: str, **_kwargs: Any) -> tuple[Any, bool, bool]:
99 if key in cache_store:
100 return cache_store[key], True, True
101 return None, False, False
102
103 async def _cache_set(key: str, data: Any, **_kwargs: Any) -> None:
104 cache_store[key] = data
105
106 def _create_task(target: Any, *_args: Any, **_kwargs: Any) -> asyncio.Future[Any]:
107 task: asyncio.Future[Any] = asyncio.ensure_future(target)
108 background_tasks.append(task)
109 return task
110
111 provider.mass.cache.get_with_freshness = AsyncMock( # type: ignore[method-assign]
112 side_effect=_cache_get
113 )
114 provider.mass.cache.set = AsyncMock(side_effect=_cache_set) # type: ignore[method-assign]
115 provider.mass.create_task = Mock(side_effect=_create_task) # type: ignore[method-assign]
116
117 return provider, background_tasks
118
119
120@pytest.mark.asyncio
121async def test_get_recommendations_returns_hub_rows_without_items() -> None:
122 """Rows mirror the Plex hubs with stable item_ids and empty items."""
123 provider, background_tasks = _make_provider(_make_hubs())
124
125 rows = await provider.get_recommendations()
126 await asyncio.gather(*background_tasks)
127
128 assert [row.item_id for row in rows] == [
129 f"{INSTANCE_ID}_{HUB_RECENT}",
130 f"{INSTANCE_ID}_{HUB_NEW}",
131 ]
132 assert [row.name for row in rows] == ["Recently Played", "New Releases"]
133 assert all(row.provider == INSTANCE_ID for row in rows)
134 assert all(row.icon == "mdi-music" for row in rows)
135 assert all(len(row.items) == 0 for row in rows)
136 provider._plex_library.fetchItems.assert_called_once_with(
137 "/hubs/sections/10?count=10&includeStations=1"
138 )
139
140
141@pytest.mark.asyncio
142async def test_rows_and_items_served_from_single_payload_fetch() -> None:
143 """The rows call and both items calls share one cached hubs fetch."""
144 provider, background_tasks = _make_provider(_make_hubs())
145
146 await provider.get_recommendations()
147 # let the background cache-store task complete before the items calls
148 await asyncio.gather(*background_tasks)
149 items_recent = await provider.get_recommendation_items(f"{INSTANCE_ID}_{HUB_RECENT}")
150 items_new = await provider.get_recommendation_items(f"{INSTANCE_ID}_{HUB_NEW}")
151
152 assert provider._plex_library.fetchItems.call_count == 1
153 assert [item.name for item in items_recent] == ["Recent Track"]
154 assert [item.name for item in items_new] == ["New Track"]
155
156
157@pytest.mark.asyncio
158async def test_get_recommendation_items_unknown_id_returns_empty() -> None:
159 """An unknown item_id yields an empty UniqueList."""
160 provider, background_tasks = _make_provider(_make_hubs())
161
162 result = await provider.get_recommendation_items(f"{INSTANCE_ID}_bogus_hub")
163 await asyncio.gather(*background_tasks)
164
165 assert isinstance(result, UniqueList)
166 assert len(result) == 0
167
168
169@pytest.mark.asyncio
170async def test_hub_without_parseable_items_is_skipped() -> None:
171 """A hub whose items all fail to parse does not produce a row."""
172 hubs = _make_hubs()
173 hubs.append(_FakeHub("hub.music.stations", "Stations", [_FakeHubItem("station", "/s1", "S1")]))
174 provider, background_tasks = _make_provider(hubs)
175
176 async def _parse_tracks_only(item: Any) -> ItemMapping | None:
177 if item.type != "track":
178 return None
179 return ItemMapping(
180 media_type=MediaType.TRACK,
181 item_id=item.key,
182 provider=INSTANCE_ID,
183 name=item.title,
184 )
185
186 provider._parse = AsyncMock(side_effect=_parse_tracks_only)
187
188 rows = await provider.get_recommendations()
189 await asyncio.gather(*background_tasks)
190
191 assert [row.item_id for row in rows] == [
192 f"{INSTANCE_ID}_{HUB_RECENT}",
193 f"{INSTANCE_ID}_{HUB_NEW}",
194 ]
195
196
197@pytest.mark.asyncio
198async def test_backend_error_propagates() -> None:
199 """
200 A failing hubs fetch raises so the mixin/controller can serve stale data instead.
201
202 Returning [] here would be cached persistently as a valid empty payload; the
203 error must propagate for the stale-while-revalidate fallback to work.
204 """
205 provider, background_tasks = _make_provider([])
206 provider._plex_library.fetchItems = Mock(side_effect=RuntimeError("plex down"))
207
208 with pytest.raises(RuntimeError, match="plex down"):
209 await provider.get_recommendations()
210 await asyncio.gather(*background_tasks, return_exceptions=True)
211