chuanqi/cq2_sdk/tests/unit/test_debug_log.py

"""Unit tests for cq2_sdk.debug_log.DebugLogger.

Covers:
  * file creation & naming
  * auto session_id + manual override
  * session_start / session_end markers
  * info / warn / error levels
  * http_request / http_response correlation via req_id
  * body_preview truncation
  * step() markers
  * context manager + auto-error on exception
  * thread-safety (concurrent writes don't corrupt lines)
  * extra= kwargs surfaced in session_start
"""
from __future__ import annotations
import json
import threading
from pathlib import Path

import pytest

from cq2_sdk.debug_log import DebugLogger, _preview, BODY_PREVIEW_LIMIT


# ---------- helpers ----------

def _read_jsonl(path: Path) -> list[dict]:
    with path.open("r", encoding="utf-8") as f:
        return [json.loads(line) for line in f if line.strip()]


# ---------- file creation & session ----------

def test_creates_jsonl_file_in_log_dir(tmp_path):
    with DebugLogger(log_dir=tmp_path) as log:
        log.info("hello")
    files = list(tmp_path.glob("session_*.jsonl"))
    assert len(files) == 1
    lines = _read_jsonl(files[0])
    assert lines[0]["event"] == "session_start"
    assert lines[-1]["event"] == "session_end"
    assert lines[1]["msg"] == "hello"


def test_session_id_format():
    log = DebugLogger(log_dir=Path("/tmp"))
    try:
        # s_YYYYMMDD_HHMMSS_xxxxxx  (6 hex chars)
        assert log.session_id.startswith("s_")
        parts = log.session_id.split("_")
        assert len(parts) == 4  # s, YYYYMMDD, HHMMSS, suffix
        assert len(parts[3]) == 6
        assert all(c in "0123456789abcdef" for c in parts[3])
    finally:
        log.close()


def test_manual_session_id(tmp_path):
    with DebugLogger(session_id="s_test_2026", log_dir=tmp_path) as log:
        log.info("x")
    files = list(tmp_path.glob("session_*.jsonl"))
    assert files[0].name == "session_s_test_2026.jsonl"
    rec = _read_jsonl(files[0])[1]
    assert rec["session"] == "s_test_2026"


def test_script_field_inferred_from_argv(tmp_path):
    # when run via pytest, sys.argv[0] is the pytest binary
    with DebugLogger(log_dir=tmp_path) as log:
        log.info("x")
    lines = _read_jsonl(log.path)
    assert lines[1]["script"]  # non-empty string


# ---------- log() / info / warn / error ----------

def test_log_writes_one_jsonl_line(tmp_path):
    with DebugLogger(log_dir=tmp_path) as log:
        log.log("custom_event", foo=1, bar="x")
    rec = _read_jsonl(log.path)[1]
    assert rec["event"] == "custom_event"
    assert rec["foo"] == 1
    assert rec["bar"] == "x"
    assert "ts" in rec
    # Python's datetime.isoformat() with timezone.utc renders as "+00:00";
    # we accept either that or the "Z" shorthand. Just verify it's parseable.
    from datetime import datetime
    parsed = datetime.fromisoformat(rec["ts"])
    assert parsed.tzinfo is not None  # has tz
    assert parsed.utcoffset().total_seconds() == 0  # is UTC


def test_info_warn_levels(tmp_path):
    with DebugLogger(log_dir=tmp_path) as log:
        log.info("an info", k=1)
        log.warn("a warning", k=2)
    recs = _read_jsonl(log.path)
    assert recs[1]["level"] == "info"
    assert recs[2]["level"] == "warn"
    assert recs[2]["event"] == "warn"


def test_error_from_exception(tmp_path):
    with DebugLogger(log_dir=tmp_path) as log:
        try:
            raise ValueError("bad sig")
        except ValueError as e:
            log.error(e, step="sign_body", account="x")
    rec = _read_jsonl(log.path)[1]
    assert rec["event"] == "error"
    assert rec["level"] == "error"
    assert rec["msg"] == "bad sig"
    assert rec["exc_type"] == "ValueError"
    assert rec["step"] == "sign_body"
    assert rec["account"] == "x"


def test_error_from_string(tmp_path):
    with DebugLogger(log_dir=tmp_path) as log:
        log.error("just a string", step="x")
    rec = _read_jsonl(log.path)[1]
    assert rec["msg"] == "just a string"
    assert rec["exc_type"] is None


# ---------- http correlation ----------

def test_http_request_returns_correlatable_id(tmp_path):
    with DebugLogger(log_dir=tmp_path) as log:
        rid = log.http_request("POST", "https://api/foo", body="<rsa>", step="login")
        log.http_response(rid, status=200, body="<ok>", duration_ms=12.5, step="login")
    recs = _read_jsonl(log.path)
    req = recs[1]
    resp = recs[2]
    assert req["event"] == "http_request"
    assert req["req_id"] == rid
    assert req["method"] == "POST"
    assert req["step"] == "login"
    assert resp["event"] == "http_response"
    assert resp["req_id"] == req["req_id"]
    assert resp["status"] == 200
    assert resp["duration_ms"] == 12.5


def test_http_request_id_is_unique_per_instance(tmp_path):
    with DebugLogger(log_dir=tmp_path) as log:
        ids = {log.http_request("GET", "u") for _ in range(50)}
    assert len(ids) == 50
    assert all(i.startswith("r_") for i in ids)


