/
/
/
1"""
2Regression tests for the provider config save -> (re)load flow.
3
4Covers three fixes for the Spotify auth failures caused by refresh-token rotation:
5- Saving an enabled-but-unloaded provider triggers exactly one (re)load. A second,
6 redundant load reused the same (already rotated and revoked) refresh token, which
7 wiped the credentials right after a successful auth.
8- A failed (re)load records ``last_error`` so the provider surfaces a clear status
9 (e.g. auth_required) instead of appearing stuck loading (a spinning hourglass).
10- A raw provider value stored with ``immediate=True`` is flushed straight away instead of
11 on the debounced timer, so a rotated (and thus revoked) token is not lost on a crash.
12
13Also covers the post-load dependent scan, which must neither resolve the option values of
14every provider nor record its own failures against the provider that just loaded fine.
15"""
16
17from __future__ import annotations
18
19from typing import cast
20from unittest.mock import AsyncMock, MagicMock, patch
21
22import pytest
23from music_assistant_models.config_entries import ConfigEntry, ProviderConfig, ProviderError
24from music_assistant_models.enums import ConfigEntryType, ProviderStatus, ProviderType
25from music_assistant_models.errors import LoginFailed, SetupFailedError
26from music_assistant_models.provider import ProviderManifest
27
28from music_assistant.constants import CONF_LOG_LEVEL, CONF_PROVIDERS
29from music_assistant.controllers.config.helpers import _provider_status
30from music_assistant.mass import MusicAssistant
31
32
33def _prov_conf(instance_id: str) -> ProviderConfig:
34 return ProviderConfig(
35 values={},
36 type=ProviderType.MUSIC,
37 domain="spotify",
38 instance_id=instance_id,
39 enabled=True,
40 )
41
42
43async def test_save_enabled_unloaded_provider_loads_once(mass_minimal: MusicAssistant) -> None:
44 """Saving config for an enabled provider that isn't loaded yet triggers a single load."""
45 config = mass_minimal.config
46 instance = "spotify--test"
47 prov_conf = _prov_conf(instance)
48 with (
49 patch.object(config, "get_provider_config", AsyncMock(return_value=prov_conf)),
50 patch.object(mass_minimal, "load_provider_config", AsyncMock()) as mock_load,
51 ):
52 await config.save_provider_config("spotify", {"enabled": True}, instance_id=instance)
53 mock_load.assert_awaited_once()
54
55
56async def test_failed_reload_records_auth_error(mass_minimal: MusicAssistant) -> None:
57 """A load failure is persisted as last_error so the UI shows auth_required, not a spinner."""
58 config = mass_minimal.config
59 instance = "spotify--test"
60 # the provider config must exist for last_error to be persisted (see #5728)
61 config.set(
62 f"{CONF_PROVIDERS}/{instance}",
63 {"domain": "spotify", "type": "music", "instance_id": instance, "enabled": True},
64 )
65 with (
66 patch.object(
67 mass_minimal, "_load_provider", AsyncMock(side_effect=LoginFailed("bad token"))
68 ),
69 pytest.raises(LoginFailed),
70 ):
71 await mass_minimal.load_provider_config(_prov_conf(instance))
72
73 stored = config.get(f"{CONF_PROVIDERS}/{instance}/last_error")
74 assert stored is not None
75 assert stored["error_code"] == LoginFailed.error_code
76 prov_conf = _prov_conf(instance)
77 prov_conf.last_error = ProviderError.from_dict(stored)
78 assert _provider_status(prov_conf, is_loaded=False) == ProviderStatus.AUTH_REQUIRED
79
80
81async def test_failed_post_load_step_unloads_the_provider(mass_minimal: MusicAssistant) -> None:
82 """A provider that fails to finish loading is unloaded, so it can never read as loaded."""
83 instance = "spotify--test"
84 provider = MagicMock()
85 provider.instance_id = instance
86 provider.domain = "spotify"
87 with (
88 patch.object(
89 mass_minimal,
90 "_update_available_providers_cache",
91 AsyncMock(side_effect=TimeoutError),
92 ),
93 patch.object(mass_minimal, "unload_provider", AsyncMock()) as mock_unload,
94 pytest.raises(SetupFailedError, match="timed out while trying to finish loading"),
95 ):
96 await mass_minimal._register_loaded_provider(provider, _prov_conf(instance))
97
98 mock_unload.assert_awaited_once_with(instance)
99
100
101async def test_save_preserves_unknown_stored_values(mass_minimal: MusicAssistant) -> None:
102 """Stored values without config entries in the current save context survive a save."""
103 config = mass_minimal.config
104 instance = "spotify--test"
105 conf_key = f"{CONF_PROVIDERS}/{instance}"
106 config.set(
107 conf_key,
108 {
109 "domain": "spotify",
110 "type": "music",
111 "instance_id": instance,
112 "enabled": True,
113 "values": {"legacy_token": "keep-me"},
114 "setup_data": {"token": "abc"},
115 },
116 )
117 entry = ConfigEntry(key="region", type=ConfigEntryType.STRING, required=False)
118 prov_conf = cast("ProviderConfig", ProviderConfig.parse([entry], config.get(conf_key)))
119 with (
120 patch.object(config, "get_provider_config", AsyncMock(return_value=prov_conf)),
121 patch.object(mass_minimal, "load_provider_config", AsyncMock()),
122 ):
123 await config._update_provider_config(instance, {"region": "eu"})
124 stored = config.get(conf_key)
125 assert stored["values"]["region"] == "eu"
126 # a value written outside the declared entries (e.g. by a provider at runtime)
127 # is preserved instead of being dropped by the save rebuild
128 assert stored["values"]["legacy_token"] == "keep-me"
129 # setup_data travels along untouched
130 assert stored["setup_data"] == {"token": "abc"}
131
132
133async def test_dependent_scan_does_not_resolve_option_values(
134 mass_minimal: MusicAssistant,
135) -> None:
136 """The post-load dependent scan reads the configs without resolving their option values."""
137 instance = "spotify--test"
138 with (
139 patch.object(mass_minimal, "_load_provider", AsyncMock()),
140 patch.object(
141 mass_minimal.config, "get_provider_configs", AsyncMock(return_value=[])
142 ) as mock_get,
143 ):
144 await mass_minimal.load_provider_config(_prov_conf(instance))
145
146 # resolving values calls get_config_entries() on every loaded provider, which for some
147 # (e.g. Home Assistant) means live network i/o - once per provider load
148 assert mock_get.await_args is not None
149 assert mock_get.await_args.kwargs.get("include_values") is not True
150 assert True not in mock_get.await_args.args
151
152
153async def test_dependent_is_loaded_with_resolved_config_values(
154 mass_minimal: MusicAssistant,
155) -> None:
156 """A dependent is handed a config with its values resolved, not the bare scan result."""
157 config = mass_minimal.config
158 instance = "dependent--test"
159 raw = {"domain": "dependent", "type": "player", "instance_id": instance, "enabled": True}
160 config.set(f"{CONF_PROVIDERS}/{instance}", raw)
161 mass_minimal._provider_manifests["dependent"] = ProviderManifest(
162 type=ProviderType.PLAYER,
163 domain="dependent",
164 name="Dependent",
165 description="",
166 codeowners=[],
167 depends_on="spotify",
168 )
169 # what the (deliberately cheap) dependent scan yields: no entries, so no values
170 scanned = cast("ProviderConfig", ProviderConfig.parse([], raw))
171 assert scanned.get_value(CONF_LOG_LEVEL) is None
172
173 with (
174 patch.object(mass_minimal, "_load_provider", AsyncMock()) as mock_load,
175 patch.object(config, "get_provider_configs", AsyncMock(return_value=[scanned])),
176 ):
177 await mass_minimal.load_provider_config(_prov_conf("spotify--test"))
178
179 loaded = mock_load.await_args_list[-1].args[0]
180 assert loaded.instance_id == instance
181 # a provider reads its log level while being constructed, so an unresolved config
182 # (value None) makes logger.setLevel() raise and the dependent never loads at all
183 assert loaded.get_value(CONF_LOG_LEVEL) == "GLOBAL"
184
185
186async def test_dependent_scan_failure_is_not_blamed_on_loaded_provider(
187 mass_minimal: MusicAssistant,
188) -> None:
189 """A provider that loaded fine keeps a clean status when the dependent scan fails."""
190 config = mass_minimal.config
191 instance = "spotify--test"
192 config.set(
193 f"{CONF_PROVIDERS}/{instance}",
194 {"domain": "spotify", "type": "music", "instance_id": instance, "enabled": True},
195 )
196 with (
197 patch.object(mass_minimal, "_load_provider", AsyncMock()),
198 patch.object(config, "get_provider_configs", AsyncMock(side_effect=TimeoutError)),
199 ):
200 # the scan failure is swallowed: the provider itself loaded successfully
201 await mass_minimal.load_provider_config(_prov_conf(instance))
202
203 assert config.get(f"{CONF_PROVIDERS}/{instance}/last_error") is None
204 assert _provider_status(_prov_conf(instance), is_loaded=True) == ProviderStatus.LOADED
205
206
207async def test_immediate_flush_for_rotated_token(mass_minimal: MusicAssistant) -> None:
208 """Storing a raw provider value with immediate=True flushes without waiting for the debounce."""
209 config = mass_minimal.config
210 instance = "spotify--test"
211 config.set(
212 f"{CONF_PROVIDERS}/{instance}",
213 {"domain": "spotify", "type": "music", "instance_id": instance, "enabled": True},
214 )
215 with patch.object(config, "save", MagicMock()) as mock_save:
216 config.set_raw_provider_config_value(
217 instance, "refresh_token_global", "token_b", immediate=True
218 )
219 mock_save.assert_called_once_with(immediate=True)
220