/
/
/
1"""End-to-end tests for debug_health_summary."""
2
3from __future__ import annotations
4
5from datetime import datetime
6from types import SimpleNamespace
7from typing import Any
8from unittest.mock import MagicMock
9
10import pytest
11from fastmcp import Client, FastMCP
12
13from music_assistant.providers.fastmcp_server.debug import log_reader
14from music_assistant.providers.fastmcp_server.tools.debug import build_debug_server
15
16
17@pytest.fixture
18def populated_mass(mock_mass: MagicMock) -> MagicMock:
19 """Populate mock_mass with providers and queues."""
20 mock_mass.providers = [
21 SimpleNamespace(
22 instance_id="yandex_music_1",
23 domain="yandex_music",
24 type=SimpleNamespace(value="music"),
25 name="Yandex",
26 available=True,
27 enabled=True,
28 last_error=None,
29 ),
30 SimpleNamespace(
31 instance_id="sonos_1",
32 domain="sonos",
33 type=SimpleNamespace(value="player"),
34 name="Sonos",
35 available=False,
36 enabled=True,
37 last_error="auth expired",
38 ),
39 SimpleNamespace(
40 instance_id="spotify_1",
41 domain="spotify",
42 type=SimpleNamespace(value="music"),
43 name="Spotify",
44 available=True,
45 enabled=False,
46 last_error=None,
47 ),
48 ]
49 mock_mass.player_queues.all = MagicMock(
50 return_value=[
51 SimpleNamespace(queue_id="kitchen", state="playing", available=True),
52 SimpleNamespace(queue_id="lenco", state="idle", available=True),
53 SimpleNamespace(queue_id="broken", state="error", available=False),
54 ]
55 )
56 return mock_mass
57
58
59async def test_health_summary_rolls_up_state(mounted_debug: Any, populated_mass: MagicMock) -> None: # noqa: ARG001
60 """Test that health_summary correctly rolls up provider and queue state."""
61 async with Client(mounted_debug) as client:
62 result = await client.call_tool("debug_health_summary", {})
63 data = result.data
64 assert data.providers_loaded == 2 # yandex + spotify
65 assert data.providers_disabled == 1 # spotify
66 assert data.providers_error == 1 # sonos
67 assert any(p.instance_id == "sonos_1" for p in data.providers_error_details)
68 assert data.queues_total == 3
69 assert data.queues_with_active_playback == 1
70 assert data.queues_with_errors >= 1
71
72
73async def test_health_summary_marks_capabilities_disabled_when_off(
74 mounted_debug: Any,
75 populated_mass: MagicMock, # noqa: ARG001
76) -> None:
77 """Test that disabled capabilities are listed when underlying system unavailable."""
78 async with Client(mounted_debug) as client:
79 result = await client.call_tool("debug_health_summary", {})
80 # mounted_debug has no live EventBuffer and SafeLogTail.ROOT points
81 # at $HOME/.musicassistant which doesn't exist in the test environment.
82 # Both should report as disabled, NOT crash.
83 assert result.data.events_per_min_by_type is None
84 assert "DEBUG_EVENTS" in result.data.disabled_capabilities
85
86
87async def test_health_summary_events_rate_when_buffer_present(
88 mounted_debug_with_events: Any,
89 populated_mass: MagicMock, # noqa: ARG001
90) -> None:
91 """Test that events_per_min_by_type is populated when EventBuffer is active."""
92 mcp, _buf, emitter = mounted_debug_with_events
93 for _ in range(6):
94 emitter.emit(SimpleNamespace(event="player_updated", object_id="kitchen", data={}))
95 async with Client(mcp) as client:
96 result = await client.call_tool("debug_health_summary", {})
97 assert result.data.events_per_min_by_type is not None
98 assert "player_updated" in result.data.events_per_min_by_type
99
100
101async def test_health_summary_counts_recent_log_errors(
102 mounted_debug: Any,
103 populated_mass: MagicMock,
104 tmp_path: Any,
105 monkeypatch: pytest.MonkeyPatch,
106) -> None:
107 """Write a synthetic log with current-time ERROR lines, assert count."""
108 monkeypatch.setattr(log_reader.SafeLogTail, "ROOT", tmp_path, raising=True)
109 # health_summary's SafeLogTail(mass) now reads mass.storage_path first;
110 # point it at the same sandbox so the patched ROOT and the runtime
111 # resolution agree.
112 populated_mass.storage_path = str(tmp_path)
113 log_path = tmp_path / "musicassistant.log"
114 now = datetime.now().astimezone()
115 ts = now.strftime("%Y-%m-%d %H:%M:%S,000")
116 log_path.write_text(
117 f"{ts} ERROR music_assistant.providers.sonos: failure A\n"
118 f"{ts} ERROR music_assistant.providers.sonos: failure B\n"
119 f"{ts} INFO music_assistant.mass: not an error\n"
120 f"{ts} ERROR music_assistant.controllers.music: failure C\n",
121 encoding="utf-8",
122 )
123 async with Client(mounted_debug) as client:
124 result = await client.call_tool("debug_health_summary", {})
125 assert result.data.log_errors_last_5min == 3
126 assert "DEBUG_LOGS" not in result.data.disabled_capabilities
127
128
129async def test_health_summary_skips_log_read_when_logs_disabled(
130 mock_mass: MagicMock,
131 populated_mass: MagicMock, # noqa: ARG001 -- populates mock_mass providers/queues
132 monkeypatch: pytest.MonkeyPatch,
133) -> None:
134 """
135 When DEBUG_LOGS is off, health_summary must not touch the log file at all.
136
137 Reading logs to count errors when the operator disabled log access bypasses
138 the permission. The capability is reported as disabled instead.
139 """
140
141 def boom(_self: Any, **_kwargs: Any) -> int:
142 raise AssertionError("logs must not be read when DEBUG_LOGS is disabled")
143
144 monkeypatch.setattr(log_reader.SafeLogTail, "count_errors_last_5min", boom, raising=True)
145
146 mcp = FastMCP(name="test")
147 mcp.mount(
148 build_debug_server(mock_mass, require_confirmation=False, logs_enabled=False),
149 namespace="debug",
150 )
151 async with Client(mcp) as client:
152 result = await client.call_tool("debug_health_summary", {})
153 assert result.data.log_errors_last_5min is None
154 assert "DEBUG_LOGS" in result.data.disabled_capabilities
155