/
/
/
1"""End-to-end tests for the DEBUG_PROVIDERS tool group."""
2
3from __future__ import annotations
4
5from types import SimpleNamespace
6from typing import Any
7from unittest.mock import AsyncMock, MagicMock
8
9import pytest
10from fastmcp import Client
11from fastmcp.exceptions import ToolError
12
13
14def _provider(
15 instance_id: str, *, available: bool = True, last_error: str | None = None
16) -> SimpleNamespace:
17 return SimpleNamespace(
18 instance_id=instance_id,
19 domain=instance_id.split("_", maxsplit=1)[0],
20 type=SimpleNamespace(value="music"),
21 name=instance_id,
22 available=available,
23 last_error=last_error,
24 )
25
26
27async def test_list_providers_rolls_up_state(mounted_debug: Any, mock_mass: MagicMock) -> None:
28 """debug_list_providers returns instance_id, domain, type, availability, and last_error."""
29 mock_mass.providers = [
30 _provider("yandex_music_1"),
31 _provider("sonos_1", available=False, last_error="auth expired"),
32 ]
33 async with Client(mounted_debug) as client:
34 result = await client.call_tool("debug_list_providers", {})
35 summaries = result.data.providers
36 assert {s.instance_id for s in summaries} == {"yandex_music_1", "sonos_1"}
37 failing = next(s for s in summaries if s.instance_id == "sonos_1")
38 assert failing.available is False
39 assert failing.last_error == "auth expired"
40
41
42async def test_inspect_provider_config_masks_secret_string(
43 mounted_debug: Any, mock_mass: MagicMock
44) -> None:
45 """debug_inspect_provider_config calls Config.to_dict() whose __post_serialize__ masks SECURE_STRING."""
46 # The real masking is enforced by music_assistant_models __post_serialize__.
47 # The test mock here returns a pre-masked dict shape consistent with that contract.
48 masked_config = MagicMock()
49 masked_config.domain = "yandex_music"
50 masked_config.to_dict = MagicMock(
51 return_value={
52 "domain": "yandex_music",
53 "values": {
54 "username": {"key": "username", "type": "string", "value": "ren"},
55 "password": {
56 "key": "password",
57 "type": "secure_string",
58 "value": "this_value_is_encrypted",
59 },
60 },
61 }
62 )
63 mock_mass.config.get_provider_config = AsyncMock(return_value=masked_config)
64
65 async with Client(mounted_debug) as client:
66 result = await client.call_tool(
67 "debug_inspect_provider_config",
68 {"instance_id": "yandex_music_1"},
69 )
70 by_key = {v.key: v for v in result.data.values}
71 assert by_key["password"].value == "this_value_is_encrypted"
72 # The actual secret literal never appears anywhere in the response.
73 assert "actual-secret-1234" not in str(result.data)
74
75
76async def test_inspect_provider_config_not_found_raises(
77 mounted_debug: Any, mock_mass: MagicMock
78) -> None:
79 """debug_inspect_provider_config raises ToolError when instance_id is unknown."""
80 mock_mass.config.get_provider_config = AsyncMock(side_effect=Exception("not found"))
81 async with Client(mounted_debug) as client:
82 with pytest.raises(ToolError, match="provider instance_id="):
83 await client.call_tool(
84 "debug_inspect_provider_config",
85 {"instance_id": "nonexistent"},
86 )
87
88
89async def test_list_webserver_routes_returns_entries(
90 mounted_debug: Any, mock_mass: MagicMock
91) -> None:
92 """debug_list_webserver_routes enumerates routes from webserver._server.app.router."""
93 route_obj = SimpleNamespace(
94 method="GET",
95 resource=SimpleNamespace(canonical="/mcp/v1/sse"),
96 )
97 inner_app = SimpleNamespace(router=SimpleNamespace(routes=lambda: [route_obj]))
98 mock_mass.webserver._server = SimpleNamespace(app=inner_app)
99 async with Client(mounted_debug) as client:
100 result = await client.call_tool("debug_list_webserver_routes", {})
101 paths = [r.path for r in result.data.routes]
102 assert "/mcp/v1/sse" in paths
103
104
105async def test_list_webserver_routes_includes_registered_by(
106 mounted_debug: Any, mock_mass: MagicMock
107) -> None:
108 """debug_list_webserver_routes infers registered_by from path prefix."""
109 route1 = SimpleNamespace(method="GET", resource=SimpleNamespace(canonical="/mcp/v1/sse"))
110 route2 = SimpleNamespace(method="POST", resource=SimpleNamespace(canonical="/api/health"))
111 inner_app = SimpleNamespace(router=SimpleNamespace(routes=lambda: [route1, route2]))
112 mock_mass.webserver._server = SimpleNamespace(app=inner_app)
113 async with Client(mounted_debug) as client:
114 result = await client.call_tool("debug_list_webserver_routes", {})
115 by_path = {r.path: r.registered_by for r in result.data.routes}
116 assert "fastmcp_server" in by_path["/mcp/v1/sse"]
117 assert by_path["/api/health"] == "music_assistant (api)"
118
119
120async def test_list_webserver_routes_unavailable_returns_error(
121 mounted_debug: Any, mock_mass: MagicMock
122) -> None:
123 """debug_list_webserver_routes raises ToolError when webserver._server is unavailable."""
124 mock_mass.webserver._server = None
125 async with Client(mounted_debug) as client:
126 with pytest.raises(ToolError, match="routes are unavailable"):
127 await client.call_tool("debug_list_webserver_routes", {})
128
129
130async def test_list_package_versions_includes_fastmcp(mounted_debug: Any) -> None:
131 """debug_list_package_versions includes fastmcp, music_assistant_models, etc."""
132 async with Client(mounted_debug) as client:
133 result = await client.call_tool("debug_list_package_versions", {})
134 assert "fastmcp" in result.data.packages
135 assert "music_assistant_models" in result.data.packages
136 # Check that version strings are non-empty and not marked "not installed"
137 assert len(result.data.packages["fastmcp"]) > 0
138 assert result.data.packages["fastmcp"] != "<not installed>"
139