music-assistant-server

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