/
/
/
1"""Tests for provider status derivation and structured error building."""
2
3from __future__ import annotations
4
5import asyncio
6from unittest.mock import patch
7
8import pytest
9from music_assistant_models.config_entries import ProviderConfig, ProviderError
10from music_assistant_models.enums import ProviderStatus, ProviderType
11from music_assistant_models.errors import (
12 AuthenticationRequired,
13 LoginFailed,
14 SetupFailedError,
15 UnsupportedSystemError,
16)
17
18from music_assistant.controllers.config.helpers import _provider_status
19from music_assistant.mass import (
20 _provider_error_from_exc,
21 _provider_error_traceback,
22 _provider_load_step,
23)
24
25
26def _conf(*, enabled: bool = True, last_error: ProviderError | None = None) -> ProviderConfig:
27 return ProviderConfig(
28 values={},
29 type=ProviderType.MUSIC,
30 domain="demo",
31 instance_id="demo--1",
32 enabled=enabled,
33 last_error=last_error,
34 )
35
36
37def test_provider_status_derivation() -> None:
38 """Status reflects the provider's config + load state, keyed off the error code."""
39 assert _provider_status(_conf(enabled=False), is_loaded=False) == ProviderStatus.DISABLED
40 assert _provider_status(_conf(), is_loaded=True) == ProviderStatus.LOADED
41 assert _provider_status(_conf(), is_loaded=False) == ProviderStatus.LOADING
42 err = ProviderError(error_code=SetupFailedError.error_code, message="boom")
43 assert _provider_status(_conf(last_error=err), is_loaded=False) == ProviderStatus.ERROR
44 auth = ProviderError(error_code=AuthenticationRequired.error_code, message="auth")
45 assert _provider_status(_conf(last_error=auth), is_loaded=False) == ProviderStatus.AUTH_REQUIRED
46 login = ProviderError(error_code=LoginFailed.error_code, message="login")
47 assert (
48 _provider_status(_conf(last_error=login), is_loaded=False) == ProviderStatus.AUTH_REQUIRED
49 )
50 incompat = ProviderError(error_code=UnsupportedSystemError.error_code, message="nope")
51 assert (
52 _provider_status(_conf(last_error=incompat), is_loaded=False) == ProviderStatus.INCOMPATIBLE
53 )
54
55
56def test_recorded_error_is_reported_for_a_loaded_provider() -> None:
57 """A loaded provider carrying an error still reports it, so the UI can flag it."""
58 err = ProviderError(error_code=SetupFailedError.error_code, message="boom")
59 assert _provider_status(_conf(last_error=err), is_loaded=True) == ProviderStatus.ERROR
60 auth = ProviderError(error_code=AuthenticationRequired.error_code, message="auth")
61 assert _provider_status(_conf(last_error=auth), is_loaded=True) == ProviderStatus.AUTH_REQUIRED
62
63
64def test_provider_error_from_exc() -> None:
65 """A MusicAssistantError keeps its code/translation_key; other exceptions get code 999."""
66 err = _provider_error_from_exc(LoginFailed("bad creds"))
67 assert err.error_code == LoginFailed.error_code
68 assert err.message == "bad creds"
69 assert err.translation_key == LoginFailed.translation_key
70 generic = _provider_error_from_exc(ValueError("oops"))
71 assert generic.error_code == 999
72 assert generic.message == "oops"
73 assert generic.translation_key is None
74
75
76def test_traceback_logged_for_unexpected_errors_only() -> None:
77 """Without verbose logging, only unexpected errors are worth a traceback."""
78 with patch("music_assistant.mass.LOGGER.isEnabledFor", return_value=False):
79 assert _provider_error_traceback(LoginFailed("bad creds")) is None
80 unexpected = ValueError("oops")
81 assert _provider_error_traceback(unexpected) is unexpected
82 wrapped = SetupFailedError("timed out while trying to load")
83 wrapped.__cause__ = TimeoutError()
84 assert _provider_error_traceback(wrapped) is wrapped
85
86
87async def test_load_step_names_the_step_that_timed_out() -> None:
88 """A step that overruns its bound fails with a message naming step and bound."""
89 with pytest.raises(SetupFailedError, match="did not initialize within 0 seconds"):
90 async with _provider_load_step("demo", "initialize", 0):
91 await asyncio.sleep(5)
92
93
94async def test_load_step_separates_an_inner_timeout_from_its_own_bound() -> None:
95 """A bare TimeoutError from the provider's own code is not blamed on the bound."""
96 with pytest.raises(SetupFailedError, match="timed out while trying to load"):
97 async with _provider_load_step("demo", "load", 30):
98 raise TimeoutError
99
100
101async def test_load_step_names_the_step_for_a_message_less_error() -> None:
102 """An exception without a message is not surfaced as just its class name."""
103 with pytest.raises(SetupFailedError, match="failed to load: KeyError"):
104 async with _provider_load_step("demo", "load", 30):
105 raise KeyError
106
107
108async def test_load_step_leaves_informative_errors_alone() -> None:
109 """Errors that carry a message of their own are passed through untouched."""
110 with pytest.raises(LoginFailed, match="bad creds"):
111 async with _provider_load_step("demo", "load", 30):
112 raise LoginFailed("bad creds")
113 with pytest.raises(ValueError, match="oops"):
114 async with _provider_load_step("demo", "load", 30):
115 raise ValueError("oops")
116