/
/
/
1"""Unit tests for Plex provider audiobook and podcast methods."""
2
3from __future__ import annotations
4
5from datetime import UTC, datetime
6from typing import Any
7from unittest import mock
8from unittest.mock import AsyncMock, MagicMock
9
10import plexapi.exceptions
11import pytest
12from music_assistant_models.enums import MediaType, ProviderFeature
13from music_assistant_models.errors import MediaNotFoundError
14
15from music_assistant.providers.plex import PlexProvider
16from music_assistant.providers.plex.helpers import SUPPORTED_FEATURES
17
18LIBRARY_TYPE_AUDIOBOOKS = "audiobooks"
19LIBRARY_TYPE_MUSIC = "music"
20LIBRARY_TYPE_PODCASTS = "podcasts"
21
22
23# ---------------------------------------------------------------------------
24# Helpers / fixtures
25# ---------------------------------------------------------------------------
26
27
28def _make_provider(library_type: str = LIBRARY_TYPE_MUSIC) -> Any:
29 """Create a minimal PlexProvider instance for testing."""
30 mock_mass = MagicMock()
31 mock_mass.cache = MagicMock()
32
33 mock_config = MagicMock()
34 mock_config.instance_id = "plex_instance_1"
35
36 mock_config_values: dict[str, Any] = {
37 "library_type": library_type,
38 "log_level": "INFO",
39 "token": "local_auth",
40 }
41
42 class MockValue:
43 """Simple wrapper for mock config values."""
44
45 def __init__(self, val: Any) -> None:
46 self.value = val
47
48 mock_config.values = {k: MockValue(v) for k, v in mock_config_values.items()}
49 mock_config.get_value = lambda key: mock_config_values.get(key)
50
51 # the auth token + library type are read via get_setup_value (setup_data store)
52 setup_data = {"library_type": library_type, "token": "local_auth"}
53 mock_mass.config.get = lambda key, default=None: (
54 setup_data if str(key).endswith("/setup_data") else default
55 )
56 mock_mass.config.get_raw_provider_config_value = lambda _instance_id, _key: None
57 mock_mass.config.decrypt_string = lambda value: value
58
59 mock_manifest = MagicMock()
60 mock_manifest.type = "music"
61 mock_manifest.domain = "plex"
62
63 provider = PlexProvider(mock_mass, mock_manifest, mock_config, SUPPORTED_FEATURES)
64 provider._baseurl = "http://localhost:32400"
65 provider._plex_server = MagicMock()
66 provider._plex_library = MagicMock()
67 provider._myplex_account = MagicMock()
68
69 return provider
70
71
72class FakePlexTrack:
73 """Minimal PlexTrack stub for testing."""
74
75 def __init__( # noqa: D107
76 self,
77 key: str = "/library/metadata/1",
78 title: str = "Track 1",
79 duration: int = 1000,
80 has_media: bool = True,
81 has_parts: bool = True,
82 parent_index: int | None = 1,
83 track_number: int | None = 1,
84 container: str | None = "mp3",
85 view_offset: int = 0,
86 parent_key: str = "/library/metadata/100",
87 ) -> None:
88 self.key = key
89 self.title = title
90 self.duration = duration
91 self.parentIndex = parent_index
92 self.trackNumber = track_number
93 self.parentKey = parent_key
94 self.viewOffset = view_offset
95 self.viewCount = 0
96 self.summary = ""
97 if has_media:
98 media = MagicMock()
99 media.container = container
100 media.parts = [MagicMock()] if has_parts else []
101 self.media = [media]
102 else:
103 self.media = []
104
105 def getWebURL(self, baseurl: str) -> str: # noqa: N802, D102
106 return f"{baseurl}/web/index.html#!/server/item/{self.key}"
107
108 def firstAttr(self, *attrs: str) -> str | None: # noqa: N802, D102
109 return None
110
111 def updateTimeline(self, offset_ms: int, *, state: str, duration: int | None = None) -> None: # noqa: N802, D102
112 pass
113
114
115class FakePlexAlbum:
116 """Minimal PlexAlbum stub for testing."""
117
118 def __init__( # noqa: D107
119 self,
120 tracks: list[Any],
121 title: str = "Test Album",
122 key: str = "/library/metadata/100",
123 view_count: int = 0,
124 last_viewed_at: datetime | None = None,
125 album_duration: int = 0,
126 ) -> None:
127 self._tracks = tracks
128 self.title = title
129 self.key = key
130 self.summary = ""
131 self.year = None
132 self.studio = None
133 self.parentTitle = None
134 self.grandparentTitle = None
135 self.viewCount = view_count
136 self.lastViewedAt = last_viewed_at
137 self.duration = album_duration
138
139 def tracks(self) -> list[Any]: # noqa: D102
140 return self._tracks
141
142 def getWebURL(self, baseurl: str) -> str: # noqa: N802, D102
143 return f"{baseurl}/web/index.html#!/server/item/{self.key}"
144
145 def firstAttr(self, *attrs: str) -> str | None: # noqa: N802, D102
146 return None
147
148 def markPlayed(self) -> None: # noqa: D102, N802
149 pass
150
151 def markUnplayed(self) -> None: # noqa: D102, N802
152 pass
153
154 def reload(self) -> None: # noqa: D102
155 pass
156
157
158@pytest.fixture
159def audiobook_provider() -> Any:
160 """Provide an audiobook-configured PlexProvider."""
161 return _make_provider(LIBRARY_TYPE_AUDIOBOOKS)
162
163
164@pytest.fixture
165def podcast_provider() -> Any:
166 """Provide a podcast-configured PlexProvider."""
167 return _make_provider(LIBRARY_TYPE_PODCASTS)
168
169
170@pytest.fixture
171def music_provider() -> Any:
172 """Provide a music-configured PlexProvider."""
173 return _make_provider(LIBRARY_TYPE_MUSIC)
174
175
176def _make_tracks(*specs: dict[str, Any]) -> list[FakePlexTrack]:
177 """Build a list of FakePlexTrack from dict specs."""
178 return [FakePlexTrack(**s) for s in specs]
179
180
181def _make_album(tracks: list[Any], **kwargs: Any) -> FakePlexAlbum:
182 """Build a FakePlexAlbum with the given tracks."""
183 return FakePlexAlbum(tracks, **kwargs)
184
185
186# ---------------------------------------------------------------------------
187# Audiobook chapters
188# ---------------------------------------------------------------------------
189
190
191class TestBuildAudiobookChapters:
192 """Tests for _build_audiobook_chapters numbering behavior."""
193
194 @pytest.mark.asyncio
195 async def test_chapters_numbered_sequentially(self, audiobook_provider: Any) -> None:
196 """Chapters should have sequential positions starting from 1."""
197 tracks = _make_tracks(
198 {"key": "/1", "title": "Intro", "duration": 3000},
199 {"key": "/2", "title": "Chapter 1", "duration": 5000},
200 {"key": "/3", "title": "Outro", "duration": 2000},
201 )
202 chapters = await audiobook_provider._build_audiobook_chapters(_make_album(tracks))
203
204 assert len(chapters) == 3
205 assert chapters[0].position == 1
206 assert chapters[1].position == 2
207 assert chapters[2].position == 3
208 assert chapters[0].name == "Intro"
209 assert chapters[1].name == "Chapter 1"
210 assert chapters[2].name == "Outro"
211
212 @pytest.mark.asyncio
213 async def test_chapters_skip_tracks_without_media(self, audiobook_provider: Any) -> None:
214 """Tracks without media should be skipped without creating gaps in numbering."""
215 tracks = _make_tracks(
216 {"key": "/1", "title": "Valid Track 1", "duration": 3000},
217 {"key": "/2", "title": "No Media", "duration": 5000, "has_media": False},
218 {"key": "/3", "title": "Valid Track 2", "duration": 2000},
219 )
220 chapters = await audiobook_provider._build_audiobook_chapters(_make_album(tracks))
221
222 assert len(chapters) == 2
223 assert chapters[0].position == 1
224 assert chapters[1].position == 2
225 assert chapters[0].name == "Valid Track 1"
226 assert chapters[1].name == "Valid Track 2"
227
228 @pytest.mark.asyncio
229 async def test_chapters_skip_tracks_without_parts(self, audiobook_provider: Any) -> None:
230 """Tracks with media but no parts should be skipped without gaps."""
231 tracks = _make_tracks(
232 {"key": "/1", "title": "Valid", "duration": 3000},
233 {"key": "/2", "title": "No Parts", "duration": 5000, "has_parts": False},
234 {"key": "/3", "title": "Another Valid", "duration": 2000},
235 )
236 chapters = await audiobook_provider._build_audiobook_chapters(_make_album(tracks))
237
238 assert len(chapters) == 2
239 assert chapters[0].position == 1
240 assert chapters[1].position == 2
241
242 @pytest.mark.asyncio
243 async def test_chapters_cumulative_times(self, audiobook_provider: Any) -> None:
244 """Chapter start/end times should be cumulative across valid tracks."""
245 tracks = _make_tracks(
246 {"key": "/1", "title": "First", "duration": 1000},
247 {"key": "/2", "title": "Second", "duration": 2000},
248 {"key": "/3", "title": "Third", "duration": 3000},
249 )
250 chapters = await audiobook_provider._build_audiobook_chapters(_make_album(tracks))
251
252 assert chapters[0].start == 0.0
253 assert chapters[0].end == 1.0
254 assert chapters[1].start == 1.0
255 assert chapters[1].end == 3.0
256 assert chapters[2].start == 3.0
257 assert chapters[2].end == 6.0
258
259
260class TestBuildPodcastEpisodes:
261 """Tests for _build_podcast_episodes numbering behavior."""
262
263 @pytest.mark.asyncio
264 async def test_episodes_numbered_sequentially(self, podcast_provider: Any) -> None:
265 """Episodes should have sequential positions starting from 1."""
266 tracks = _make_tracks(
267 {"key": "/1", "title": "Intro"},
268 {"key": "/2", "title": "Main Episode"},
269 {"key": "/3", "title": "Outro"},
270 )
271 episodes = await podcast_provider._build_podcast_episodes(
272 _make_album(tracks, title="Test Podcast")
273 )
274
275 assert len(episodes) == 3
276 assert episodes[0].position == 1
277 assert episodes[1].position == 2
278 assert episodes[2].position == 3
279 assert episodes[0].name == "Intro"
280 assert episodes[1].name == "Main Episode"
281 assert episodes[2].name == "Outro"
282
283 @pytest.mark.asyncio
284 async def test_episodes_skip_tracks_without_media(self, podcast_provider: Any) -> None:
285 """Tracks without media should be skipped without creating gaps."""
286 tracks = _make_tracks(
287 {"key": "/1", "title": "Valid Ep 1"},
288 {"key": "/2", "title": "No Media", "has_media": False},
289 {"key": "/3", "title": "Valid Ep 2"},
290 )
291 episodes = await podcast_provider._build_podcast_episodes(
292 _make_album(tracks, title="Test Podcast")
293 )
294
295 assert len(episodes) == 2
296 assert episodes[0].position == 1
297 assert episodes[1].position == 2
298 assert episodes[0].name == "Valid Ep 1"
299 assert episodes[1].name == "Valid Ep 2"
300
301 @pytest.mark.asyncio
302 async def test_episode_default_name(self, podcast_provider: Any) -> None:
303 """Tracks without titles should use default episode name with correct number."""
304 tracks = _make_tracks(
305 {"key": "/1", "title": ""},
306 {"key": "/2", "title": "Has Title"},
307 {"key": "/3", "title": ""},
308 )
309 episodes = await podcast_provider._build_podcast_episodes(
310 _make_album(tracks, title="Test Podcast")
311 )
312
313 assert episodes[0].name == "Episode 1"
314 assert episodes[1].name == "Has Title"
315 assert episodes[2].name == "Episode 3"
316
317 @pytest.mark.asyncio
318 async def test_episode_podcast_reference(self, podcast_provider: Any) -> None:
319 """Each episode should reference the parent podcast correctly."""
320 tracks = _make_tracks({"key": "/1", "title": "Ep 1"})
321 episodes = await podcast_provider._build_podcast_episodes(
322 _make_album(tracks, title="My Podcast", key="/library/metadata/100")
323 )
324
325 assert len(episodes) == 1
326 assert episodes[0].podcast.name == "My Podcast"
327 assert episodes[0].podcast.item_id == "podcast:/library/metadata/100"
328
329
330class TestStreamDetailsGuards:
331 """Tests for library type guards in stream detail methods."""
332
333 @pytest.mark.asyncio
334 async def test_audiobook_stream_rejects_music_library(self, music_provider: Any) -> None:
335 """_get_audiobook_stream_details should raise when library type is music."""
336 with pytest.raises(MediaNotFoundError, match="not configured for audiobooks"):
337 await music_provider._get_audiobook_stream_details("audiobook:/library/metadata/1")
338
339 @pytest.mark.asyncio
340 async def test_audiobook_stream_rejects_podcast_library(self, podcast_provider: Any) -> None:
341 """_get_audiobook_stream_details should raise when library type is podcasts."""
342 with pytest.raises(MediaNotFoundError, match="not configured for audiobooks"):
343 await podcast_provider._get_audiobook_stream_details("audiobook:/library/metadata/1")
344
345 @pytest.mark.asyncio
346 async def test_podcast_stream_rejects_music_library(self, music_provider: Any) -> None:
347 """_get_podcast_episode_stream_details should raise when library type is music."""
348 with pytest.raises(MediaNotFoundError, match="not configured for podcasts"):
349 await music_provider._get_podcast_episode_stream_details(
350 "podcast_episode:/library/metadata/1"
351 )
352
353 @pytest.mark.asyncio
354 async def test_podcast_stream_rejects_audiobook_library(self, audiobook_provider: Any) -> None:
355 """_get_podcast_episode_stream_details should raise when library type is audiobooks."""
356 with pytest.raises(MediaNotFoundError, match="not configured for podcasts"):
357 await audiobook_provider._get_podcast_episode_stream_details(
358 "podcast_episode:/library/metadata/1"
359 )
360
361
362class TestSupportedFeaturesProperty:
363 """The supported_features property reflects the configured library type."""
364
365 def test_music_returns_all_music_features(self) -> None:
366 """A music library exposes the full music feature set."""
367 result = _make_provider(LIBRARY_TYPE_MUSIC).supported_features
368
369 assert ProviderFeature.LIBRARY_ARTISTS in result
370 assert ProviderFeature.LIBRARY_ALBUMS in result
371 assert ProviderFeature.LIBRARY_TRACKS in result
372 assert ProviderFeature.LIBRARY_PLAYLISTS in result
373 assert ProviderFeature.LIBRARY_AUDIOBOOKS not in result
374 assert ProviderFeature.LIBRARY_PODCASTS not in result
375
376 def test_audiobooks_returns_only_audiobook_features(self) -> None:
377 """An audiobooks library narrows the feature set to audiobooks."""
378 result = _make_provider(LIBRARY_TYPE_AUDIOBOOKS).supported_features
379
380 assert ProviderFeature.LIBRARY_AUDIOBOOKS in result
381 assert ProviderFeature.BROWSE in result
382 assert ProviderFeature.SEARCH in result
383 assert ProviderFeature.LIBRARY_ARTISTS not in result
384 assert ProviderFeature.LIBRARY_ALBUMS not in result
385 assert ProviderFeature.LIBRARY_TRACKS not in result
386 assert ProviderFeature.LIBRARY_PLAYLISTS not in result
387 assert ProviderFeature.LIBRARY_PODCASTS not in result
388
389 def test_podcasts_returns_only_podcast_features(self) -> None:
390 """A podcasts library narrows the feature set to podcasts."""
391 result = _make_provider(LIBRARY_TYPE_PODCASTS).supported_features
392
393 assert ProviderFeature.LIBRARY_PODCASTS in result
394 assert ProviderFeature.BROWSE in result
395 assert ProviderFeature.SEARCH in result
396 assert ProviderFeature.LIBRARY_ARTISTS not in result
397 assert ProviderFeature.LIBRARY_ALBUMS not in result
398 assert ProviderFeature.LIBRARY_TRACKS not in result
399 assert ProviderFeature.LIBRARY_PLAYLISTS not in result
400 assert ProviderFeature.LIBRARY_AUDIOBOOKS not in result
401
402
403# ---------------------------------------------------------------------------
404# Resume / progress system tests
405# ---------------------------------------------------------------------------
406
407
408class TestCalcResumePosition:
409 """Tests for _calc_resume_position_ms."""
410
411 @pytest.mark.asyncio
412 async def test_no_progress_returns_zero(self, audiobook_provider: Any) -> None:
413 """When no track has a viewOffset, resume position is zero."""
414 tracks = _make_tracks(
415 {"duration": 10000, "view_offset": 0},
416 {"duration": 20000, "view_offset": 0},
417 )
418 result = await audiobook_provider._calc_resume_position_ms(
419 _make_album(tracks), fully_played=False
420 )
421 assert result == 0
422
423 @pytest.mark.asyncio
424 async def test_last_offset_determines_position(self, audiobook_provider: Any) -> None:
425 """The last track with a non-zero viewOffset sets the resume point."""
426 tracks = _make_tracks(
427 {"duration": 10000, "view_offset": 5000},
428 {"duration": 20000, "view_offset": 12000},
429 )
430 result = await audiobook_provider._calc_resume_position_ms(
431 _make_album(tracks), fully_played=False
432 )
433 assert result == 10000 + 12000
434
435 @pytest.mark.asyncio
436 async def test_fully_played_with_no_offset_uses_album_duration(
437 self, audiobook_provider: Any
438 ) -> None:
439 """When fully played and no offsets, return album duration."""
440 tracks = _make_tracks(
441 {"duration": 10000, "view_offset": 0},
442 {"duration": 20000, "view_offset": 0},
443 )
444 result = await audiobook_provider._calc_resume_position_ms(
445 _make_album(tracks, album_duration=30000), fully_played=True
446 )
447 assert result == 30000
448
449 @pytest.mark.asyncio
450 async def test_mixed_offsets_with_gap(self, audiobook_provider: Any) -> None:
451 """Only the last offset matters; gaps are ignored."""
452 tracks = _make_tracks(
453 {"duration": 10000, "view_offset": 8000},
454 {"duration": 20000, "view_offset": 0},
455 {"duration": 15000, "view_offset": 5000},
456 )
457 result = await audiobook_provider._calc_resume_position_ms(
458 _make_album(tracks), fully_played=False
459 )
460 assert result == 30000 + 5000
461
462
463class TestFindTrackForPosition:
464 """Tests for _find_track_for_position."""
465
466 @pytest.mark.asyncio
467 async def test_position_in_first_track(self, audiobook_provider: Any) -> None:
468 """Position inside the first track returns it with correct offset."""
469 tracks = _make_tracks(
470 {"key": "/1", "duration": 10000},
471 {"key": "/2", "duration": 20000},
472 )
473 track, offset = await audiobook_provider._find_track_for_position(
474 _make_album(tracks), position=5
475 )
476 assert track is not None
477 assert track.key == "/1"
478 assert offset == 5000 # 5 seconds = 5000 ms
479
480 @pytest.mark.asyncio
481 async def test_position_in_second_track(self, audiobook_provider: Any) -> None:
482 """Position inside second track returns it with offset relative to track start."""
483 tracks = _make_tracks(
484 {"key": "/1", "duration": 10000},
485 {"key": "/2", "duration": 20000},
486 )
487 track, offset = await audiobook_provider._find_track_for_position(
488 _make_album(tracks), position=12
489 )
490 assert track is not None
491 assert track.key == "/2"
492 assert offset == 2000
493
494 @pytest.mark.asyncio
495 async def test_position_past_end_clamps_to_last_track(self, audiobook_provider: Any) -> None:
496 """Position beyond all tracks clamps to the end of the last track."""
497 tracks = _make_tracks(
498 {"key": "/1", "duration": 10000},
499 {"key": "/2", "duration": 20000},
500 )
501 track, offset = await audiobook_provider._find_track_for_position(
502 _make_album(tracks), position=40
503 )
504 assert track is not None
505 assert track.key == "/2"
506 assert offset == 20000 # full duration of last track
507
508 @pytest.mark.asyncio
509 async def test_empty_album_returns_none(self, audiobook_provider: Any) -> None:
510 """An album with no tracks returns (None, 0)."""
511 track, offset = await audiobook_provider._find_track_for_position(
512 _make_album([]), position=5
513 )
514 assert track is None
515 assert offset == 0
516
517
518class TestOnPlayed:
519 """Tests for on_played progress sync."""
520
521 @staticmethod
522 def _setup_call_log(provider: Any, album: Any) -> list[Any]:
523 """Attach run_async logger and fetchItem mock to a provider."""
524 provider._plex_library.fetchItem = MagicMock(return_value=album)
525 call_log: list[Any] = []
526
527 async def _run_async(call: Any, *args: Any, **kwargs: Any) -> Any:
528 call_log.append((call, args, kwargs))
529 return call(*args, **kwargs)
530
531 provider._run_async = _run_async
532 return call_log
533
534 @pytest.mark.asyncio
535 async def test_fully_played_calls_mark_played(self, audiobook_provider: Any) -> None:
536 """When fully_played=True, markPlayed is called on the album."""
537 album = _make_album(
538 _make_tracks({"key": "/1", "duration": 10000}),
539 key="/library/metadata/100",
540 )
541 call_log = self._setup_call_log(audiobook_provider, album)
542
543 await audiobook_provider.on_played(
544 MediaType.AUDIOBOOK,
545 "audiobook:/library/metadata/100",
546 fully_played=True,
547 position=0,
548 media_item=MagicMock(),
549 )
550
551 assert any(call_info[0] == album.markPlayed for call_info in call_log)
552
553 @pytest.mark.asyncio
554 async def test_zero_position_calls_mark_unplayed(self, audiobook_provider: Any) -> None:
555 """When position is 0, markUnplayed is called on the album."""
556 album = _make_album(
557 _make_tracks({"key": "/1", "duration": 10000}),
558 key="/library/metadata/100",
559 )
560 call_log = self._setup_call_log(audiobook_provider, album)
561
562 await audiobook_provider.on_played(
563 MediaType.AUDIOBOOK,
564 "audiobook:/library/metadata/100",
565 fully_played=False,
566 position=0,
567 media_item=MagicMock(),
568 )
569
570 assert any(call_info[0] == album.markUnplayed for call_info in call_log)
571
572 @pytest.mark.asyncio
573 async def test_mid_position_updates_timeline(self, audiobook_provider: Any) -> None:
574 """When position is in the middle, updateTimeline is called on the correct track."""
575 track = FakePlexTrack(key="/1", duration=30000)
576 album = _make_album([track], key="/library/metadata/100")
577 call_log = self._setup_call_log(audiobook_provider, album)
578
579 await audiobook_provider.on_played(
580 MediaType.AUDIOBOOK,
581 "audiobook:/library/metadata/100",
582 fully_played=False,
583 position=10,
584 media_item=MagicMock(),
585 is_playing=True,
586 )
587
588 assert any(
589 call_info[0] == track.updateTimeline and call_info[2].get("state") == "playing"
590 for call_info in call_log
591 )
592
593
594class TestGetResumePosition:
595 """Tests for get_resume_position."""
596
597 @pytest.mark.asyncio
598 async def test_audiobook_with_progress(self, audiobook_provider: Any) -> None:
599 """Returns correct resume position for an audiobook with track offsets."""
600 tracks = _make_tracks(
601 {"duration": 10000, "view_offset": 0},
602 {"duration": 20000, "view_offset": 15000},
603 )
604 viewed_at = datetime(2024, 1, 1, 12, 0, 0, tzinfo=UTC)
605 album = _make_album(tracks, key="/library/metadata/100", last_viewed_at=viewed_at)
606
607 audiobook_provider._plex_library.fetchItem = MagicMock(return_value=album)
608
609 fully_played, position_ms, timestamp = await audiobook_provider.get_resume_position(
610 "audiobook:/library/metadata/100", MediaType.AUDIOBOOK
611 )
612
613 assert fully_played is False
614 assert position_ms == 10000 + 15000
615 assert timestamp == viewed_at
616
617 @pytest.mark.asyncio
618 async def test_fully_played_audiobook(self, audiobook_provider: Any) -> None:
619 """When fully played, returns album duration as resume position."""
620 tracks = _make_tracks({"duration": 30000, "view_offset": 0})
621 album = _make_album(tracks, key="/library/metadata/100", view_count=1, album_duration=30000)
622
623 audiobook_provider._plex_library.fetchItem = MagicMock(return_value=album)
624
625 fully_played, position_ms, _ = await audiobook_provider.get_resume_position(
626 "audiobook:/library/metadata/100", MediaType.AUDIOBOOK
627 )
628
629 assert fully_played is True
630 assert position_ms == 30000
631
632 @pytest.mark.asyncio
633 async def test_podcast_episode_reads_own_view_offset(self, podcast_provider: Any) -> None:
634 """Podcast episode reads its own track-level viewOffset for resume position."""
635 episode_track = FakePlexTrack(
636 key="/library/metadata/200",
637 duration=15000,
638 view_offset=12000,
639 parent_key="/library/metadata/100",
640 )
641
642 podcast_provider._plex_library.fetchItem = MagicMock(return_value=episode_track)
643
644 fully_played, position_ms, _ = await podcast_provider.get_resume_position(
645 "podcast_episode:/library/metadata/200", MediaType.PODCAST_EPISODE
646 )
647
648 assert fully_played is False
649 assert position_ms == 12000
650 # Verify we did NOT fetch the parent album
651 assert podcast_provider._plex_library.fetchItem.call_count == 1
652
653 @pytest.mark.asyncio
654 async def test_podcast_episode_reads_own_fully_played(self, podcast_provider: Any) -> None:
655 """Podcast episode reads its own viewCount to determine fully_played."""
656 episode_track = FakePlexTrack(
657 key="/library/metadata/200",
658 duration=15000,
659 view_offset=0,
660 parent_key="/library/metadata/100",
661 )
662
663 podcast_provider._plex_library.fetchItem = MagicMock(return_value=episode_track)
664
665 # Simulate viewCount by overriding the viewCount attribute
666 episode_track.viewCount = 1
667
668 fully_played, position_ms, _ = await podcast_provider.get_resume_position(
669 "podcast_episode:/library/metadata/200", MediaType.PODCAST_EPISODE
670 )
671
672 assert fully_played is True
673 assert position_ms == 0
674
675 @pytest.mark.asyncio
676 async def test_not_found_raises_media_not_found(self, audiobook_provider: Any) -> None:
677 """If the Plex item is missing, MediaNotFoundError is raised."""
678 audiobook_provider._run_async = AsyncMock(
679 side_effect=plexapi.exceptions.NotFound("Item not found")
680 )
681
682 with pytest.raises(MediaNotFoundError):
683 await audiobook_provider.get_resume_position(
684 "audiobook:/library/metadata/999", MediaType.AUDIOBOOK
685 )
686
687
688# ---------------------------------------------------------------------------
689# stale library-mapping cleanup (idempotent, runs on load)
690# ---------------------------------------------------------------------------
691
692
693class TestCleanupStaleLibraryMappings:
694 """Tests for _cleanup_stale_library_mappings removing entries not matching the type."""
695
696 @staticmethod
697 def _setup_mocks(provider: Any, query_result: list[dict[str, Any]] | None = None) -> Any:
698 """Wire standard mocks onto provider.mass for the cleanup tests."""
699 provider.mass.music.database = MagicMock()
700 provider.mass.music.database.get_rows_from_query = AsyncMock(
701 return_value=query_result or []
702 )
703 provider.mass.music.get_controller = MagicMock()
704 mock_ctrl = MagicMock()
705 mock_ctrl.remove_provider_mappings = AsyncMock()
706 provider.mass.music.get_controller.return_value = mock_ctrl
707 return mock_ctrl
708
709 @pytest.mark.asyncio
710 async def test_music_library_cleans_non_music_types(self) -> None:
711 """A music library removes mappings for the 3 non-music media types."""
712 provider = _make_provider(LIBRARY_TYPE_MUSIC)
713 mock_ctrl = self._setup_mocks(provider, query_result=[{"item_id": 10}])
714
715 await provider._cleanup_stale_library_mappings()
716
717 # stale for music = audiobook, podcast, podcast_episode
718 assert provider.mass.music.get_controller.call_count == 3
719 mock_ctrl.remove_provider_mappings.assert_has_awaits(
720 [mock.call(10, "plex_instance_1")] * 3, any_order=True
721 )
722
723 @pytest.mark.asyncio
724 async def test_audiobooks_library_cleans_all_other_types(self) -> None:
725 """An audiobooks library removes mappings for the 6 non-audiobook media types."""
726 provider = _make_provider(LIBRARY_TYPE_AUDIOBOOKS)
727 mock_ctrl = self._setup_mocks(provider, query_result=[{"item_id": 1}])
728
729 await provider._cleanup_stale_library_mappings()
730
731 # stale for audiobooks = artist, album, track, playlist, podcast, podcast_episode
732 assert provider.mass.music.get_controller.call_count == 6
733 assert mock_ctrl.remove_provider_mappings.await_count == 6
734
735 @pytest.mark.asyncio
736 async def test_no_stale_rows_removes_nothing(self) -> None:
737 """With no stale mappings present, no removals are performed."""
738 provider = _make_provider(LIBRARY_TYPE_MUSIC)
739 mock_ctrl = self._setup_mocks(provider, query_result=[])
740
741 await provider._cleanup_stale_library_mappings()
742
743 mock_ctrl.remove_provider_mappings.assert_not_awaited()
744
745 @pytest.mark.asyncio
746 async def test_missing_database_returns_early(self) -> None:
747 """Cleanup is a no-op when the music database is not available."""
748 provider = _make_provider(LIBRARY_TYPE_MUSIC)
749 provider.mass.music.database = None
750 provider.mass.music.get_controller = MagicMock()
751
752 await provider._cleanup_stale_library_mappings()
753
754 provider.mass.music.get_controller.assert_not_called()
755