music-assistant-server

1"""Lifecycle tests for the debug sub-server inside MCPServerRuntime."""
2
3from __future__ import annotations
4
5from unittest.mock import MagicMock
6
7
8async def test_setup_subscribes_when_debug_events_enabled(
9    mock_mass: MagicMock, mock_config: MagicMock
10) -> None:
11    """When DEBUG_EVENTS=True, EventBuffer is started during runtime.start()."""
12    mock_config.get_value.side_effect = lambda key, default=None: {
13        "debug_events": True,
14        "debug_event_buffer_capacity": 100,
15    }.get(key, default if default is not None else False)
16
17    from music_assistant.providers.fastmcp_server.server import MCPServerRuntime  # noqa: PLC0415
18
19    runtime = MCPServerRuntime(mock_mass, mock_config, logger=MagicMock())
20    await runtime.start()
21    try:
22        assert mock_mass.subscribe.called
23        assert runtime._event_buffer is not None
24    finally:
25        await runtime.stop()
26
27
28async def test_unload_does_not_raise_when_events_disabled(
29    mock_mass: MagicMock, mock_config: MagicMock
30) -> None:
31    """With DEBUG_EVENTS=False, no subscription is created and stop() doesn't raise."""
32    mock_config.get_value.return_value = False
33    from music_assistant.providers.fastmcp_server.server import MCPServerRuntime  # noqa: PLC0415
34
35    runtime = MCPServerRuntime(mock_mass, mock_config, logger=MagicMock())
36    await runtime.start()
37    await runtime.stop()  # must not raise
38    assert mock_mass.subscribe.called is False
39
40
41async def test_unload_stops_event_buffer_before_unmount(
42    mock_mass: MagicMock, mock_config: MagicMock
43) -> None:
44    """EventBuffer.stop() runs BEFORE FastMCP transport teardown (lifespan-race protection)."""
45    mock_config.get_value.side_effect = lambda key, default=None: {
46        "debug_events": True,
47        "debug_event_buffer_capacity": 100,
48    }.get(key, default if default is not None else False)
49
50    from music_assistant.providers.fastmcp_server.server import MCPServerRuntime  # noqa: PLC0415
51
52    runtime = MCPServerRuntime(mock_mass, mock_config, logger=MagicMock())
53    await runtime.start()
54
55    call_order: list[str] = []
56    buf = runtime._event_buffer
57    assert buf is not None
58    original_stop = buf.stop
59
60    def _stop_record() -> None:
61        call_order.append("buffer_stop")
62        original_stop()
63
64    setattr(buf, "stop", _stop_record)  # noqa: B010 -- patch bound method for call-order assertion
65
66    original_unmount = runtime._unmount
67
68    async def _unmount_record() -> None:
69        call_order.append("unmount")
70        if original_unmount is not None:
71            await original_unmount()
72
73    setattr(runtime, "_unmount", _unmount_record)  # noqa: B010 -- patch teardown hook for call-order assertion
74
75    await runtime.stop()
76    assert call_order, "no recorded teardown calls"
77    assert call_order[0] == "buffer_stop", call_order
78
79
80async def test_event_buffer_stop_idempotent_during_unload(
81    mock_mass: MagicMock, mock_config: MagicMock
82) -> None:
83    """Double-stop must not raise."""
84    mock_config.get_value.side_effect = lambda key, default=None: {
85        "debug_events": True,
86        "debug_event_buffer_capacity": 100,
87    }.get(key, default if default is not None else False)
88
89    from music_assistant.providers.fastmcp_server.server import MCPServerRuntime  # noqa: PLC0415
90
91    runtime = MCPServerRuntime(mock_mass, mock_config, logger=MagicMock())
92    await runtime.start()
93    await runtime.stop()
94    await runtime.stop()  # must not raise — second call is a no-op
95