music-assistant-server

1"""Lifecycle test: config sub-server is mounted with secret-writes wiring."""
2
3from __future__ import annotations
4
5from unittest.mock import MagicMock
6
7
8async def test_config_server_mounted_and_visible(
9    mock_mass: MagicMock, mock_config: MagicMock
10) -> None:
11    """Config sub-server is mounted and tools are visible through the root MCP."""
12    mock_config.get_value.side_effect = lambda key, default=None: {
13        "config_read": True,
14        "config_write_provider": True,
15        "config_write_secret": True,
16    }.get(key, default if default is not None else False)
17
18    from music_assistant.providers.fastmcp_server.server import MCPServerRuntime  # noqa: PLC0415
19
20    runtime = MCPServerRuntime(mock_mass, mock_config, logger=MagicMock())
21    await runtime.start()
22    try:
23        from fastmcp import Client  # noqa: PLC0415
24
25        async with Client(runtime._mcp) as client:
26            names = {t.name for t in await client.list_tools()}
27        assert "config_get_provider" in names
28        assert "config_set_provider_value" in names
29    finally:
30        await runtime.stop()
31
32
33async def test_config_secret_flag_threaded(mock_mass: MagicMock, mock_config: MagicMock) -> None:
34    """secret_writes_enabled is threaded from CONF_CONFIG_WRITE_SECRET config value."""
35    # config_write_secret OFF → provider/read/write visible, but the
36    # secret value-gate is closed. We assert the runtime starts cleanly
37    # with the flag off (the gate behavior itself is unit-tested elsewhere).
38    mock_config.get_value.side_effect = lambda key, default=None: {
39        "config_read": True,
40        "config_write_provider": True,
41        "config_write_secret": False,
42    }.get(key, default if default is not None else False)
43
44    from music_assistant.providers.fastmcp_server.server import MCPServerRuntime  # noqa: PLC0415
45
46    runtime = MCPServerRuntime(mock_mass, mock_config, logger=MagicMock())
47    await runtime.start()
48    try:
49        from fastmcp import Client  # noqa: PLC0415
50
51        async with Client(runtime._mcp) as client:
52            names = {t.name for t in await client.list_tools()}
53        assert "config_set_provider_value" in names
54    finally:
55        await runtime.stop()
56