/
/
1"""Tests that a single unusable Plex item does not abort a library sync."""
2
3from __future__ import annotations
4
5from typing import Any
6from unittest.mock import AsyncMock, MagicMock
7
8import plexapi.exceptions
9import pytest
10from music_assistant_models.errors import InvalidDataError
11from music_assistant_models.media_items import Track
12
13from music_assistant.providers.plex import PlexProvider
14from music_assistant.providers.plex.helpers import SUPPORTED_FEATURES
15
16LIBRARY_TYPE_AUDIOBOOKS = "audiobooks"
17LIBRARY_TYPE_MUSIC = "music"
18LIBRARY_TYPE_PODCASTS = "podcasts"
19
20# both listings guard the same destructive failure, so they are covered the same way
21SPOKEN_LIBRARIES = [
22 (LIBRARY_TYPE_AUDIOBOOKS, "get_library_audiobooks", "_parse_audiobook"),
23 (LIBRARY_TYPE_PODCASTS, "get_library_podcasts", "_parse_podcast"),
24]
25SPOKEN_LIBRARY_LISTINGS = [(library, generator) for library, generator, _ in SPOKEN_LIBRARIES]
26SPOKEN_LIBRARY_PARSERS = [(library, parser) for library, _, parser in SPOKEN_LIBRARIES]
27
28
29class _FakePlexData:
30 """Stub for the cached xml payload of a plex object."""
31
32 def __init__(self, attrib: dict[str, str]) -> None:
33 self.attrib = attrib
34
35 def findall(self, _tag: str) -> list[Any]:
36 return []
37
38
39class _FakePlexTrack:
40 """
41 Minimal PlexTrack stub carrying what the track parser reads.
42
43 Leaving both original_title and grandparent_key empty reproduces a track that Plex
44 holds no artist for, which the parser rejects with an InvalidDataError.
45 """
46
47 def __init__(
48 self,
49 key: str = "/library/metadata/1",
50 title: str = "Track 1",
51 original_title: str | None = None,
52 grandparent_key: str | None = "/library/metadata/10",
53 ) -> None:
54 self.key = key
55 self.title = title
56 self.originalTitle = original_title
57 self.grandparentKey = grandparent_key
58 self.grandparentTitle = "Artist 1"
59 self.parentKey = "/library/metadata/100"
60 self.parentTitle = "Album 1"
61 self.parentIndex = 1
62 self.trackNumber = 1
63 self.duration = 180000
64 self.genres: list[Any] = []
65 self.moods: list[Any] = []
66 media = MagicMock()
67 media.container = "mp3"
68 self.media = [media]
69 self._data = _FakePlexData({"title": title, "key": key})
70
71 def getWebURL(self, baseurl: str) -> str: # noqa: N802
72 return f"{baseurl}/web/index.html#!/server/item/{self.key}"
73
74 def firstAttr(self, *attrs: str) -> str | None: # noqa: N802
75 return None
76
77
78class _FakePlexAlbum:
79 """Minimal PlexAlbum stub for the audiobook library."""
80
81 def __init__(
82 self,
83 key: str = "/library/metadata/500",
84 title: str = "Audiobook 1",
85 year: int | None = None,
86 ) -> None:
87 self.key = key
88 self.title = title
89 self.year = year
90 self.studio = None
91 self.parentTitle = "Author 1"
92 self.grandparentTitle = None
93 self.summary = ""
94 self._data = _FakePlexData({"title": title, "key": key})
95
96 def getWebURL(self, baseurl: str) -> str: # noqa: N802
97 return f"{baseurl}/web/index.html#!/server/item/{self.key}"
98
99 def firstAttr(self, *attrs: str) -> str | None: # noqa: N802
100 return None
101
102
103def _make_provider(library_type: str = LIBRARY_TYPE_MUSIC) -> Any:
104 """Create a minimal PlexProvider instance for testing."""
105 mock_mass = MagicMock()
106 mock_config = MagicMock()
107 mock_config.instance_id = "plex_instance_1"
108 config_values: dict[str, Any] = {"library_type": library_type, "log_level": "INFO"}
109 mock_config.get_value = lambda key: config_values.get(key)
110
111 setup_data = {"library_type": library_type, "token": "local_auth"}
112 mock_mass.config.get = lambda key, default=None: (
113 setup_data if str(key).endswith("/setup_data") else default
114 )
115 mock_mass.config.get_raw_provider_config_value = lambda _instance_id, _key: None
116 mock_mass.config.decrypt_string = lambda value: value
117
118 mock_manifest = MagicMock()
119 mock_manifest.type = "music"
120 mock_manifest.domain = "plex"
121
122 provider = PlexProvider(mock_mass, mock_manifest, mock_config, SUPPORTED_FEATURES)
123 provider._baseurl = "http://localhost:32400"
124 provider._plex_server = MagicMock()
125 provider._plex_library = MagicMock()
126 return provider
127
128
129async def test_library_tracks_skips_track_without_artist() -> None:
130 """A track that Plex holds no artist for is skipped, the rest still syncs."""
131 provider = _make_provider()
132 batches = [[_FakePlexTrack(key="/1", grandparent_key=None), _FakePlexTrack(key="/2")], []]
133 provider._plex_library.searchTracks = MagicMock(side_effect=batches)
134
135 tracks = [track async for track in provider.get_library_tracks()]
136
137 assert [track.item_id for track in tracks] == ["/2"]
138
139
140async def test_library_artists_skips_item_without_valid_id() -> None:
141 """An artist Plex holds no usable id for is skipped, the rest still syncs."""
142 provider = _make_provider()
143 good_artist = MagicMock()
144 provider._plex_library.all = MagicMock(return_value=[MagicMock(), good_artist])
145
146 async def _parse_artist(plex_artist: Any) -> Any:
147 if plex_artist is good_artist:
148 return "parsed"
149 raise InvalidDataError("Artist does not have a valid ID")
150
151 provider._parse_artist = _parse_artist
152
153 artists = [artist async for artist in provider.get_library_artists()]
154
155 assert artists == ["parsed"]
156
157
158@pytest.mark.parametrize(
159 "error",
160 [
161 plexapi.exceptions.Unauthorized("invalid token"),
162 plexapi.exceptions.NotFound("gone"),
163 ConnectionError("server gone"),
164 ],
165)
166async def test_library_tracks_does_not_swallow_server_errors(error: Exception) -> None:
167 """
168 An error that is not about the item itself aborts the sync instead of skipping a track.
169
170 Skipping would drop the track from this run, and the sync deletion pass then removes
171 it from the library even though it is still on the server.
172 """
173 provider = _make_provider()
174 # a terminating second batch, so widening the caught errors fails the test instead
175 # of looping on the same batch forever
176 provider._plex_library.searchTracks = MagicMock(side_effect=[[_FakePlexTrack()], []])
177
178 async def _parse_track(_plex_track: Any) -> Any:
179 raise error
180
181 provider._parse_track = _parse_track
182
183 with pytest.raises(type(error)):
184 _ = [track async for track in provider.get_library_tracks()]
185
186
187@pytest.mark.parametrize(("library_type", "generator", "parse_method"), SPOKEN_LIBRARIES)
188async def test_spoken_library_skips_unparsable_item(
189 library_type: str, generator: str, parse_method: str
190) -> None:
191 """An audiobook or podcast that cannot be parsed is skipped, the rest still syncs."""
192 provider = _make_provider(library_type)
193 good_album = _FakePlexAlbum(key="/501")
194 provider._plex_library.albums = MagicMock(return_value=[_FakePlexAlbum(), good_album])
195
196 async def _parse(plex_album: Any, **_kwargs: Any) -> Any:
197 if plex_album is good_album:
198 return "parsed"
199 raise InvalidDataError("no title")
200
201 setattr(provider, parse_method, _parse)
202
203 items = [item async for item in getattr(provider, generator)()]
204
205 assert items == ["parsed"]
206
207
208@pytest.mark.parametrize(("library_type", "generator"), SPOKEN_LIBRARY_LISTINGS)
209async def test_spoken_library_does_not_swallow_listing_error(
210 library_type: str, generator: str
211) -> None:
212 """
213 A failure to list the audiobook or podcast library aborts the sync.
214
215 Yielding nothing would look exactly like an emptied library, which makes the sync
216 deletion pass remove every item that is still on the server.
217 """
218 provider = _make_provider(library_type)
219 provider._plex_library.albums = MagicMock(side_effect=ConnectionError("server gone"))
220
221 with pytest.raises(ConnectionError):
222 _ = [item async for item in getattr(provider, generator)()]
223
224
225@pytest.mark.parametrize(("library_type", "parse_method"), SPOKEN_LIBRARY_PARSERS)
226async def test_spoken_item_ignores_out_of_range_year(library_type: str, parse_method: str) -> None:
227 """A year Plex cannot express as a date is ignored instead of failing the item."""
228 provider = _make_provider(library_type)
229
230 parsed = await getattr(provider, parse_method)(_FakePlexAlbum(year=19999))
231
232 assert parsed.metadata.release_date is None
233
234
235async def test_album_tracks_skips_unparsable_track() -> None:
236 """An unusable track no longer costs the whole tracklist of an album."""
237 provider = _make_provider()
238 bad_track, good_track = _FakePlexTrack(key="/1"), _FakePlexTrack(key="/2")
239 plex_album = MagicMock()
240 plex_album.tracks = MagicMock(return_value=[bad_track, good_track])
241 provider._get_data = AsyncMock(return_value=plex_album)
242
243 async def _parse_track(plex_track: Any) -> Any:
244 if plex_track is bad_track:
245 raise InvalidDataError("No artist was found for track")
246 return "parsed"
247
248 provider._parse_track = _parse_track
249
250 get_album_tracks: Any = PlexProvider.get_album_tracks.__wrapped__ # type: ignore[attr-defined]
251
252 assert await get_album_tracks(provider, "/library/metadata/100") == ["parsed"]
253
254
255async def test_playlist_tracks_skips_unparsable_track() -> None:
256 """
257 An unusable track no longer makes the whole playlist unplayable.
258
259 The remaining tracks keep consecutive positions, so the skip does not leave a hole in
260 the queue built from this playlist.
261 """
262 provider = _make_provider()
263 tracks = [_FakePlexTrack(key="/1"), _FakePlexTrack(key="/2"), _FakePlexTrack(key="/3")]
264 plex_playlist = MagicMock()
265 plex_playlist.items = MagicMock(return_value=tracks)
266 provider._get_data = AsyncMock(return_value=plex_playlist)
267
268 async def _parse_track(plex_track: Any) -> Any:
269 if plex_track is tracks[0]:
270 raise InvalidDataError("No artist was found for track")
271 return Track(
272 item_id=plex_track.key,
273 provider=provider.instance_id,
274 name=plex_track.title,
275 provider_mappings=set(),
276 )
277
278 provider._parse_track = _parse_track
279
280 get_playlist_tracks: Any = PlexProvider.get_playlist_tracks.__wrapped__ # type: ignore[attr-defined]
281 result = await get_playlist_tracks(provider, "/playlists/1")
282
283 assert [track.item_id for track in result] == ["/2", "/3"]
284 assert [track.position for track in result] == [1, 2]
285