/
/
/
1"""Tests that a single unusable Plex item does not abort a library sync."""
2
3from __future__ import annotations
4
5from collections.abc import Iterator
6from contextlib import contextmanager
7from typing import Any
8from unittest.mock import AsyncMock, MagicMock
9
10import plexapi.exceptions
11import pytest
12from music_assistant_models.enums import MediaType
13from music_assistant_models.errors import InvalidDataError
14from music_assistant_models.media_items import Track
15
16from music_assistant.models.music_provider import SYNC_RUN_STATE, SyncRunState
17from music_assistant.providers.plex import PlexProvider
18from music_assistant.providers.plex.constants import CONF_IMPORT_COLLECTIONS
19from music_assistant.providers.plex.helpers import SUPPORTED_FEATURES
20
21LIBRARY_TYPE_AUDIOBOOKS = "audiobooks"
22LIBRARY_TYPE_MUSIC = "music"
23LIBRARY_TYPE_PODCASTS = "podcasts"
24
25# both listings guard the same destructive failure, so they are covered the same way
26SPOKEN_LIBRARIES = [
27 (LIBRARY_TYPE_AUDIOBOOKS, "get_library_audiobooks", "_parse_audiobook", MediaType.AUDIOBOOK),
28 (LIBRARY_TYPE_PODCASTS, "get_library_podcasts", "_parse_podcast", MediaType.PODCAST),
29]
30SPOKEN_LIBRARY_LISTINGS = [(library, generator) for library, generator, _, _ in SPOKEN_LIBRARIES]
31SPOKEN_LIBRARY_PARSERS = [(library, parser) for library, _, parser, _ in SPOKEN_LIBRARIES]
32SPOKEN_LIBRARY_ITEMS = [
33 (library, generator, parser) for library, generator, parser, _ in SPOKEN_LIBRARIES
34]
35
36
37class _FakePlexData:
38 """Stub for the cached xml payload of a plex object."""
39
40 def __init__(self, attrib: dict[str, str]) -> None:
41 self.attrib = attrib
42
43 def findall(self, _tag: str) -> list[Any]:
44 return []
45
46
47class _FakePlexTrack:
48 """
49 Minimal PlexTrack stub carrying what the track parser reads.
50
51 Leaving both original_title and grandparent_key empty reproduces a track that Plex
52 holds no artist for, which the parser rejects with an InvalidDataError.
53 """
54
55 def __init__(
56 self,
57 key: str = "/library/metadata/1",
58 title: str = "Track 1",
59 original_title: str | None = None,
60 grandparent_key: str | None = "/library/metadata/10",
61 ) -> None:
62 self.key = key
63 self.title = title
64 self.originalTitle = original_title
65 self.grandparentKey = grandparent_key
66 self.grandparentTitle = "Artist 1"
67 self.parentKey = "/library/metadata/100"
68 self.parentTitle = "Album 1"
69 self.parentIndex = 1
70 self.trackNumber = 1
71 self.duration = 180000
72 self.genres: list[Any] = []
73 self.moods: list[Any] = []
74 media = MagicMock()
75 media.container = "mp3"
76 self.media = [media]
77 self._data = _FakePlexData({"title": title, "key": key})
78
79 def getWebURL(self, baseurl: str) -> str: # noqa: N802
80 return f"{baseurl}/web/index.html#!/server/item/{self.key}"
81
82 def firstAttr(self, *attrs: str) -> str | None: # noqa: N802
83 return None
84
85
86class _FakePlexAlbum:
87 """Minimal PlexAlbum stub for the audiobook library."""
88
89 def __init__(
90 self,
91 key: str = "/library/metadata/500",
92 title: str = "Audiobook 1",
93 year: int | None = None,
94 ) -> None:
95 self.key = key
96 self.title = title
97 self.year = year
98 self.studio = None
99 self.parentTitle = "Author 1"
100 self.grandparentTitle = None
101 self.summary = ""
102 self._data = _FakePlexData({"title": title, "key": key})
103
104 def getWebURL(self, baseurl: str) -> str: # noqa: N802
105 return f"{baseurl}/web/index.html#!/server/item/{self.key}"
106
107 def firstAttr(self, *attrs: str) -> str | None: # noqa: N802
108 return None
109
110
111class _FakePlexCollection:
112 """Minimal PlexCollection stub for the collections-as-playlists listing."""
113
114 def __init__(self, key: str = "/library/collections/5", title: str = "Collection 1") -> None:
115 self.key = key
116 self.title = title
117 # plexapi strips the /children suffix off the raw key, so the payload holds a
118 # different key than the one the item id is built from
119 self._data = _FakePlexData({"title": title, "key": f"{key}/children"})
120
121 def firstAttr(self, *attrs: str) -> str | None: # noqa: N802
122 return None
123
124
125@contextmanager
126def _sync_run() -> Iterator[SyncRunState]:
127 """Run a library listing as part of a sync run, so its skips are recorded."""
128 state = SyncRunState()
129 token = SYNC_RUN_STATE.set(state)
130 try:
131 yield state
132 finally:
133 SYNC_RUN_STATE.reset(token)
134
135
136def _make_provider(library_type: str = LIBRARY_TYPE_MUSIC, import_collections: bool = False) -> Any:
137 """Create a minimal PlexProvider instance for testing."""
138 mock_mass = MagicMock()
139 mock_config = MagicMock()
140 mock_config.instance_id = "plex_instance_1"
141 config_values: dict[str, Any] = {
142 "library_type": library_type,
143 "log_level": "INFO",
144 CONF_IMPORT_COLLECTIONS: import_collections,
145 }
146 mock_config.get_value = lambda key: config_values.get(key)
147
148 setup_data = {"library_type": library_type, "token": "local_auth"}
149 mock_mass.config.get = lambda key, default=None: (
150 setup_data if str(key).endswith("/setup_data") else default
151 )
152 mock_mass.config.get_raw_provider_config_value = lambda _instance_id, _key: None
153 mock_mass.config.decrypt_string = lambda value: value
154
155 mock_manifest = MagicMock()
156 mock_manifest.type = "music"
157 mock_manifest.domain = "plex"
158
159 provider = PlexProvider(mock_mass, mock_manifest, mock_config, SUPPORTED_FEATURES)
160 provider._baseurl = "http://localhost:32400"
161 provider._plex_server = MagicMock()
162 provider._plex_library = MagicMock()
163 return provider
164
165
166async def test_library_tracks_skips_track_without_artist() -> None:
167 """A track that Plex holds no artist for is skipped, the rest still syncs."""
168 provider = _make_provider()
169 batches = [[_FakePlexTrack(key="/1", grandparent_key=None), _FakePlexTrack(key="/2")], []]
170 provider._plex_library.searchTracks = MagicMock(side_effect=batches)
171
172 tracks = [track async for track in provider.get_library_tracks()]
173
174 assert [track.item_id for track in tracks] == ["/2"]
175
176
177async def test_library_artists_skips_item_without_valid_id() -> None:
178 """An artist Plex holds no usable id for is skipped, the rest still syncs."""
179 provider = _make_provider()
180 good_artist = MagicMock()
181 provider._plex_library.all = MagicMock(return_value=[MagicMock(), good_artist])
182
183 async def _parse_artist(plex_artist: Any) -> Any:
184 if plex_artist is good_artist:
185 return "parsed"
186 raise InvalidDataError("Artist does not have a valid ID")
187
188 provider._parse_artist = _parse_artist
189
190 artists = [artist async for artist in provider.get_library_artists()]
191
192 assert artists == ["parsed"]
193
194
195@pytest.mark.parametrize(
196 "error",
197 [
198 plexapi.exceptions.Unauthorized("invalid token"),
199 plexapi.exceptions.NotFound("gone"),
200 ConnectionError("server gone"),
201 ],
202)
203async def test_library_tracks_does_not_swallow_server_errors(error: Exception) -> None:
204 """
205 An error that is not about the item itself aborts the sync instead of skipping a track.
206
207 Skipping would drop the track from this run, and the sync deletion pass then removes
208 it from the library even though it is still on the server.
209 """
210 provider = _make_provider()
211 # a terminating second batch, so widening the caught errors fails the test instead
212 # of looping on the same batch forever
213 provider._plex_library.searchTracks = MagicMock(side_effect=[[_FakePlexTrack()], []])
214
215 async def _parse_track(_plex_track: Any) -> Any:
216 raise error
217
218 provider._parse_track = _parse_track
219
220 with pytest.raises(type(error)):
221 _ = [track async for track in provider.get_library_tracks()]
222
223
224@pytest.mark.parametrize(("library_type", "generator", "parse_method"), SPOKEN_LIBRARY_ITEMS)
225async def test_spoken_library_skips_unparsable_item(
226 library_type: str, generator: str, parse_method: str
227) -> None:
228 """An audiobook or podcast that cannot be parsed is skipped, the rest still syncs."""
229 provider = _make_provider(library_type)
230 good_album = _FakePlexAlbum(key="/501")
231 provider._plex_library.albums = MagicMock(return_value=[_FakePlexAlbum(), good_album])
232
233 async def _parse(plex_album: Any, **_kwargs: Any) -> Any:
234 if plex_album is good_album:
235 return "parsed"
236 raise InvalidDataError("no title")
237
238 setattr(provider, parse_method, _parse)
239
240 items = [item async for item in getattr(provider, generator)()]
241
242 assert items == ["parsed"]
243
244
245@pytest.mark.parametrize(("library_type", "generator"), SPOKEN_LIBRARY_LISTINGS)
246async def test_spoken_library_does_not_swallow_listing_error(
247 library_type: str, generator: str
248) -> None:
249 """
250 A failure to list the audiobook or podcast library aborts the sync.
251
252 Yielding nothing would look exactly like an emptied library, which makes the sync
253 deletion pass remove every item that is still on the server.
254 """
255 provider = _make_provider(library_type)
256 provider._plex_library.albums = MagicMock(side_effect=ConnectionError("server gone"))
257
258 with pytest.raises(ConnectionError):
259 _ = [item async for item in getattr(provider, generator)()]
260
261
262@pytest.mark.parametrize(("library_type", "parse_method"), SPOKEN_LIBRARY_PARSERS)
263async def test_spoken_item_ignores_out_of_range_year(library_type: str, parse_method: str) -> None:
264 """A year Plex cannot express as a date is ignored instead of failing the item."""
265 provider = _make_provider(library_type)
266
267 parsed = await getattr(provider, parse_method)(_FakePlexAlbum(year=19999))
268
269 assert parsed.metadata.release_date is None
270
271
272async def test_album_tracks_skips_unparsable_track() -> None:
273 """An unusable track no longer costs the whole tracklist of an album."""
274 provider = _make_provider()
275 bad_track, good_track = _FakePlexTrack(key="/1"), _FakePlexTrack(key="/2")
276 plex_album = MagicMock()
277 plex_album.tracks = MagicMock(return_value=[bad_track, good_track])
278 provider._get_data = AsyncMock(return_value=plex_album)
279
280 async def _parse_track(plex_track: Any) -> Any:
281 if plex_track is bad_track:
282 raise InvalidDataError("No artist was found for track")
283 return "parsed"
284
285 provider._parse_track = _parse_track
286
287 get_album_tracks: Any = PlexProvider.get_album_tracks.__wrapped__ # type: ignore[attr-defined]
288
289 assert await get_album_tracks(provider, "/library/metadata/100") == ["parsed"]
290
291
292async def test_playlist_tracks_skips_unparsable_track() -> None:
293 """
294 An unusable track no longer makes the whole playlist unplayable.
295
296 The remaining tracks keep consecutive positions, so the skip does not leave a hole in
297 the queue built from this playlist.
298 """
299 provider = _make_provider()
300 tracks = [_FakePlexTrack(key="/1"), _FakePlexTrack(key="/2"), _FakePlexTrack(key="/3")]
301 plex_playlist = MagicMock()
302 plex_playlist.items = MagicMock(return_value=tracks)
303 provider._get_data = AsyncMock(return_value=plex_playlist)
304
305 async def _parse_track(plex_track: Any) -> Any:
306 if plex_track is tracks[0]:
307 raise InvalidDataError("No artist was found for track")
308 return Track(
309 item_id=plex_track.key,
310 provider=provider.instance_id,
311 name=plex_track.title,
312 provider_mappings=set(),
313 )
314
315 provider._parse_track = _parse_track
316
317 get_playlist_tracks: Any = PlexProvider.get_playlist_tracks.__wrapped__ # type: ignore[attr-defined]
318 result = await get_playlist_tracks(provider, "/playlists/1")
319
320 assert [track.item_id for track in result] == ["/2", "/3"]
321 assert [track.position for track in result] == [1, 2]
322
323
324async def test_skipped_track_is_reported_with_its_item_id() -> None:
325 """
326 A skipped track is reported under the id its provider mapping carries.
327
328 That id is what the deletion pass resolves the item by, so the track it could not read
329 is left alone instead of being removed from the library.
330 """
331 provider = _make_provider()
332 provider._plex_library.searchTracks = MagicMock(
333 side_effect=[[_FakePlexTrack(key="/1", grandparent_key=None)], []]
334 )
335
336 with _sync_run() as state:
337 tracks = [track async for track in provider.get_library_tracks()]
338
339 assert tracks == []
340 assert state.skipped_item_ids == {MediaType.TRACK: {"/1"}}
341 assert not state.incomplete_media_types
342
343
344async def test_unidentifiable_artist_holds_back_the_deletion_pass() -> None:
345 """An artist Plex holds no key for cannot be named, so the whole run is held back."""
346 provider = _make_provider()
347 provider._plex_library.all = MagicMock(return_value=[MagicMock(key="")])
348
349 with _sync_run() as state:
350 assert [artist async for artist in provider.get_library_artists()] == []
351
352 assert state.skipped_item_ids == {}
353 assert MediaType.ARTIST in state.incomplete_media_types
354
355
356async def test_skipped_collection_is_reported_as_a_prefixed_playlist() -> None:
357 """
358 A skipped collection is reported under the prefixed id it is stored as a playlist under.
359
360 The bare Plex key belongs to no library item, so reporting that would leave the
361 collection unprotected against the deletion pass.
362 """
363 provider = _make_provider(import_collections=True)
364 collection = _FakePlexCollection()
365 provider._plex_library.playlists = MagicMock(return_value=[])
366 provider._plex_library.collections = MagicMock(return_value=[collection])
367 # take the expectation from the parser itself, so the two can not drift apart
368 parsed = await provider._parse_collection(collection)
369
370 async def _parse_collection(_collection: Any) -> Any:
371 raise InvalidDataError("no title")
372
373 provider._parse_collection = _parse_collection
374
375 with _sync_run() as state:
376 assert [item async for item in provider.get_library_playlists()] == []
377
378 assert state.skipped_item_ids == {MediaType.PLAYLIST: {parsed.item_id}}
379 assert parsed.item_id.startswith("collection:")
380
381
382@pytest.mark.parametrize(
383 ("library_type", "generator", "parse_method", "media_type"), SPOKEN_LIBRARIES
384)
385async def test_spoken_library_reports_skip_with_its_item_id(
386 library_type: str, generator: str, parse_method: str, media_type: MediaType
387) -> None:
388 """An audiobook or podcast is reported under the prefixed id it is stored under."""
389 provider = _make_provider(library_type)
390 album = _FakePlexAlbum()
391 provider._plex_library.albums = MagicMock(return_value=[album])
392 # take the expectation from the parser itself, so the two can not drift apart
393 parsed = await getattr(provider, parse_method)(album)
394
395 async def _parse(_album: Any, **_kwargs: Any) -> Any:
396 raise InvalidDataError("no title")
397
398 setattr(provider, parse_method, _parse)
399
400 with _sync_run() as state:
401 assert [item async for item in getattr(provider, generator)()] == []
402
403 assert state.skipped_item_ids == {media_type: {parsed.item_id}}
404 assert parsed.item_id != album.key
405