music-assistant-server

2.1 KBPY
test_provider_load_retry.py
2.1 KB64 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