music-assistant-server
2.1 KB•PY
test_provider_load_retry.py
2.1 KB • 64 lines • python
1"""Tests for provider load retry decisions."""
2
3from __future__ import annotations
4
5from typing import cast
6from unittest.mock import AsyncMock, MagicMock
7
8import pytest
9from music_assistant_models.errors import (
10 AuthenticationFailed,
11 AuthenticationRequired,
12 InvalidToken,
13 LoginFailed,
14 MusicAssistantError,
15 ProviderUnavailableError,
16)
17
18from music_assistant.mass import MusicAssistant
19
20
21def _mass_with_load_error(error: MusicAssistantError) -> MusicAssistant:
22 """Return a minimal MusicAssistant whose provider load raises the given error."""
23 mass = object.__new__(MusicAssistant)
24 provider_config = MagicMock(enabled=True, instance_id="test--instance", name="Test")
25 mass.config = MagicMock()
26 mass.config.get_provider_config = AsyncMock(return_value=provider_config)
27 mass.load_provider_config = AsyncMock(side_effect=error) # type: ignore[method-assign]
28 mass.call_later = MagicMock() # type: ignore[method-assign]
29 mass._tracked_timers = {}
30 return mass
31
32
33@pytest.mark.parametrize(
34 "error",
35 [
36 AuthenticationRequired("Authentication required"),
37 AuthenticationFailed("Authentication failed"),
38 LoginFailed("Login failed"),
39 InvalidToken("Invalid token"),
40 ],
41)
42async def test_authentication_load_errors_are_not_retried(error: MusicAssistantError) -> None:
43 """Provider authentication failures wait for reconfiguration instead of retrying."""
44 mass = _mass_with_load_error(error)
45
46 await mass.load_provider("test--instance", allow_retry=True)
47
48 cast("MagicMock", mass.call_later).assert_not_called()
49
50
51async def test_transient_handled_load_error_is_retried() -> None:
52 """A transient handled provider failure still schedules a delayed retry."""
53 mass = _mass_with_load_error(ProviderUnavailableError("Temporarily unavailable"))
54
55 await mass.load_provider("test--instance", allow_retry=True)
56
57 cast("MagicMock", mass.call_later).assert_called_once_with(
58 120,
59 mass.load_provider,
60 "test--instance",
61 True,
62 task_id="load_provider_test--instance",
63 )
64