/
/
/
1"""Tests for the Audiobookshelf session sync retry logic."""
2
3from __future__ import annotations
4
5import time
6from unittest.mock import AsyncMock, Mock
7
8from aioaudiobookshelf.exceptions import SessionSyncError as AbsSessionSyncError
9from music_assistant_models.enums import MediaType
10from music_assistant_models.media_items import PodcastEpisode
11
12from music_assistant.providers.audiobookshelf import Audiobookshelf
13from music_assistant.providers.audiobookshelf.helpers import SessionHelper
14
15
16def _make_episode() -> Mock:
17 """Create a minimal podcast episode as returned by the provider."""
18 episode = Mock(spec=PodcastEpisode)
19 episode.duration = 600
20 episode.name = "Test Episode"
21 return episode
22
23
24def _install_session(provider: Audiobookshelf) -> SessionHelper:
25 """Add a session helper to the provider, keyed by mass item id."""
26 # handle_async_init() normally initializes these, but the fixture skips it.
27 # The progress guard would throttle rapid test calls, so bypass it.
28 provider.sessions = {}
29 provider.progress_guard = Mock()
30 provider.progress_guard.guard_ok_mass.return_value = True
31 # the non-session fallback path also updates media progress
32 provider._client.update_my_media_progress = AsyncMock() # type: ignore[method-assign]
33 session = SessionHelper(abs_session_id="abs_session_1", last_sync_time=time.time())
34 provider.sessions["pod1 ep1"] = session
35 return session
36
37
38async def _play(provider: Audiobookshelf) -> None:
39 """Invoke on_played for a podcast episode at position 100."""
40 await provider.on_played(
41 media_type=MediaType.PODCAST_EPISODE,
42 prov_item_id="pod1 ep1",
43 fully_played=False,
44 position=100,
45 media_item=_make_episode(),
46 )
47
48
49async def test_session_kept_below_failure_threshold(
50 provider: Audiobookshelf,
51) -> None:
52 """A session survives consecutive failures below the threshold."""
53 _install_session(provider)
54 provider._client.sync_open_session = AsyncMock( # type: ignore[method-assign]
55 side_effect=AbsSessionSyncError()
56 )
57
58 for _ in range(4):
59 await _play(provider)
60
61 assert "pod1 ep1" in provider.sessions
62 session = provider.sessions["pod1 ep1"]
63 assert session.failed_sync_count == 4
64
65
66async def test_session_removed_at_failure_threshold(
67 provider: Audiobookshelf,
68) -> None:
69 """A session is removed after the 5th consecutive failure."""
70 _install_session(provider)
71 provider._client.sync_open_session = AsyncMock( # type: ignore[method-assign]
72 side_effect=AbsSessionSyncError()
73 )
74
75 for _ in range(5):
76 await _play(provider)
77
78 assert "pod1 ep1" not in provider.sessions
79
80
81async def test_failed_sync_count_reset_on_success(
82 provider: Audiobookshelf,
83) -> None:
84 """A successful sync resets the consecutive failure counter."""
85 session = _install_session(provider)
86 # fail twice, then succeed
87 provider._client.sync_open_session = AsyncMock( # type: ignore[method-assign]
88 side_effect=[AbsSessionSyncError(), AbsSessionSyncError(), Mock()]
89 )
90
91 for _ in range(2):
92 await _play(provider)
93 assert session.failed_sync_count == 2
94
95 await _play(provider)
96 assert session.failed_sync_count == 0
97 assert "pod1 ep1" in provider.sessions
98