planet/backend/tests/test_system_logs.py

from __future__ import annotations

import json

from pathlib import Path

from app.services import system_logs


class FakeRedis:
    def __init__(self) -> None:
        self.store: dict[str, list[str]] = {}

    def rpush(self, key: str, value: str) -> None:
        self.store.setdefault(key, []).append(value)

    def ltrim(self, key: str, start: int, end: int) -> None:
        items = self.store.get(key, [])
        normalized_end = None if end == -1 else end + 1
        self.store[key] = items[start:normalized_end]

    def expire(self, key: str, seconds: int) -> None:
        return None

    def lrange(self, key: str, start: int, end: int) -> list[str]:
        items = self.store.get(key, [])
        normalized_end = None if end == -1 else end + 1
        return items[start:normalized_end]

    def llen(self, key: str) -> int:
        return len(self.store.get(key, []))


def test_read_log_snapshot_uses_structured_buffer_timestamp_level_and_search(monkeypatch):
    fake_redis = FakeRedis()
    monkeypatch.setattr(system_logs, "redis_client", fake_redis)
    monkeypatch.setattr(
        system_logs,
        "LOG_SOURCES",
        {
            "earth-client": system_logs.LogSource(
                source_id="earth-client",
                name="Earth 浏览器端",
                kind="buffer",
                location="redis://planet:system_logs:earth-client",
                description="Earth 浏览器端上报日志",
                category="client",
                buffer_key=system_logs.get_buffer_log_key("earth-client"),
            )
        },
    )

    fake_redis.rpush(
        system_logs.get_buffer_log_key("earth-client"),
        json.dumps(
            {
                "timestamp": "2026-04-22T10:15:30Z",
                "level": "warning",
                "message": "news feed degraded",
                "context": {"module": "news", "detail": "timeout"},
            },
            ensure_ascii=False,
        ),
    )
    fake_redis.rpush(
        system_logs.get_buffer_log_key("earth-client"),
        json.dumps(
            {
                "timestamp": "2026-04-23T06:01:00Z",
                "level": "error",
                "message": "landing points failed",
                "context": {"module": "layer-startup", "detail": "http 500"},
            },
            ensure_ascii=False,
        ),
    )

    snapshot = system_logs.read_log_snapshot(
        "earth-client",
        50,
        levels="error,warning",
        start_date="2026-04-23",
        end_date="2026-04-23",
        search="landing",
    )

    assert snapshot is not None
    assert snapshot["selected_levels"] == ["error", "warning"]
    assert snapshot["search_query"] == "landing"
    assert snapshot["line_count"] == 1
    assert snapshot["lines"][0].startswith("2026-04-23 06:01:00 ERROR landing points failed")
    assert snapshot["daily_markers"] == [
        {"date_token": "2026-04-23", "total": 1, "dominant_level": "error"}
    ]


def test_read_log_snapshot_parses_file_timestamp_and_builds_markers(tmp_path: Path, monkeypatch):
    log_path = tmp_path / "backend.log"
    log_path.write_text(
        "\n".join(
            [
                "2026-04-22 08:00:00 INFO service booted",
                "2026-04-23 09:15:00 WARNING disk pressure detected",
                "2026-04-23 09:16:00 ERROR sync failed",
                "2026-04-24 10:00:00 DEBUG collector trace",
            ]
        ),
        encoding="utf-8",
    )

    monkeypatch.setattr(
        system_logs,
        "LOG_SOURCES",
        {
            "backend": system_logs.LogSource(
                source_id="backend",
                name="后端服务",
                kind="file",
                location=str(log_path),
                description="测试文件日志",
                category="service",
            )
        },
    )

    snapshot = system_logs.read_log_snapshot(
        "backend",
        50,
        levels="warning,error",
        search="failed",
    )

    assert snapshot is not None
    assert snapshot["line_count"] == 1
    assert snapshot["lines"] == ["2026-04-23 09:16:00 ERROR sync failed"]
    assert snapshot["daily_markers"] == [
        {"date_token": "2026-04-23", "total": 1, "dominant_level": "error"}
    ]
    assert snapshot["status"] == "ok"


def test_append_buffer_log_persists_normalized_level(monkeypatch):
    fake_redis = FakeRedis()
    monkeypatch.setattr(system_logs, "redis_client", fake_redis)

    system_logs.append_buffer_log(
        "earth-client",
        level="warn",
        message="feed delayed",
        context={"module": "news"},
    )

    stored_items = fake_redis.lrange(system_logs.get_buffer_log_key("earth-client"), 0, -1)
    payload = json.loads(stored_items[0])
    assert payload["level"] == "warning"
    assert payload["message"] == "feed delayed"


def test_infer_log_level_prefers_leading_prefix_over_query_string():
    line = 'INFO: 127.0.0.1 - "GET /api/v1/system/logs/backend?limit=200&level=error&levels=error HTTP/1.1" 200 OK'

    entry = system_logs.parse_text_log_entry(line)

    assert entry.level == "info"


def test_parse_text_log_entry_does_not_promote_exception_context_to_error():
    line = "websockets.exceptions.ConnectionClosedError: sent 1011 (internal error) keepalive ping timeout"

    entry = system_logs.parse_text_log_entry(line)

    assert entry.level is None


def test_parse_text_log_entry_still_detects_explicit_error_prefix():
    line = "ERROR: [Errno 98] Address already in use"

    entry = system_logs.parse_text_log_entry(line)

    assert entry.level == "error"


def test_read_log_snapshot_strips_nul_bytes_from_file_lines(tmp_path: Path, monkeypatch):
    log_path = tmp_path / "backend.log"
    log_path.write_bytes(
        (
            b"INFO: service booted\n"
            b"ERROR: bind failed\n"
            + b"\x00" * 32
            + b"2026-04-23 23:41:32 INFO service=backend message=request served\n"
        )
    )

    monkeypatch.setattr(
        system_logs,
        "LOG_SOURCES",
        {
            "backend": system_logs.LogSource(
                source_id="backend",
                name="后端服务",
                kind="file",
                location=str(log_path),
                description="测试文件日志",
                category="service",
            )
        },
    )

    snapshot = system_logs.read_log_snapshot("backend", 50)

    assert snapshot is not None
    assert snapshot["line_count"] == 3
    assert snapshot["lines"] == [
        "INFO: service booted",
        "ERROR: bind failed",
        "2026-04-23 23:41:32 INFO service=backend message=request served",
    ]