/
/
/
1"""Tests for the always-on diagnostics facility."""
2
3from __future__ import annotations
4
5import asyncio
6import logging
7import sys
8from typing import TYPE_CHECKING, Any
9from unittest.mock import Mock, patch
10
11import pytest
12from music_assistant_models.auth import Scope
13from music_assistant_models.media_items import (
14 Artist,
15 ProviderMapping,
16 Track,
17 UniqueList,
18)
19
20from music_assistant.controllers.diagnostics import DiagnosticsController
21from music_assistant.helpers.diagnostics import (
22 LOG_RING_MAXLEN,
23 MAX_EXCEPTION_FINGERPRINTS,
24 DiagnosticsLogHandler,
25 install_diagnostics_log_handler,
26 sanitize_data,
27 sanitize_text,
28)
29from music_assistant.helpers.json import json_dumps
30
31if TYPE_CHECKING:
32 from music_assistant.mass import MusicAssistant
33
34
35@pytest.mark.parametrize(
36 ("raw", "must_not_contain", "must_contain"),
37 [
38 # home directories
39 ("error in /Users/johndoe/.musicassistant/file", ["johndoe"], ["~/.musicassistant"]),
40 ("error in /home/johndoe/music-assistant/data", ["johndoe"], ["~"]),
41 (r"error in C:\Users\johndoe\AppData", ["johndoe"], ["~"]),
42 # media file paths reveal library content
43 (
44 "failed to open /media/Pink Floyd/The Wall/01 - In The Flesh.flac",
45 ["Pink Floyd", "The Wall", "In The Flesh"],
46 [".flac", "<path-"],
47 ),
48 ("cannot read Highway to Hell.mp3", ["Highway", "Hell"], [".mp3", "<path-"]),
49 ("failed: '01 - In The Flesh.flac' not found", ["In The Flesh"], ["failed:", "not found"]),
50 # URL credentials and query strings
51 (
52 "GET http://admin:[email protected]/api?token=s3cr3t&x=1",
53 ["admin:hunter2", "s3cr3t", "192.168.1.10"],
54 ["<redacted>@", "<redacted-query>"],
55 ),
56 # query strings on relative request paths (e.g. OAuth callbacks)
57 (
58 "GET /callback?code=s3cr3tcode&state=abc failed",
59 ["s3cr3tcode"],
60 ["/callback?<redacted-query>", "failed"],
61 ),
62 # secret assignments in all common shapes
63 ("password=hunter2", ["hunter2"], ["password=<redacted>"]),
64 ('"api_key": "abc-def-123"', ["abc-def-123"], ["<redacted>"]),
65 ("Authorization: Bearer SflKxwRJSMeKKF2QT4fwpMeJf36POk6yJV", ["SflKxwRJ"], ["<redacted>"]),
66 (
67 "token eyJhbGciOiJIUzI1NiJ9.eyJzdWIiOiIxMjM0In0.SflKxwRJSMeKKF2QT4fwpM",
68 ["eyJhbGci"],
69 ["<redacted-token>"],
70 ),
71 # long token blobs
72 ("blob A1b2C3d4E5f6A1b2C3d4E5f6A1b2C3d4E5f6", ["A1b2C3d4"], ["<redacted-token>"]),
73 # e-mail addresses
74 (
75 "user [email protected] failed",
76 ["john.doe", "example.com"],
77 ["<redacted-email>"],
78 ),
79 # IP addresses (loopback stays)
80 ("connect to 192.168.1.100 failed", ["192.168.1.100"], ["<redacted-ip>"]),
81 ("connect to 2001:db8:85a3::8a2e:370:7334 failed", ["2001:db8"], ["<redacted-ip>"]),
82 ("listening on 127.0.0.1 and ::1", [], ["127.0.0.1", "::1"]),
83 ("connection.20:F8:3B:09:03:E2 closed", ["20:F8:3B"], ["<mac-"]),
84 ("device 20-f8-3b-09-03-e2 offline", ["20-f8-3b"], ["<mac-"]),
85 ("scan done at 01:06:02 (took 12:34:56)", [], ["01:06:02", "12:34:56"]),
86 # timestamps and version numbers must survive
87 ("at 12:34:56.789 version 1.2.3 happened", [], ["12:34:56.789", "1.2.3"]),
88 ],
89)
90def test_sanitize_text(raw: str, must_not_contain: list[str], must_contain: list[str]) -> None:
91 """
92 Test the sanitizer against adversarial fixtures.
93
94 :param raw: The raw input text.
95 :param must_not_contain: Substrings that may not survive sanitization.
96 :param must_contain: Substrings that must be present after sanitization.
97 """
98 result = sanitize_text(raw)
99 for fragment in must_not_contain:
100 assert fragment not in result, f"{fragment!r} leaked into {result!r}"
101 for fragment in must_contain:
102 assert fragment in result, f"{fragment!r} missing from {result!r}"
103
104
105def test_sanitize_text_code_paths() -> None:
106 """Test that absolute code paths are rewritten to be relative to the app root."""
107 raw = 'File "/opt/venv/lib/python3.13/site-packages/aiohttp/web.py", line 12'
108 assert "/opt/venv" not in sanitize_text(raw)
109 assert 'File "aiohttp/web.py", line 12' in sanitize_text(raw)
110 raw = 'File "/opt/app/music_assistant/controllers/music.py", line 5'
111 assert "/opt/app" not in sanitize_text(raw)
112 assert 'File "music_assistant/controllers/music.py", line 5' in sanitize_text(raw)
113
114
115def test_sanitize_data_recurses() -> None:
116 """Test that sanitize_data sanitizes string values in nested structures."""
117 data: dict[str, Any] = {
118 "outer": [{"msg": "password=hunter2"}, "mail me at [email protected]"],
119 "count": 42,
120 "flag": True,
121 "point": (1, "[email protected]"),
122 "/home/marcel/Music/secret song.mp3": "path as key",
123 }
124 result = sanitize_data(data)
125 assert result["outer"][0]["msg"] == "password=<redacted>"
126 assert "[email protected]" not in result["outer"][1]
127 assert result["count"] == 42
128 assert result["flag"] is True
129 assert result["point"] == (1, "<redacted-email>")
130 # dict keys must be sanitized too
131 assert not any("secret song" in key for key in result)
132
133
134def _emit_exception(handler: DiagnosticsLogHandler, message: str = "it broke") -> None:
135 """Raise a ValueError and emit it as an exception log record to the given handler."""
136 logger = logging.getLogger("test.diagnostics")
137 logger.propagate = False
138 logger.addHandler(handler)
139 try:
140 raise ValueError("boom")
141 except ValueError:
142 logger.exception(message)
143 finally:
144 logger.removeHandler(handler)
145
146
147def test_exception_aggregation() -> None:
148 """Test that repeated exceptions aggregate on one fingerprint with count."""
149 handler = DiagnosticsLogHandler()
150 _emit_exception(handler)
151 _emit_exception(handler)
152 _, exceptions = handler.snapshot()
153 assert len(exceptions) == 1
154 entry = exceptions[0]
155 assert entry.count == 2
156 assert entry.exc_type == "ValueError"
157 assert entry.logger_name == "test.diagnostics"
158 assert "ValueError: boom" in entry.render_traceback()
159 assert entry.last_seen >= entry.first_seen
160
161
162def test_exception_lru_bound() -> None:
163 """Test that the exception aggregation stays bounded (LRU eviction)."""
164 handler = DiagnosticsLogHandler()
165 for index in range(MAX_EXCEPTION_FINGERPRINTS + 20):
166 # unique exception type per iteration -> unique fingerprint
167 exc_type = type(f"CustomError{index}", (Exception,), {})
168 try:
169 raise exc_type("boom")
170 except Exception:
171 record = logging.LogRecord(
172 "test", logging.ERROR, __file__, 1, "failed", None, sys.exc_info()
173 )
174 handler.emit(record)
175 _, exceptions = handler.snapshot()
176 assert len(exceptions) == MAX_EXCEPTION_FINGERPRINTS
177
178
179def test_log_ring_bound_and_level() -> None:
180 """Test that the log ring is bounded and only captures WARNING and above."""
181 handler = DiagnosticsLogHandler()
182 logger = logging.getLogger("test.diagnostics.ring")
183 logger.propagate = False
184 logger.setLevel(logging.DEBUG)
185 logger.addHandler(handler)
186 try:
187 logger.info("not captured")
188 for index in range(LOG_RING_MAXLEN + 50):
189 logger.warning("warning %s", index)
190 finally:
191 logger.removeHandler(handler)
192 records, _ = handler.snapshot()
193 assert len(records) == LOG_RING_MAXLEN
194 assert all(record.level == "WARNING" for record in records)
195 assert records[-1].message == f"warning {LOG_RING_MAXLEN + 49}"
196
197
198def test_emit_never_raises() -> None:
199 """Test that a poisoned log record cannot break the capture handler."""
200 handler = DiagnosticsLogHandler()
201 # args/format mismatch makes record.getMessage() raise
202 record = logging.LogRecord("test", logging.ERROR, __file__, 1, "%d", ("nan",), None)
203 handler.emit(record) # must not raise
204
205
206def test_emit_does_no_sanitization_work() -> None:
207 """Test that the always-on capture path never invokes the (expensive) sanitizer."""
208 handler = DiagnosticsLogHandler()
209 with patch("music_assistant.helpers.diagnostics.sanitize_text") as mock_sanitize:
210 _emit_exception(handler)
211 mock_sanitize.assert_not_called()
212
213
214def test_emit_does_no_disk_io() -> None:
215 """Test that capturing an exception never reads source files (linecache)."""
216 handler = DiagnosticsLogHandler()
217 try:
218 raise ValueError("boom")
219 except ValueError:
220 record = logging.LogRecord(
221 "test.diagnostics", logging.ERROR, __file__, 1, "it broke", None, sys.exc_info()
222 )
223 # emit directly: routing through a logger would also invoke unrelated handlers
224 # (e.g. pytest's own log capture) whose formatting does read source files
225 with (
226 patch("linecache.getline", side_effect=AssertionError("linecache hit on emit")),
227 patch("linecache.updatecache", side_effect=AssertionError("linecache hit on emit")),
228 ):
229 handler.emit(record)
230 _, exceptions = handler.snapshot()
231 assert len(exceptions) == 1
232
233
234def test_install_diagnostics_log_handler_idempotent() -> None:
235 """Test that installing the capture handler twice returns the same instance."""
236 handler = install_diagnostics_log_handler()
237 try:
238 assert install_diagnostics_log_handler() is handler
239 assert logging.getLogger().handlers.count(handler) == 1
240 finally:
241 logging.getLogger().removeHandler(handler)
242
243
244async def test_get_report(mass: MusicAssistant) -> None:
245 """
246 Test the full report: shape, bounded size, sanitization and JSON serializability.
247
248 :param mass: Full Music Assistant test instance.
249 """
250 logging.getLogger("music_assistant.test").warning("test warning for the ring")
251 # device identifiers embedded in logger names must be redacted too
252 logging.getLogger("aiosendspin.server.connection.20:F8:3B:09:03:E2").warning("closed")
253 try:
254 raise RuntimeError("report traceback probe")
255 except RuntimeError:
256 logging.getLogger("music_assistant.test").exception("probe failed")
257 report = await mass.diagnostics.get_report()
258 assert report["schema_version"] == 1
259 assert "redaction_notice" in report
260 assert report["system"]["python_version"]
261 assert report["system"]["counts"]["threads"] > 0
262 assert isinstance(report["install"]["providers"], list)
263 assert isinstance(report["install"]["library"]["tracks"], int)
264 assert isinstance(report["exceptions"], list)
265 assert any(
266 entry["type"] == "RuntimeError" and "report traceback probe" in entry["traceback"]
267 for entry in report["exceptions"]
268 )
269 # streams controller contributes its section through the get_diagnostics hook
270 assert "active_output_streams" in report["sections"]["core.streams"]
271 assert "ffmpeg_version" in report["sections"]["core.streams"]
272 # core controllers each contribute their own section
273 assert "db_schema_version" in report["sections"]["core.music"]
274 assert "by_state" in report["sections"]["core.player_queues"]
275 assert "players_synced" in report["sections"]["core.players"]
276 assert "by_status" in report["sections"]["core.tasks"]
277 assert "db_size_mb" in report["sections"]["core.cache"]
278 # log tail is opt-in
279 assert "log_tail" not in report
280 report_with_tail = await mass.diagnostics.get_report(include_log_tail=True)
281 messages = [record["message"] for record in report_with_tail["log_tail"]]
282 assert "test warning for the ring" in messages
283 loggers = [record["logger"] for record in report_with_tail["log_tail"]]
284 assert "aiosendspin.server.connection.<mac-7f85b52c>" in loggers
285 # the whole report must be JSON serializable and stay small
286 assert len(json_dumps(report_with_tail)) < 100_000
287
288
289async def _seed_library_track(mass: MusicAssistant) -> None:
290 """Add a single library track mapped to one (fake) provider instance."""
291
292 def _mapping(item_id: str) -> set[ProviderMapping]:
293 return {
294 ProviderMapping(
295 item_id=item_id,
296 provider_domain="prov_a",
297 provider_instance="prov_a_inst",
298 in_library=True,
299 )
300 }
301
302 artist = await mass.music.artists.add_item_to_library(
303 Artist(
304 item_id="0",
305 provider="library",
306 name="Census Artist",
307 provider_mappings=_mapping("census_artist"),
308 )
309 )
310 await mass.music.tracks.add_item_to_library(
311 Track(
312 item_id="0",
313 provider="library",
314 name="Census Track",
315 provider_mappings=_mapping("census_track"),
316 artists=UniqueList([artist]),
317 )
318 )
319
320
321async def test_library_census_ignores_requesting_user_provider_filter(
322 mass: MusicAssistant,
323) -> None:
324 """
325 Test that the library census reports true totals, not what the requesting admin sees.
326
327 diagnostics/get runs inside the requesting user's context, so a census built from the
328 user-scoped library_count() would silently understate the library in support reports.
329
330 :param mass: Full Music Assistant test instance.
331 """
332 await _seed_library_track(mass)
333 unfiltered_census = await mass.diagnostics._census_library()
334 with patch(
335 "music_assistant.controllers.music.media.base.get_current_user",
336 return_value=Mock(provider_filter=["no_such_provider"]),
337 ):
338 census = await mass.diagnostics._census_library()
339 # the seeded items have no mapping on the filtered provider, so a user-scoped count
340 # would report 0 for them
341 assert census["artists"] == 1
342 assert census["tracks"] == 1
343 # nothing at all may shift when a filtered user is the one asking
344 assert census == unfiltered_census
345
346
347async def test_get_report_command_admin_only(mass: MusicAssistant) -> None:
348 """
349 Test that the diagnostics/get API command is registered with admin-only scope.
350
351 :param mass: Full Music Assistant test instance.
352 """
353 handler = mass.command_handlers["diagnostics/get"]
354 assert handler.required_scope == Scope.SYSTEM_MANAGE
355
356
357async def test_register_section(mass: MusicAssistant) -> None:
358 """
359 Test section registration: contribution, duplicates and unregistration.
360
361 :param mass: Full Music Assistant test instance.
362 """
363
364 async def async_section() -> dict[str, Any]:
365 return {"queued_jobs": 3}
366
367 unregister = mass.diagnostics.register_section("profiler", async_section)
368 unregister_sync = mass.diagnostics.register_section("sync_section", lambda: {"value": 1})
369 with pytest.raises(ValueError, match="already registered"):
370 mass.diagnostics.register_section("profiler", async_section)
371 report = await mass.diagnostics.get_report()
372 assert report["sections"]["profiler"] == {"queued_jobs": 3}
373 assert report["sections"]["sync_section"] == {"value": 1}
374 unregister()
375 unregister_sync()
376 report = await mass.diagnostics.get_report()
377 assert "profiler" not in report["sections"]
378 assert "sync_section" not in report["sections"]
379 # a stale unregister handle must not remove a newer registration with the same name
380 mass.diagnostics.register_section("profiler", lambda: {"value": 2})
381 unregister()
382 report = await mass.diagnostics.get_report()
383 assert report["sections"]["profiler"] == {"value": 2}
384
385
386async def test_section_failure_isolation(mass: MusicAssistant) -> None:
387 """
388 Test that a broken or slow section cannot break the report.
389
390 :param mass: Full Music Assistant test instance.
391 """
392
393 def broken_section() -> dict[str, Any]:
394 raise RuntimeError("contributor exploded with secret password=hunter2")
395
396 async def slow_section() -> dict[str, Any]:
397 await asyncio.sleep(30)
398 return {}
399
400 unregister_broken = mass.diagnostics.register_section("broken", broken_section)
401 unregister_slow = mass.diagnostics.register_section("slow", slow_section)
402 try:
403 with patch("music_assistant.controllers.diagnostics.SECTION_TIMEOUT", 0.1):
404 report = await mass.diagnostics.get_report()
405 assert "hunter2" not in report["sections"]["broken"]["error"]
406 assert "RuntimeError" in report["sections"]["broken"]["error"]
407 assert "error" in report["sections"]["slow"]
408 # healthy sections are unaffected
409 assert "core.streams" in report["sections"]
410 finally:
411 unregister_broken()
412 unregister_slow()
413
414
415async def test_section_sanitization(mass: MusicAssistant) -> None:
416 """
417 Test that section content is sanitized in depth.
418
419 :param mass: Full Music Assistant test instance.
420 """
421 unregister = mass.diagnostics.register_section(
422 "leaky", lambda: {"nested": ["mail [email protected]", {"file": "/music/Artist/song.flac"}]}
423 )
424 try:
425 report = await mass.diagnostics.get_report()
426 leaky = json_dumps(report["sections"]["leaky"])
427 assert "[email protected]" not in leaky
428 assert "Artist" not in leaky
429 finally:
430 unregister()
431
432
433async def test_no_background_work(mass: MusicAssistant) -> None:
434 """
435 Test that the diagnostics facility schedules no background work of its own.
436
437 :param mass: Full Music Assistant test instance.
438 """
439 assert isinstance(mass.diagnostics, DiagnosticsController)
440 assert not [task for task in mass._tracked_tasks if "diagnostics" in task.lower()]
441 assert not [timer for timer in mass._tracked_timers if "diagnostics" in timer.lower()]
442