/
/
/
1"""End-to-end tests for the CONFIG_WRITE_PROVIDER tool group."""
2# ruff: noqa: D103, PLC0415
3# D103: test functions don't need docstrings.
4# PLC0415: mock reconfiguration inside test bodies requires deferred imports.
5
6from __future__ import annotations
7
8import logging
9from typing import Any
10
11import pytest
12from fastmcp import Client
13from fastmcp.exceptions import ToolError
14
15
16def _decliner() -> Any:
17 from fastmcp.client.elicitation import ElicitResult
18
19 async def handler(*_a: Any, **_kw: Any) -> ElicitResult:
20 return ElicitResult(action="decline")
21
22 return handler
23
24
25async def test_set_provider_value_persists(mounted_config: Any, mock_config_targets: Any) -> None:
26 async with Client(mounted_config) as client:
27 result = await client.call_tool(
28 "config_set_provider_value",
29 {"instance_id": "yandex_music", "key": "log_level", "value": "DEBUG"},
30 )
31 mock_config_targets.config.save_provider_config.assert_awaited_once()
32 assert result.data.applied is True
33
34
35async def test_dry_run_returns_diff_no_persist(
36 mounted_config: Any, mock_config_targets: Any
37) -> None:
38 async with Client(mounted_config) as client:
39 result = await client.call_tool(
40 "config_set_provider_value",
41 {"instance_id": "yandex_music", "key": "log_level", "value": "DEBUG", "dry_run": True},
42 )
43 assert result.data.applied is False
44 assert result.data.diff is not None
45 mock_config_targets.config.save_provider_config.assert_not_called()
46
47
48async def test_set_provider_value_validation_rejects(
49 mounted_config: Any,
50 mock_config_targets: Any, # noqa: ARG001
51) -> None:
52 async with Client(mounted_config) as client:
53 with pytest.raises(ToolError, match="failed validation"):
54 await client.call_tool(
55 "config_set_provider_value",
56 {"instance_id": "yandex_music", "key": "http_port", "value": 999999},
57 )
58
59
60async def test_requires_reload_flag_surfaced(mounted_config: Any, mock_config_targets: Any) -> None: # noqa: ARG001
61 # token entry has requires_reload=True in the fixture and is SECURE_STRING;
62 # mounted_config has all tags incl. secret, so this is allowed.
63 async with Client(mounted_config) as client:
64 result = await client.call_tool(
65 "config_set_provider_value",
66 {"instance_id": "yandex_music", "key": "token", "value": "newtok"},
67 )
68 assert result.data.requires_reload is True
69
70
71async def test_confirm_declined_blocks_write(mock_config_targets: Any) -> None:
72 from fastmcp import FastMCP
73
74 from music_assistant.providers.fastmcp_server.tools.config import build_config_server
75
76 mcp = FastMCP(name="t")
77 mcp.mount(
78 build_config_server(mock_config_targets, require_confirmation=True), namespace="config"
79 )
80 async with Client(mcp, elicitation_handler=_decliner()) as client:
81 with pytest.raises(ToolError):
82 await client.call_tool(
83 "config_set_provider_value",
84 {"instance_id": "yandex_music", "key": "log_level", "value": "DEBUG"},
85 )
86 mock_config_targets.config.save_provider_config.assert_not_called()
87
88
89async def test_trigger_action_relays_entries(mounted_config: Any, mock_config_targets: Any) -> None:
90 async with Client(mounted_config) as client:
91 result = await client.call_tool(
92 "config_trigger_provider_action",
93 {"instance_id": "yandex_music", "action_key": "auth_qr"},
94 )
95 assert result.data.action_key == "auth_qr"
96 mock_config_targets.config.invoke_provider_config_action.assert_awaited()
97
98
99async def test_save_provider_bulk_persists(mounted_config: Any, mock_config_targets: Any) -> None:
100 async with Client(mounted_config) as client:
101 result = await client.call_tool(
102 "config_save_provider",
103 {"instance_id": "yandex_music", "values": {"log_level": "DEBUG", "http_port": 8095}},
104 )
105 mock_config_targets.config.save_provider_config.assert_awaited_once()
106 assert result.data.applied is True
107
108
109async def test_secret_write_delegates_plaintext_and_never_logs_it(
110 mounted_config: Any, mock_config_targets: Any, caplog: Any
111) -> None:
112 with caplog.at_level(logging.INFO, logger="music_assistant.providers.fastmcp_server.config"):
113 async with Client(mounted_config) as client:
114 await client.call_tool(
115 "config_set_provider_value",
116 {"instance_id": "yandex_music", "key": "token", "value": "sup3rsecret"},
117 )
118 args = mock_config_targets.config.save_provider_config.await_args
119 assert "sup3rsecret" in str(args) # plaintext passed to MA (which encrypts)
120 assert "sup3rsecret" not in caplog.text # never logged
121
122
123def test_action_outcome_localizes_a_translation_key() -> None:
124 from unittest.mock import MagicMock
125
126 from music_assistant_models.config_entries import ConfigActionResult
127
128 from music_assistant.providers.fastmcp_server.tools.config import _action_outcome
129
130 mass = MagicMock()
131 # stand in for the resolver, echoing what it was asked to resolve
132 mass.translations.get_translation.side_effect = lambda key, owner=None, params=None: (
133 f"{key}|{owner}|{','.join(params or ())}"
134 )
135 result = ConfigActionResult(
136 translation_key="clear_cache.result", translation_owner="core.cache"
137 )
138
139 assert _action_outcome(mass, result) == {
140 "message": "config_actions.clear_cache.result|core.cache|"
141 }
142 # positional args reach the resolver so a templated message renders filled in
143 templated = ConfigActionResult(
144 translation_key="cleanup.result", translation_owner="core.cache", translation_args=[3]
145 )
146
147 assert _action_outcome(mass, templated) == {
148 "message": "config_actions.cleanup.result|core.cache|3"
149 }
150
151
152def test_action_outcome_reports_a_url_and_omits_absent_fields() -> None:
153 from unittest.mock import MagicMock
154
155 from music_assistant_models.config_entries import ConfigActionResult
156
157 from music_assistant.providers.fastmcp_server.tools.config import _action_outcome
158
159 result = ConfigActionResult(open_url="/mcp/v1/connect?bootstrap=jwt-xyz")
160
161 assert _action_outcome(MagicMock(), result) == {"open_url": "/mcp/v1/connect?bootstrap=jwt-xyz"}
162
163
164async def test_write_tools_use_interactive_timeout() -> None:
165 """
166 Confirmation-gated writes must not use the 10s fast timeout.
167
168 The elicitation round-trip + save+reload needs a human-scale window.
169 Regression for a live timeout-mid-confirmation bug.
170 """
171 from unittest.mock import MagicMock
172
173 from music_assistant.providers.fastmcp_server.tools._common import (
174 TIMEOUT_FAST,
175 TIMEOUT_INTERACTIVE,
176 )
177 from music_assistant.providers.fastmcp_server.tools.config import build_config_server
178
179 assert TIMEOUT_INTERACTIVE > TIMEOUT_FAST
180 sub = build_config_server(MagicMock(), require_confirmation=True)
181 tools = await sub.list_tools()
182 write_names = {
183 "set_provider_value",
184 "save_provider",
185 "trigger_provider_action",
186 "set_core_value",
187 "save_core",
188 "set_player_value",
189 "save_player",
190 "save_dsp",
191 }
192 read_names = {
193 "list_targets",
194 "get_provider",
195 "get_core",
196 "get_player",
197 "get_entries",
198 "get_dsp",
199 }
200 tools_by_name = {tool.name: tool for tool in tools}
201 for name in write_names:
202 assert name in tools_by_name, f"{name} not found in tools"
203 tool = tools_by_name[name]
204 assert tool.timeout == TIMEOUT_INTERACTIVE, (
205 f"{name} should use TIMEOUT_INTERACTIVE (120s), got {tool.timeout}s"
206 )
207 for name in read_names:
208 assert name in tools_by_name, f"{name} not found in tools"
209 tool = tools_by_name[name]
210 assert tool.timeout == TIMEOUT_FAST, (
211 f"{name} should keep TIMEOUT_FAST (10s), got {tool.timeout}s"
212 )
213