music-assistant-server

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    with (
218        patch("linecache.getline", side_effect=AssertionError("linecache hit on emit")),
219        patch("linecache.updatecache", side_effect=AssertionError("linecache hit on emit")),
220    ):
221        _emit_exception(handler)
222    _, exceptions = handler.snapshot()
223    assert len(exceptions) == 1
224
225
226def test_install_diagnostics_log_handler_idempotent() -> None:
227    """Test that installing the capture handler twice returns the same instance."""
228    handler = install_diagnostics_log_handler()
229    try:
230        assert install_diagnostics_log_handler() is handler
231        assert logging.getLogger().handlers.count(handler) == 1
232    finally:
233        logging.getLogger().removeHandler(handler)
234
235
236async def test_get_report(mass: MusicAssistant) -> None:
237    """
238    Test the full report: shape, bounded size, sanitization and JSON serializability.
239
240    :param mass: Full Music Assistant test instance.
241    """
242    logging.getLogger("music_assistant.test").warning("test warning for the ring")
243    # device identifiers embedded in logger names must be redacted too
244    logging.getLogger("aiosendspin.server.connection.20:F8:3B:09:03:E2").warning("closed")
245    try:
246        raise RuntimeError("report traceback probe")
247    except RuntimeError:
248        logging.getLogger("music_assistant.test").exception("probe failed")
249    report = await mass.diagnostics.get_report()
250    assert report["schema_version"] == 1
251    assert "redaction_notice" in report
252    assert report["system"]["python_version"]
253    assert report["system"]["counts"]["threads"] > 0
254    assert isinstance(report["install"]["providers"], list)
255    assert isinstance(report["install"]["library"]["tracks"], int)
256    assert isinstance(report["exceptions"], list)
257    assert any(
258        entry["type"] == "RuntimeError" and "report traceback probe" in entry["traceback"]
259        for entry in report["exceptions"]
260    )
261    # streams controller contributes its section through the get_diagnostics hook
262    assert "active_output_streams" in report["sections"]["core.streams"]
263    assert "ffmpeg_version" in report["sections"]["core.streams"]
264    # core controllers each contribute their own section
265    assert "db_schema_version" in report["sections"]["core.music"]
266    assert "by_state" in report["sections"]["core.player_queues"]
267    assert "players_synced" in report["sections"]["core.players"]
268    assert "by_status" in report["sections"]["core.tasks"]
269    assert "db_size_mb" in report["sections"]["core.cache"]
270    # log tail is opt-in
271    assert "log_tail" not in report
272    report_with_tail = await mass.diagnostics.get_report(include_log_tail=True)
273    messages = [record["message"] for record in report_with_tail["log_tail"]]
274    assert "test warning for the ring" in messages
275    loggers = [record["logger"] for record in report_with_tail["log_tail"]]
276    assert "aiosendspin.server.connection.<mac-7f85b52c>" in loggers
277    # the whole report must be JSON serializable and stay small
278    assert len(json_dumps(report_with_tail)) < 100_000
279
280
281async def _seed_library_track(mass: MusicAssistant) -> None:
282    """Add a single library track mapped to one (fake) provider instance."""
283
284    def _mapping(item_id: str) -> set[ProviderMapping]:
285        return {
286            ProviderMapping(
287                item_id=item_id,
288                provider_domain="prov_a",
289                provider_instance="prov_a_inst",
290                in_library=True,
291            )
292        }
293
294    artist = await mass.music.artists.add_item_to_library(
295        Artist(
296            item_id="0",
297            provider="library",
298            name="Census Artist",
299            provider_mappings=_mapping("census_artist"),
300        )
301    )
302    await mass.music.tracks.add_item_to_library(
303        Track(
304            item_id="0",
305            provider="library",
306            name="Census Track",
307            provider_mappings=_mapping("census_track"),
308            artists=UniqueList([artist]),
309        )
310    )
311
312
313async def test_library_census_ignores_requesting_user_provider_filter(
314    mass: MusicAssistant,
315) -> None:
316    """
317    Test that the library census reports true totals, not what the requesting admin sees.
318
319    diagnostics/get runs inside the requesting user's context, so a census built from the
320    user-scoped library_count() would silently understate the library in support reports.
321
322    :param mass: Full Music Assistant test instance.
323    """
324    await _seed_library_track(mass)
325    unfiltered_census = await mass.diagnostics._census_library()
326    with patch(
327        "music_assistant.controllers.music.media.base.get_current_user",
328        return_value=Mock(provider_filter=["no_such_provider"]),
329    ):
330        census = await mass.diagnostics._census_library()
331    # the seeded items have no mapping on the filtered provider, so a user-scoped count
332    # would report 0 for them
333    assert census["artists"] == 1
334    assert census["tracks"] == 1
335    # nothing at all may shift when a filtered user is the one asking
336    assert census == unfiltered_census
337
338
339async def test_get_report_command_admin_only(mass: MusicAssistant) -> None:
340    """
341    Test that the diagnostics/get API command is registered with admin-only scope.
342
343    :param mass: Full Music Assistant test instance.
344    """
345    handler = mass.command_handlers["diagnostics/get"]
346    assert handler.required_scope == Scope.SYSTEM_MANAGE
347
348
349async def test_register_section(mass: MusicAssistant) -> None:
350    """
351    Test section registration: contribution, duplicates and unregistration.
352
353    :param mass: Full Music Assistant test instance.
354    """
355
356    async def async_section() -> dict[str, Any]:
357        return {"queued_jobs": 3}
358
359    unregister = mass.diagnostics.register_section("profiler", async_section)
360    unregister_sync = mass.diagnostics.register_section("sync_section", lambda: {"value": 1})
361    with pytest.raises(ValueError, match="already registered"):
362        mass.diagnostics.register_section("profiler", async_section)
363    report = await mass.diagnostics.get_report()
364    assert report["sections"]["profiler"] == {"queued_jobs": 3}
365    assert report["sections"]["sync_section"] == {"value": 1}
366    unregister()
367    unregister_sync()
368    report = await mass.diagnostics.get_report()
369    assert "profiler" not in report["sections"]
370    assert "sync_section" not in report["sections"]
371    # a stale unregister handle must not remove a newer registration with the same name
372    mass.diagnostics.register_section("profiler", lambda: {"value": 2})
373    unregister()
374    report = await mass.diagnostics.get_report()
375    assert report["sections"]["profiler"] == {"value": 2}
376
377
378async def test_section_failure_isolation(mass: MusicAssistant) -> None:
379    """
380    Test that a broken or slow section cannot break the report.
381
382    :param mass: Full Music Assistant test instance.
383    """
384
385    def broken_section() -> dict[str, Any]:
386        raise RuntimeError("contributor exploded with secret password=hunter2")
387
388    async def slow_section() -> dict[str, Any]:
389        await asyncio.sleep(30)
390        return {}
391
392    unregister_broken = mass.diagnostics.register_section("broken", broken_section)
393    unregister_slow = mass.diagnostics.register_section("slow", slow_section)
394    try:
395        with patch("music_assistant.controllers.diagnostics.SECTION_TIMEOUT", 0.1):
396            report = await mass.diagnostics.get_report()
397        assert "hunter2" not in report["sections"]["broken"]["error"]
398        assert "RuntimeError" in report["sections"]["broken"]["error"]
399        assert "error" in report["sections"]["slow"]
400        # healthy sections are unaffected
401        assert "core.streams" in report["sections"]
402    finally:
403        unregister_broken()
404        unregister_slow()
405
406
407async def test_section_sanitization(mass: MusicAssistant) -> None:
408    """
409    Test that section content is sanitized in depth.
410
411    :param mass: Full Music Assistant test instance.
412    """
413    unregister = mass.diagnostics.register_section(
414        "leaky", lambda: {"nested": ["mail [email protected]", {"file": "/music/Artist/song.flac"}]}
415    )
416    try:
417        report = await mass.diagnostics.get_report()
418        leaky = json_dumps(report["sections"]["leaky"])
419        assert "[email protected]" not in leaky
420        assert "Artist" not in leaky
421    finally:
422        unregister()
423
424
425async def test_no_background_work(mass: MusicAssistant) -> None:
426    """
427    Test that the diagnostics facility schedules no background work of its own.
428
429    :param mass: Full Music Assistant test instance.
430    """
431    assert isinstance(mass.diagnostics, DiagnosticsController)
432    assert not [task for task in mass._tracked_tasks if "diagnostics" in task.lower()]
433    assert not [timer for timer in mass._tracked_timers if "diagnostics" in timer.lower()]
434