music-assistant-server

1"""
2End-to-end test for the permission-tag hot-swap on a live FastMCP root.
3
4``MCPServerRuntime.apply_permission_change`` has unit-level routing tests
5(in ``test_apply_permission_change.py``) and a tag-filter middleware test
6(in ``test_middleware.py``), but **no integration test** that mounts a real
7FastMCP root, flips a permission via the runtime, and asserts the visible
8tool surface actually changed. An inverted ``permission_only`` predicate or
9a stale closure capture would slip through every existing unit test.
10
11This file builds a runtime whose tag-set is mutated by
12``apply_permission_change`` and verifies that the in-memory ``Client``'s
13``list_tools`` reflects the new set **without** a stop/start cycle.
14"""
15# mypy: disable-error-code="arg-type, no-untyped-def, type-arg, assignment, operator, misc, attr-defined"
16
17from __future__ import annotations
18
19import logging
20from typing import Any
21from unittest.mock import MagicMock
22
23import pytest
24from fastmcp import Client, FastMCP
25
26from music_assistant.providers.fastmcp_server.middleware import TagFilterMiddleware
27from music_assistant.providers.fastmcp_server.server import MCPServerRuntime, build_tag_lookup
28
29
30def _build_runtime_with_mounted_server(
31    mock_mass: MagicMock, mock_config: MagicMock
32) -> tuple[MCPServerRuntime, FastMCP]:
33    """
34    Construct a real ``MCPServerRuntime`` with a small FastMCP mounted.
35
36    We skip the full ``start()`` (which would mount into MA's webserver) and
37    build the FastMCP root by hand — same shape as the production start path,
38    minus the ASGI / route registration that has nothing to do with the
39    hot-swap behaviour under test.
40    """
41    from music_assistant.providers.fastmcp_server.tags import enabled_tags  # noqa: PLC0415
42    from music_assistant.providers.fastmcp_server.tools import (  # noqa: PLC0415
43        build_library_server,
44        build_volume_server,
45    )
46
47    runtime = MCPServerRuntime(mock_mass, mock_config, logging.getLogger("t"))
48
49    mcp: FastMCP = FastMCP(name="hotswap-test")
50    mcp.mount(build_library_server(mock_mass), namespace="library")
51    mcp.mount(build_volume_server(mock_mass), namespace="volume")
52    runtime._mcp = mcp
53
54    # Hand-wire the same middleware closure that ``MCPServerRuntime.start``
55    # would install — mutating ``runtime._allowed_tags`` after this point
56    # should change the visible tool set on the next ``list_tools`` call.
57    runtime._allowed_tags = {str(t) for t in enabled_tags(mock_config)}
58    mcp.add_middleware(TagFilterMiddleware(lambda: runtime._allowed_tags, build_tag_lookup(mcp)))
59    return runtime, mcp
60
61
62def _set_config_values(cfg: MagicMock, **overrides: Any) -> None:
63    """Mutate the ``_values`` dict on a ``mock_config`` fixture in place."""
64    cfg._values.update(overrides)
65
66
67@pytest.mark.asyncio
68async def test_hot_swap_makes_disabled_tool_visible_without_restart(
69    mock_mass: MagicMock, mock_config: MagicMock
70) -> None:
71    """
72    Enabling a permission via ``apply_permission_change`` exposes its tools.
73
74    With ``control_volume=False`` (the default in ``mock_config``), the
75    ``volume_*`` tools are invisible to clients. Flipping the bit to True
76    via ``apply_permission_change(changed_keys={"control_volume"})`` must
77    make them visible on the next ``list_tools`` — without any ``stop()``
78    or ``start()`` call.
79    """
80    runtime, mcp = _build_runtime_with_mounted_server(mock_mass, mock_config)
81
82    async with Client(mcp) as client:
83        names_before = {t.name for t in await client.list_tools()}
84        assert not any(n.startswith("volume_") for n in names_before), (
85            f"volume tools should be hidden by default; got names={names_before!r}"
86        )
87        # library tools are enabled by default, so at least one should show up.
88        assert any(n.startswith("library_") for n in names_before), names_before
89
90    # Flip the permission ON and dispatch the hot-swap.
91    _set_config_values(mock_config, control_volume=True)
92    await runtime.apply_permission_change(mock_config, changed_keys={"control_volume"})
93
94    async with Client(mcp) as client:
95        names_after = {t.name for t in await client.list_tools()}
96    assert any(n.startswith("volume_") for n in names_after), (
97        f"control_volume hot-swap did not expose volume_* tools; got names={names_after!r}"
98    )
99
100
101@pytest.mark.asyncio
102async def test_hot_swap_hides_previously_visible_tool(
103    mock_mass: MagicMock, mock_config: MagicMock
104) -> None:
105    """
106    The reverse direction also works: disabling a permission hides its tools.
107
108    Start with ``query_library=True`` (default), confirm library_* is visible,
109    flip to ``False``, confirm it's hidden on the next list — same FastMCP
110    root, no restart.
111    """
112    runtime, mcp = _build_runtime_with_mounted_server(mock_mass, mock_config)
113
114    async with Client(mcp) as client:
115        names_before = {t.name for t in await client.list_tools()}
116        assert any(n.startswith("library_") for n in names_before), names_before
117
118    _set_config_values(mock_config, query_library=False)
119    await runtime.apply_permission_change(mock_config, changed_keys={"query_library"})
120
121    async with Client(mcp) as client:
122        names_after = {t.name for t in await client.list_tools()}
123    assert not any(n.startswith("library_") for n in names_after), (
124        f"hot-swap to query_library=False did not hide library_* tools; got {names_after!r}"
125    )
126