/
/
/
1"""Tests for the Provider base class serialization contract."""
2
3from __future__ import annotations
4
5import logging
6from dataclasses import fields
7from typing import cast
8from unittest.mock import MagicMock
9
10from music_assistant_models.enums import EventType, ProviderType
11from music_assistant_models.errors import LoginFailed
12from music_assistant_models.provider import ProviderInstance
13
14from music_assistant.constants import VERBOSE_LOG_LEVEL
15from music_assistant.models.provider import Provider
16
17
18def _make_base_provider(log_level: str = "GLOBAL") -> Provider:
19 """Construct a minimal base Provider with stubbed mass/manifest/config."""
20 mass = MagicMock()
21 manifest = MagicMock()
22 manifest.type = ProviderType.MUSIC
23 manifest.domain = "test_provider"
24 config = MagicMock()
25 config.name = "Test Provider"
26 config.instance_id = "test_instance"
27 config.get_value = MagicMock(return_value=log_level)
28 return Provider(mass, manifest, config, supported_features=set())
29
30
31def test_to_dict_matches_provider_instance_schema() -> None:
32 """to_dict() emits exactly the fields declared by the ProviderInstance model."""
33 result = _make_base_provider().to_dict()
34 assert set(result) == {f.name for f in fields(ProviderInstance)}
35 # the served payload must also deserialize back into the model
36 ProviderInstance.from_dict(result)
37
38
39def test_default_name_uses_instance_number_fallback() -> None:
40 """A multi-instance provider without a custom postfix uses its instance number."""
41 provider = _make_base_provider()
42 provider.config.name = None
43 provider.config.instance_id = "test_instance_2"
44 provider.manifest.name = "Test Provider"
45 cast("MagicMock", provider.mass.config.get).return_value = {
46 "test_instance_1": {
47 "domain": "test_provider",
48 "instance_id": "test_instance_1",
49 },
50 "test_instance_2": {
51 "domain": "test_provider",
52 "instance_id": "test_instance_2",
53 },
54 }
55
56 assert provider.default_name == "Test Provider [2]"
57
58
59def test_signal_provider_event() -> None:
60 """signal_provider_event() emits a PROVIDER_EVENT with the instance_id as object_id."""
61 provider = _make_base_provider()
62 provider.signal_provider_event({"foo": "bar"})
63 cast("MagicMock", provider.mass).signal_event.assert_called_once_with(
64 EventType.PROVIDER_EVENT, object_id="test_instance", data={"foo": "bar"}
65 )
66
67
68def test_signal_provider_event_with_sub_scope() -> None:
69 """signal_provider_event() appends the sub_scope to the object_id."""
70 provider = _make_base_provider()
71 provider.signal_provider_event({"round": 1}, sub_scope="game_state")
72 cast("MagicMock", provider.mass).signal_event.assert_called_once_with(
73 EventType.PROVIDER_EVENT, object_id="test_instance/game_state", data={"round": 1}
74 )
75
76
77def test_unload_with_error_schedules_error_unload() -> None:
78 """unload_with_error schedules unload_provider_with_error so the error is recorded."""
79 provider = _make_base_provider()
80 provider.unload_with_error("boom")
81 mass = cast("MagicMock", provider.mass)
82 mass.call_later.assert_called_once_with(
83 1, mass.unload_provider_with_error, "test_instance", "boom"
84 )
85
86
87def test_unload_with_error_forwards_exception() -> None:
88 """An exception is forwarded unchanged so its error code + localized message are preserved."""
89 provider = _make_base_provider()
90 err = LoginFailed("token revoked")
91 provider.unload_with_error(err)
92 mass = cast("MagicMock", provider.mass)
93 mass.call_later.assert_called_once_with(
94 1, mass.unload_provider_with_error, "test_instance", err
95 )
96
97
98async def test_config_change_arms_the_reload_under_the_load_task_id() -> None:
99 """A config change arms the reload so a (re)load starting first cancels it."""
100 provider = _make_base_provider()
101 config = MagicMock()
102 config.instance_id = "test_instance"
103 await provider.update_config(config, {"values/some_setting"})
104 mass = cast("MagicMock", provider.mass)
105 mass.call_later.assert_called_once_with(
106 1, mass.load_provider_config, config, task_id="load_provider_test_instance"
107 )
108
109
110def test_verbose_log_level_stays_scoped_to_the_provider() -> None:
111 """A provider on VERBOSE logs its own records without un-gating unrelated loggers."""
112 logging.addLevelName(VERBOSE_LOG_LEVEL, "VERBOSE")
113 root_logger = logging.getLogger()
114 # a library that never gets an explicit level, so it follows the root logger
115 third_party_logger = logging.getLogger("test_unconfigured_library")
116 emitted: list[str] = []
117
118 class _CaptureHandler(logging.Handler):
119 def emit(self, record: logging.LogRecord) -> None:
120 emitted.append(record.name)
121
122 handler = _CaptureHandler()
123 previous_root_level = root_logger.level
124 root_logger.setLevel(logging.INFO)
125 try:
126 provider = _make_base_provider("VERBOSE")
127 assert provider.logger.level == VERBOSE_LOG_LEVEL
128 assert root_logger.level == logging.INFO
129 assert not third_party_logger.isEnabledFor(logging.DEBUG)
130
131 # the provider's own records must still reach the root handlers
132 root_logger.addHandler(handler)
133 provider.logger.log(VERBOSE_LOG_LEVEL, "verbose record")
134 assert emitted == [provider.logger.name]
135 finally:
136 root_logger.removeHandler(handler)
137 root_logger.setLevel(previous_root_level)
138