"""客户端运行日志上报:writer 单测 + 端点集成测试。""" from __future__ import annotations import json from pathlib import Path import pytest from app.core import client_log @pytest.fixture() def client_log_file(tmp_path, monkeypatch): """把客户端日志切到临时文件,并重置 writer 单例使其按当时 env 重建。""" p = tmp_path / "app-client.log" monkeypatch.setenv("CLIENT_LOG_FILE", str(p)) client_log.reset_client_logger() yield p client_log.reset_client_logger() def _read_lines(p: Path) -> list[dict]: text = p.read_text(encoding="utf-8").strip() return [json.loads(ln) for ln in text.splitlines() if ln] # ---------------- writer 层 ---------------- def _meta(**kw) -> dict: base = {"device_id": "d-1", "user_id": None, "app_ver": None, "platform": None, "sent_at": None} base.update(kw) return base def test_writer_writes_one_line_per_record(client_log_file): recs = [ {"client_ts": 1737000000000, "level": "info", "msg": "hello"}, {"client_ts": 1737000000001, "level": "error", "msg": "boom", "tag": "net"}, ] received, dropped = client_log.write_records( recs, meta=_meta(user_id=42, app_ver="1.2.3", platform="android"), client_ip="1.2.3.4", ) assert (received, dropped) == (2, 0) lines = _read_lines(client_log_file) assert len(lines) == 2 assert lines[0]["source"] == "client" assert lines[0]["service"] == "app-client" assert lines[0]["device_id"] == "d-1" assert lines[0]["user_id"] == 42 assert lines[0]["app_ver"] == "1.2.3" assert lines[0]["client_ip"] == "1.2.3.4" assert lines[0]["level"] == "INFO" # 归一化大写 assert lines[0]["msg"] == "hello" assert lines[0]["client_ts"] == 1737000000000 assert lines[1]["tag"] == "net" def test_writer_hoists_trace_id_to_top_level(client_log_file): client_log.write_records( [{"client_ts": 1, "level": "info", "msg": "x", "trace_id": "abc123"}], meta=_meta(), client_ip="", ) assert _read_lines(client_log_file)[0]["trace_id"] == "abc123" def test_writer_sweeps_unknown_keys_into_data(client_log_file): client_log.write_records( [{"client_ts": 1, "level": "info", "msg": "x", "foo": 123, "data": {"bar": "baz"}}], meta=_meta(), client_ip="", ) line = _read_lines(client_log_file)[0] assert "foo" not in line # 白名单外不进顶层 assert line["data"]["foo"] == 123 # 兜底进 data assert line["data"]["bar"] == "baz" # 客户端自带 data 合并进来 def test_writer_truncates_oversize_msg(client_log_file, monkeypatch): monkeypatch.setenv("APPLOG_MAX_MSG_BYTES", "10") received, dropped = client_log.write_records( [{"client_ts": 1, "level": "info", "msg": "x" * 100}], meta=_meta(), client_ip="", ) assert (received, dropped) == (1, 0) # 截断而非丢弃 line = _read_lines(client_log_file)[0] assert line["msg_truncated"] is True assert line["msg"].endswith("…[truncated]") body = line["msg"].removesuffix("…[truncated]") assert len(body.encode("utf-8")) <= 10 # 截断体不超过字节上限 def test_writer_caps_indexed_field_lengths(client_log_file): client_log.write_records( [{"client_ts": 1, "level": "info" * 20, "msg": "x", "trace_id": "t" * 1000, "tag": "g" * 1000}], meta=_meta(), client_ip="", ) line = _read_lines(client_log_file)[0] assert len(line["level"]) <= 16 # 索引字段(spec §8)做长度上限 assert len(line["trace_id"]) <= 256 assert len(line["tag"]) <= 128 def test_writer_logger_does_not_propagate(client_log_file): lg = client_log.get_logger() assert lg.name == "shagua.client_log" assert lg.propagate is False # 不冒泡到 root → 不写 app-server.log def test_writer_drops_bad_record_without_raising(client_log_file, monkeypatch): def _boom(*a, **kw): raise RuntimeError("boom") monkeypatch.setattr(client_log, "_build_line", _boom) received, dropped = client_log.write_records( [{"client_ts": 1, "level": "info", "msg": "x"}], meta=_meta(), client_ip="", ) assert (received, dropped) == (0, 1) # 坏条计 dropped,且不抛给上层 # ---------------- 端点层 ---------------- def _post(client, body): return client.post("/api/v1/applog/batch", json=body) def test_endpoint_happy_path(client, client_log_file): body = { "device_id": "d-1", "user_id": 42, "app_ver": "1.2.3", "platform": "android", "logs": [ {"client_ts": 1, "level": "info", "msg": "a", "trace_id": "t1"}, {"client_ts": 2, "level": "warn", "msg": "b"}, ], } resp = _post(client, body) assert resp.status_code == 200 assert resp.json() == {"ok": True, "received": 2, "dropped": 0} lines = _read_lines(client_log_file) assert len(lines) == 2 assert lines[0]["trace_id"] == "t1" # 端到端:trace_id 落到文件顶层 assert lines[0]["client_ip"] # 服务端补了 IP def test_endpoint_rejects_batch_over_max(client, client_log_file): body = {"device_id": "d-1", "logs": [{"client_ts": i, "level": "info", "msg": str(i)} for i in range(501)]} resp = _post(client, body) assert resp.status_code == 422 # Pydantic max_length=500 def test_endpoint_rejects_body_over_limit(client, client_log_file, monkeypatch): # 只测有 Content-Length 的常规情形(TestClient/requests 恒发该头)。缺该头(chunked) # 时依赖 nginx client_max_body_size 兜底,不在应用层测。 monkeypatch.setenv("APPLOG_MAX_BODY_BYTES", "50") body = {"device_id": "d-1", "logs": [{"client_ts": 1, "level": "info", "msg": "x" * 500}]} resp = _post(client, body) assert resp.status_code == 413 # 依赖查 Content-Length,body 校验前拦截 def test_endpoint_returns_200_on_write_failure(client, client_log_file, monkeypatch): class _BoomLogger: name = "shagua.client_log" propagate = False handlers: list = [] def info(self, *a, **k): raise RuntimeError("disk full") monkeypatch.setattr(client_log, "get_logger", lambda: _BoomLogger()) body = {"device_id": "d-1", "logs": [ {"client_ts": 1, "level": "info", "msg": "a"}, {"client_ts": 2, "level": "info", "msg": "b"}]} resp = _post(client, body) assert resp.status_code == 200 # fire-and-forget:写失败不 500 assert resp.json() == {"ok": True, "received": 0, "dropped": 2} def test_endpoint_missing_device_id_is_422(client, client_log_file): resp = _post(client, {"logs": [{"client_ts": 1, "level": "info", "msg": "a"}]}) assert resp.status_code == 422 # device_id 必填