music-assistant-server

1"""Unit tests for SafeLogTail (path allowlist, byte cap, redactor)."""
2
3from __future__ import annotations
4
5import re
6import threading
7from pathlib import Path
8from types import SimpleNamespace
9from typing import TYPE_CHECKING, Any, cast
10
11import pytest
12from fastmcp import Client
13from fastmcp.exceptions import ToolError
14
15from music_assistant.providers.fastmcp_server.debug.log_reader import (
16    _MAX_RESPONSE_BYTES,
17    SafeLogTail,
18)
19from music_assistant.providers.fastmcp_server.models import LogTailResult
20
21if TYPE_CHECKING:
22    from music_assistant.mass import MusicAssistant
23
24
25def test_tail_returns_last_n_lines(tmp_log_dir: Path) -> None:  # noqa: ARG001 -- fixture activates SafeLogTail.ROOT patch via monkeypatch
26    """SafeLogTail.tail returns exactly N requested lines."""
27    tail = SafeLogTail()
28    result = tail.tail(lines=5)
29    assert len(result.lines) == 5
30    assert result.bytes_scanned > 0
31
32
33def test_tail_parses_real_ma_log_line_format(tmp_path: Path) -> None:
34    """
35    Pin parser support for both MA-runtime and Python-default log shapes.
36
37    Reproduces the regression caught live in the dev container: real MA writes
38    ``<ts> <LEVEL> (<thread>) [<component>] <msg>``, the synthetic fixture used
39    ``<ts> <LEVEL> <component>: <msg>``. Without this pin the parser silently
40    drops every real MA line to ``timestamp/level/component=None`` and
41    ``debug_tail_log(level="ERROR")`` returns empty.
42    """
43    log_path = tmp_path / "musicassistant.log"
44    log_path.write_text(
45        # MA runtime format with (MainThread) and [bracket] component.
46        "2026-05-28 19:02:21.989 INFO (MainThread) [mcp.server.lowlevel.server] Processing request\n"
47        # Python default format with colon-after-component.
48        "2026-05-28 09:00:00,001 INFO music_assistant.mass: Starting Music Assistant\n",
49        encoding="utf-8",
50    )
51
52    setattr(SafeLogTail, "ROOT", tmp_path)  # noqa: B010 -- redirect class-level log root for this test
53    try:
54        result = SafeLogTail().tail(lines=10)
55    finally:
56        SafeLogTail.ROOT = Path.home() / ".musicassistant"
57
58    assert len(result.lines) == 2
59    by_component = {ln.component: ln for ln in result.lines}
60    assert "mcp.server.lowlevel.server" in by_component
61    assert by_component["mcp.server.lowlevel.server"].level == "INFO"
62    assert "music_assistant.mass" in by_component
63    assert by_component["music_assistant.mass"].level == "INFO"
64
65
66def test_tail_prefers_mass_storage_path_over_class_root(tmp_path: Path) -> None:
67    """
68    When constructed with ``mass``, SafeLogTail reads from ``mass.storage_path``.
69
70    Pins the regression that surfaced live in the dev container: MA is started
71    with ``--data-dir /data`` so the real log lives at ``/data/musicassistant.log``,
72    not at ``Path.home() / ".musicassistant"``. The class-level ``ROOT`` default
73    is wrong for any non-default deployment.
74    """
75    log_path = tmp_path / "musicassistant.log"
76    log_path.write_text("2026-05-28 09:00:00,001 INFO music_assistant.mass: hello\n")
77    mass = cast("MusicAssistant", SimpleNamespace(storage_path=str(tmp_path)))
78
79    tail = SafeLogTail(mass)
80    result = tail.tail(lines=5)
81    assert result.log_path == str(log_path)
82    assert len(result.lines) == 1
83    assert result.lines[0].message == "hello"
84    assert result.truncated is False
85
86
87def test_tail_redacts_bearer_token(tmp_log_dir: Path) -> None:  # noqa: ARG001 -- fixture activates SafeLogTail.ROOT patch via monkeypatch
88    """SafeLogTail redacts Authorization: Bearer tokens."""
89    tail = SafeLogTail()
90    result = tail.tail(lines=200)
91    joined = "\n".join(line.message for line in result.lines)
92    assert "abc.def.ghi" not in joined
93    assert "<redacted>" in joined
94
95
96def test_tail_redacts_query_string_secrets(tmp_log_dir: Path) -> None:  # noqa: ARG001 -- fixture activates SafeLogTail.ROOT patch via monkeypatch
97    """SafeLogTail redacts token= and password= query string values."""
98    tail = SafeLogTail()
99    result = tail.tail(lines=200)
100    joined = "\n".join(line.message for line in result.lines)
101    assert "secret_token_42" not in joined
102    assert "hunter2" not in joined
103
104
105def test_tail_filters_by_level(tmp_log_dir: Path) -> None:  # noqa: ARG001 -- fixture activates SafeLogTail.ROOT patch via monkeypatch
106    """SafeLogTail filters by log level."""
107    tail = SafeLogTail()
108    result = tail.tail(lines=200, level="ERROR")
109    assert all(line.level == "ERROR" for line in result.lines)
110    assert any("lookup failed" in line.message for line in result.lines)
111
112
113def test_tail_filters_by_component_regex(tmp_log_dir: Path) -> None:  # noqa: ARG001 -- fixture activates SafeLogTail.ROOT patch via monkeypatch
114    """SafeLogTail filters by component regex."""
115    tail = SafeLogTail()
116    result = tail.tail(lines=200, component_regex=r"providers\.yandex.*")
117    assert all(
118        line.component and line.component.startswith("music_assistant.providers.yandex")
119        for line in result.lines
120    )
121
122
123@pytest.mark.parametrize(
124    "name",
125    [
126        "../etc/passwd",
127        "/etc/passwd",
128        "musicassistant.log\x00.txt",
129        "..",
130        ".",
131        "",
132        "musicassistant.log.99",
133    ],
134)
135def test_path_traversal_rejected(tmp_log_dir: Path, name: str) -> None:  # noqa: ARG001 -- fixture activates SafeLogTail.ROOT patch via monkeypatch
136    """SafeLogTail rejects path traversal attempts."""
137    tail = SafeLogTail()
138    with pytest.raises(ToolError):
139        tail.tail(lines=1, name=name)
140
141
142def test_symlink_escape_rejected(tmp_log_dir: Path) -> None:
143    """SafeLogTail rejects symlinks pointing outside ROOT."""
144    outside = tmp_log_dir.parent / "outside.log"
145    outside.write_text("LEAK\n")
146    symlink = tmp_log_dir / "musicassistant.log.1"  # in allowlist by basename
147    symlink.symlink_to(outside)
148
149    tail = SafeLogTail()
150    with pytest.raises(ToolError):
151        tail.tail(lines=1, name="musicassistant.log.1")
152
153
154def test_scan_bytes_cap_marks_truncated(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
155    """
156    The 10 MB scan cap fires when filters keep skipping records.
157
158    With filter-then-tail semantics the scan exits early once the page fills,
159    so the cap is observable only when the filter never matches — the reader
160    must then stop at 10 MB, flag ``truncated`` and return an empty page.
161    """
162    from music_assistant.providers.fastmcp_server.debug import log_reader  # noqa: PLC0415
163
164    monkeypatch.setattr(log_reader.SafeLogTail, "ROOT", tmp_path, raising=True)
165    huge = tmp_path / "musicassistant.log"
166    line = b"2026-05-28 09:00:00,001 INFO music_assistant.mass: filler\n"
167    with huge.open("wb") as fh:
168        # 20 MB of repeated, parseable lines.
169        while fh.tell() < 20 * 1024 * 1024:
170            fh.write(line)
171
172    tail = log_reader.SafeLogTail()
173    result = tail.tail(lines=10, level="CRITICAL")
174    assert result.truncated is True
175    assert result.lines == []
176    assert result.bytes_scanned <= 10 * 1024 * 1024 + len(line)
177    assert result.next_call_hint is not None
178
179
180def test_scan_bytes_cap_drops_partial_first_line(
181    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
182) -> None:
183    """
184    When the byte cap fires mid-line, the partial leading fragment must be dropped.
185
186    Without the guard, the backwards line iterator would emit a tail substring
187    of a real line as a headerless record — which slips past level filtering
188    and confuses callers.
189    """
190    from music_assistant.providers.fastmcp_server.debug import log_reader  # noqa: PLC0415
191
192    monkeypatch.setattr(log_reader.SafeLogTail, "ROOT", tmp_path, raising=True)
193    huge = tmp_path / "musicassistant.log"
194    line = b"2026-05-28 09:00:00,001 INFO music_assistant.mass: filler\n"
195    with huge.open("wb") as fh:
196        # 11 MB of identical, parseable lines — enough to trigger the 10 MB cap.
197        while fh.tell() < 11 * 1024 * 1024:
198            fh.write(line)
199
200    tail = log_reader.SafeLogTail()
201    state = log_reader._ScanState()
202    records = list(tail._iter_records_backwards(huge, state))
203    assert state.truncated is True
204    # Every surfaced record must be fully parsed — the partial fragment at the
205    # cap boundary is dropped, never yielded as a headerless record.
206    assert all(rec.entry.timestamp is not None for rec in records), (
207        "partial leading fragment leaked: "
208        + str([r.entry for r in records if r.entry.timestamp is None][:3])
209    )
210
211
212# ---- E2E tests via MCP transport (debug_tail_log tool) ----
213
214
215async def test_e2e_debug_tail_log(mounted_debug: Any, tmp_log_dir: Path) -> None:  # noqa: ARG001 -- fixture activates SafeLogTail.ROOT patch via monkeypatch
216    """debug_tail_log tool returns the last 5 lines via MCP."""
217    async with Client(mounted_debug) as client:
218        result = await client.call_tool("debug_tail_log", {"lines": 5})
219    assert len(result.data.lines) == 5
220    assert result.data.truncated is False
221
222
223async def test_e2e_debug_tail_log_invalid_name(mounted_debug: Any, tmp_log_dir: Path) -> None:  # noqa: ARG001 -- fixture activates SafeLogTail.ROOT patch via monkeypatch
224    """debug_tail_log rejects path traversal attempts via MCP."""
225    async with Client(mounted_debug) as client:
226        with pytest.raises(ToolError):
227            await client.call_tool("debug_tail_log", {"name": "../etc/passwd"})
228
229
230async def test_debug_tail_log_runs_off_event_loop_thread(
231    mounted_debug: Any, monkeypatch: pytest.MonkeyPatch
232) -> None:
233    """
234    The blocking log read is offloaded to a worker thread, not the event loop.
235
236    Synchronous file I/O up to the 10 MB scan cap must not run on MA's single
237    event loop. This pins that ``tail_log`` dispatches the read via a worker
238    thread (it would fail if the tool called ``SafeLogTail.tail`` directly).
239    """
240    main_thread = threading.current_thread()
241    captured: dict[str, Any] = {}
242
243    def fake_tail(_self: Any, **_kwargs: Any) -> LogTailResult:
244        captured["thread"] = threading.current_thread()
245        return LogTailResult(log_path="x", lines=[], bytes_scanned=0, truncated=False)
246
247    monkeypatch.setattr(SafeLogTail, "tail", fake_tail)
248    async with Client(mounted_debug) as client:
249        await client.call_tool("debug_tail_log", {"lines": 5})
250    assert captured["thread"] is not main_thread
251
252
253# ---- Spec 0017: filter-then-tail, record grouping, search, budgets, stats ----
254
255
256def _write_log(tmp_path: Path, text: str) -> None:
257    (tmp_path / "musicassistant.log").write_text(text, encoding="utf-8")
258
259
260@pytest.fixture
261def log_root(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
262    """Empty sandboxed log root; tests write their own log content."""
263    from music_assistant.providers.fastmcp_server.debug import log_reader  # noqa: PLC0415
264
265    monkeypatch.setattr(log_reader.SafeLogTail, "ROOT", tmp_path, raising=True)
266    return tmp_path
267
268
269def _line(ts_ms: int, level: str, component: str, msg: str) -> str:
270    return f"2026-05-28 09:00:{ts_ms // 1000:02d},{ts_ms % 1000:03d} {level} {component}: {msg}\n"
271
272
273def test_tail_filter_then_tail_finds_buried_errors(log_root: Path) -> None:
274    """lines=N with level counts N *matching* records, not N raw lines."""
275    body = _line(1000, "ERROR", "music_assistant.streams", "err one")
276    body += _line(2000, "ERROR", "music_assistant.streams", "err two")
277    for i in range(300):
278        body += _line(3000 + i * 10, "INFO", "music_assistant.mass", f"chatter {i}")
279    _write_log(log_root, body)
280
281    result = SafeLogTail().tail(lines=2, level="ERROR")
282    assert [ln.message for ln in result.lines] == ["err one", "err two"]
283
284
285def test_tail_groups_traceback_into_error_record(log_root: Path) -> None:
286    """Continuation lines (tracebacks) attach to the preceding record."""
287    body = _line(1000, "INFO", "music_assistant.mass", "fine")
288    body += _line(2000, "ERROR", "music_assistant.streams", "Unhandled exception")
289    body += "Traceback (most recent call last):\n"
290    body += '  File "stream.py", line 10, in pump\n'
291    body += "ValueError: bad frame\n"
292    body += _line(3000, "INFO", "music_assistant.mass", "recovered")
293    _write_log(log_root, body)
294
295    result = SafeLogTail().tail(lines=10, level="ERROR")
296    assert len(result.lines) == 1
297    record = result.lines[0]
298    assert record.level == "ERROR"
299    assert "Unhandled exception" in record.message
300    assert "Traceback (most recent call last):" in record.message
301    assert "ValueError: bad frame" in record.message
302
303
304def test_tail_level_threshold_case_insensitive(log_root: Path) -> None:
305    """Level is a case-insensitive minimum-severity threshold."""
306    body = _line(1000, "DEBUG", "c.a", "dbg")
307    body += _line(2000, "INFO", "c.b", "inf")
308    body += _line(3000, "WARNING", "c.c", "warn")
309    body += _line(4000, "ERROR", "c.d", "err")
310    body += _line(5000, "CRITICAL", "c.e", "crit")
311    _write_log(log_root, body)
312
313    result = SafeLogTail().tail(lines=10, level="warning")
314    assert [ln.level for ln in result.lines] == ["WARNING", "ERROR", "CRITICAL"]
315
316
317def test_tail_invalid_level_raises(log_root: Path) -> None:
318    """An unknown level name raises ToolError naming valid levels."""
319    _write_log(log_root, _line(1000, "INFO", "c.a", "x"))
320    with pytest.raises(ToolError, match=r"(?i)level"):
321        SafeLogTail().tail(lines=1, level="LOUD")
322
323
324def test_tail_search_matches_message_and_traceback(log_root: Path) -> None:
325    """Search greps the full record text, including continuation lines."""
326    body = _line(1000, "INFO", "c.a", "playback started")
327    body += _line(2000, "ERROR", "c.b", "boom")
328    body += "ValueError: playback pipeline broke\n"
329    body += _line(3000, "INFO", "c.c", "unrelated")
330    _write_log(log_root, body)
331
332    result = SafeLogTail().tail(lines=10, search="PLAYBACK")
333    messages = [ln.message for ln in result.lines]
334    assert len(messages) == 2
335    assert any("playback started" in m for m in messages)
336    assert any("pipeline broke" in m for m in messages)
337
338
339def test_tail_invalid_search_regex_raises(log_root: Path) -> None:
340    """A malformed search regex raises ToolError naming the parameter."""
341    _write_log(log_root, _line(1000, "INFO", "c.a", "x"))
342    with pytest.raises(ToolError, match="search"):
343        SafeLogTail().tail(lines=1, search="[unclosed")
344
345
346def test_tail_has_more_and_next_call_hint(log_root: Path) -> None:
347    """When matches remain beyond the page, has_more is set and the hint pages by timestamp."""
348    body = ""
349    for i in range(10):
350        body += _line(1000 * (i + 1), "ERROR", "c.a", f"err {i}")
351    _write_log(log_root, body)
352
353    result = SafeLogTail().tail(lines=3, level="ERROR")
354    assert [ln.message for ln in result.lines] == ["err 7", "err 8", "err 9"]
355    assert result.has_more is True
356    assert result.next_call_hint is not None
357    assert "before=" in result.next_call_hint
358
359
360def test_tail_no_hint_when_page_complete(log_root: Path) -> None:
361    """A complete page reports has_more=False and no hint."""
362    _write_log(log_root, _line(1000, "ERROR", "c.a", "only"))
363    result = SafeLogTail().tail(lines=5, level="ERROR")
364    assert result.has_more is False
365    assert result.response_truncated is False
366    assert result.next_call_hint is None
367
368
369def test_tail_response_budget_truncates_with_hint(log_root: Path) -> None:
370    """The response byte budget cuts the page short with an explicit flag + hint."""
371    big = "x" * 8000
372    body = ""
373    for i in range(20):
374        body += _line(1000 * (i + 1), "INFO", "c.a", f"{i} {big}")
375    _write_log(log_root, body)
376
377    result = SafeLogTail().tail(lines=20)
378    assert result.response_truncated is True
379    assert 0 < len(result.lines) < 20
380    assert result.next_call_hint is not None
381
382
383def test_tail_before_cursor_pages_older_records(log_root: Path) -> None:
384    """before=<ISO ts> returns only records strictly older than the cursor."""
385    body = ""
386    for i in range(5):
387        body += _line(1000 * (i + 1), "INFO", "c.a", f"msg {i}")
388    _write_log(log_root, body)
389
390    first = SafeLogTail().tail(lines=2)
391    assert [ln.message for ln in first.lines] == ["msg 3", "msg 4"]
392    oldest_ts = first.lines[0].timestamp
393    assert oldest_ts is not None
394
395    second = SafeLogTail().tail(lines=2, before=oldest_ts)
396    assert [ln.message for ln in second.lines] == ["msg 1", "msg 2"]
397
398
399def test_stats_counts_levels_components_and_range(log_root: Path) -> None:
400    """stats() aggregates per-level counts, top components and the time range."""
401    body = _line(1000, "INFO", "c.alpha", "one")
402    body += _line(2000, "ERROR", "c.beta", "two")
403    body += "Traceback (most recent call last):\n"
404    body += _line(3000, "ERROR", "c.beta", "three")
405    body += _line(4000, "WARNING", "c.alpha", "four")
406    _write_log(log_root, body)
407
408    stats = SafeLogTail().stats()
409    assert stats.total_records == 4
410    assert stats.level_counts["ERROR"] == 2
411    assert stats.level_counts["INFO"] == 1
412    assert stats.level_counts["WARNING"] == 1
413    top = {c.component: c.count for c in stats.top_components}
414    assert top == {"c.alpha": 2, "c.beta": 2}
415    assert stats.first_timestamp is not None
416    assert stats.last_timestamp is not None
417    assert stats.first_timestamp < stats.last_timestamp
418
419
420def test_count_errors_includes_critical(log_root: Path) -> None:
421    """count_errors_last_5min counts ERROR and above within the window."""
422    from datetime import datetime  # noqa: PLC0415
423
424    # Log timestamps carry no TZ and are parsed as local time — write local.
425    recent = datetime.now().strftime("%Y-%m-%d %H:%M:%S,000")  # noqa: DTZ005
426    body = "2020-01-01 00:00:00,000 ERROR c.c: ancient err\n"
427    body += f"{recent} ERROR c.a: recent err\n"
428    body += f"{recent} CRITICAL c.b: recent crit\n"
429    _write_log(log_root, body)
430
431    assert SafeLogTail().count_errors_last_5min() == 2
432
433
434async def test_e2e_debug_log_stats(mounted_debug: Any, tmp_log_dir: Path) -> None:  # noqa: ARG001 -- fixture activates SafeLogTail.ROOT patch via monkeypatch
435    """debug_log_stats tool aggregates the sample log via MCP."""
436    async with Client(mounted_debug) as client:
437        result = await client.call_tool("debug_log_stats", {})
438    assert result.data.total_records > 0
439    assert result.data.level_counts["ERROR"] >= 1
440    assert len(result.data.top_components) > 0
441
442
443def test_tail_offset_cursor_survives_same_timestamp_burst(log_root: Path) -> None:
444    """
445    Paging via the hint's offset cursor never loses same-timestamp records.
446
447    A timestamp-only cursor drops tied records (ms resolution + error bursts);
448    the hint must page by file offset so every record is reachable.
449    """
450    body = "".join(_line(1000, "ERROR", "c.a", f"burst {i}") for i in range(5))
451    _write_log(log_root, body)
452
453    page1 = SafeLogTail().tail(lines=2, level="ERROR")
454    assert [ln.message for ln in page1.lines] == ["burst 3", "burst 4"]
455    assert page1.has_more is True
456    assert page1.next_call_hint is not None
457    match = re.search(r"before='(offset:\d+)'", page1.next_call_hint)
458    assert match, f"hint lacks offset cursor: {page1.next_call_hint!r}"
459
460    page2 = SafeLogTail().tail(lines=2, level="ERROR", before=match.group(1))
461    assert [ln.message for ln in page2.lines] == ["burst 1", "burst 2"]
462    match2 = re.search(r"before='(offset:\d+)'", page2.next_call_hint or "")
463    assert match2
464
465    page3 = SafeLogTail().tail(lines=2, level="ERROR", before=match2.group(1))
466    assert [ln.message for ln in page3.lines] == ["burst 0"]
467    assert page3.has_more is False
468
469
470def test_tail_oversized_record_truncated_by_bytes(log_root: Path) -> None:
471    """
472    A single oversized record is cut against the byte budget, not characters.
473
474    Multi-byte UTF-8 (Cyrillic/CJK) must not blow the response budget when the
475    character count is under it but the encoded size is several times larger.
476    """
477    big = "я" * 100_000  # 200k bytes in UTF-8
478    _write_log(log_root, _line(1000, "ERROR", "c.a", big))
479
480    result = SafeLogTail().tail(lines=5)
481    assert result.response_truncated is True
482    assert len(result.lines) == 1
483    encoded = result.lines[0].message.encode("utf-8", errors="replace")
484    assert len(encoded) <= _MAX_RESPONSE_BYTES + 100  # small slack for the marker
485    assert "…[message truncated]" in result.lines[0].message
486