def test_http_request_with_headers_and_body_dict(tmp_path):
    with DebugLogger(log_dir=tmp_path) as log:
        log.http_request("POST", "https://x", headers={"X-A": "1"},
                         body={"k": "v", "n": 42})
    rec = _read_jsonl(log.path)[1]
    assert rec["headers"] == {"X-A": "1"}
    # body_preview stringifies dict to JSON
    assert '"k": "v"' in rec["body_preview"] or '"k":"v"' in rec["body_preview"]


def test_http_error_event(tmp_path):
    with DebugLogger(log_dir=tmp_path) as log:
        rid = log.http_request("POST", "https://x")
        try:
            raise TimeoutError("connect")
        except TimeoutError as e:
            log.http_error(rid, e, step="connect")
    recs = _read_jsonl(log.path)
    err = recs[2]
    assert err["event"] == "http_error"
    assert err["req_id"] == rid
    assert err["exc_type"] == "TimeoutError"
    assert err["level"] == "error"


# ---------- body_preview ----------

def test_preview_truncates_long_strings():
    big = "x" * (BODY_PREVIEW_LIMIT + 200)
    out = _preview(big)
    assert isinstance(out, str)
    assert out.startswith("x" * BODY_PREVIEW_LIMIT)
    assert "truncated" in out


def test_preview_short_string_passes_through():
    assert _preview("short") == "short"


def test_preview_dict_serializes():
    out = _preview({"a": 1, "b": "x"})
    parsed = json.loads(out)
    assert parsed == {"a": 1, "b": "x"}


def test_preview_none_stays_none():
    assert _preview(None) is None


# ---------- step ----------

def test_step_emits_step_event(tmp_path):
    with DebugLogger(log_dir=tmp_path) as log:
        log.step("login_attempt", account="u1")
        log.step("launch_game", server="玖扬复古176")
    recs = _read_jsonl(log.path)
    assert recs[1]["event"] == "step" and recs[1]["name"] == "login_attempt"
    assert recs[2]["event"] == "step" and recs[2]["name"] == "launch_game"
    assert recs[2]["server"] == "玖扬复古176"  # Chinese preserved


# ---------- context manager ----------

def test_context_manager_writes_session_end(tmp_path):
    with DebugLogger(log_dir=tmp_path) as log:
        log.info("inside")
    recs = _read_jsonl(log.path)
    assert recs[0]["event"] == "session_start"
    assert recs[-1]["event"] == "session_end"


def test_context_manager_auto_logs_exception_on_exit(tmp_path):
    with pytest.raises(RuntimeError):
        with DebugLogger(log_dir=tmp_path) as log:
            raise RuntimeError("boom")
    recs = _read_jsonl(log.path)
    err = [r for r in recs if r["event"] == "error"]
    assert len(err) == 1
    assert err[0]["msg"] == "boom"
    assert err[0]["exc_type"] == "RuntimeError"
    assert err[0]["step"] == "__exit__"


def test_close_idempotent(tmp_path):
    log = DebugLogger(log_dir=tmp_path)
    log.close()
    log.close()  # must not raise
    recs = _read_jsonl(log.path)
    # exactly one session_end
    ends = [r for r in recs if r["event"] == "session_end"]
    assert len(ends) == 1


def test_log_after_close_is_silent(tmp_path):
    log = DebugLogger(log_dir=tmp_path)
    log.close()
    log.info("ignored")  # must not raise
    recs = _read_jsonl(log.path)
    # session_start and session_end have no "msg" key — filter to actual events.
    msg_events = [r for r in recs if r["event"] not in ("session_start", "session_end")]
    assert all(r.get("msg") != "ignored" for r in msg_events)


# ---------- extra kwargs in session_start ----------

def test_extra_kwargs_appear_in_session_start(tmp_path):
    with DebugLogger(log_dir=tmp_path, extra={"branch": "main",
                                              "commit": "5c37d16"}):
        pass
    rec = _read_jsonl(_next_log(tmp_path))[0]
    assert rec["event"] == "session_start"
    assert rec["branch"] == "main"
    assert rec["commit"] == "5c37d16"


# ---------- thread safety ----------

def test_concurrent_writes_dont_corrupt_lines(tmp_path):
    """100 threads × 10 events each. Each line must be valid JSON, no
    interleaving. Verifies the per-instance threading.Lock."""
    n_threads = 100
    per_thread = 10
    errors: list[Exception] = []

    def worker(tid: int, log: DebugLogger) -> None:
        try:
            for i in range(per_thread):
                log.info(f"t{tid}_i{i}", payload={"tid": tid, "i": i})
        except Exception as e:
            errors.append(e)

    with DebugLogger(log_dir=tmp_path) as log:
        threads = [
            threading.Thread(target=worker, args=(t, log))
            for t in range(n_threads)
        ]
        for t in threads:
            t.start()
        for t in threads:
            t.join()

    assert not errors, f"writer errors: {errors}"
    recs = _read_jsonl(log.path)
    # session_start + 100*10 info events + session_end
    assert len(recs) == n_threads * per_thread + 2
    # every line is valid JSON (already parsed) and has session field
    assert all(r["session"] == log.session_id for r in recs)


# ---------- helpers ----------

def _next_log(dirpath: Path) -> Path:
    files = list(dirpath.glob("session_*.jsonl"))
    assert len(files) == 1, files
    return files[0]


# ---------- non-ASCII & Chinese safe ----------

def test_chinese_in_body_and_fields(tmp_path):
    with DebugLogger(log_dir=tmp_path) as log:
        log.info("服务器返回了玖扬复古176的开服列表",  # noqa: E501
                 servers=["玖扬复古176", "复古180"], note="中文OK")
    recs = _read_jsonl(log.path)
    info = recs[1]
    assert "玖扬复古176" in info["msg"]
    assert "玖扬复古176" in info["servers"]
    assert info["note"] == "中文OK"