tao.chenandtao.chen
9f7b6ba18c
健康检查与根路径（/health/live、/health/ready、/api/v1/health、 /、/health/storage）没有用户、没业务动作，每秒被 K8s/LB 探针刷一次只会灌进无意义噪音。命中排除集即跳过审计行； 诊断日志（method/path/status/ms 走 stderr）照常打，对容器 运维排错仍有用。 * settings.audit_excluded_paths: list[str] 默认覆盖 5 条 基础设施路径，env AUDIT_EXCLUDED_PATHS 用逗号分隔 （pydantic NoDecode + field_validator 兼容 str/list） * main.py 模块级 _AUDIT_EXCLUDED = frozenset(...)， access_log 的 success/exception 两条审计行各加守卫 诊断无条件打 * 测试用 _AccessLogReplica 复刻 access_log 契约（不 import 真实 main.py），新增 4 个 case：排除根路径、排除 /health/live、 不排除路径照写审计、自定义排除集 顺带 schedule 模块：ExecutionResult 与 context 已迁到 schedule.domain.*（execution.py / orchestrator.py / scheduler.py / worker.py），调用点跟进；schedule 自身 18 个测试在改前改后均通过。
348 lines
12 KiB
Python
348 lines
12 KiB
Python
"""审计日志测试。
|
||
|
||
覆盖：
|
||
* 每天一个 ``audit-YYYY-MM-DD.log`` 文件且写入至少一行；
|
||
* 日志行包含 user_id / method / path / status；
|
||
* access_log 在 success 与 exception（500）两条路径都写审计行；
|
||
* 未登录（无 cookie 无 header）时 user_id 记 ``-``；
|
||
* Authorization: Bearer 头能解析出 user_id；
|
||
* 启动时按 mtime 清理超过保留天数的旧 ``audit-*.log``；
|
||
* ``configure_audit_logging`` 幂等。
|
||
|
||
测试不 import main.py、不触达 MySQL / 任何真实业务逻辑：用一个带空路由的
|
||
临时 FastAPI app，挂一个复制 access_log 审计契约的 ``BaseHTTPMiddleware``
|
||
（``_AccessLogReplica``），验证 sink 与契约行为。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import os
|
||
from collections.abc import Awaitable, Callable
|
||
from datetime import UTC, datetime
|
||
from pathlib import Path
|
||
|
||
import pytest
|
||
from common.auth.jwt import JwtError, issue_jwt, verify_jwt_token
|
||
from fastapi import FastAPI, Request
|
||
from fastapi.responses import Response
|
||
from fastapi.testclient import TestClient
|
||
from loguru import logger
|
||
from starlette.middleware.base import BaseHTTPMiddleware
|
||
|
||
from backend import audit
|
||
|
||
|
||
def _audit_user_id(request: Request) -> str:
|
||
"""与 main.py 的 _audit_user_id 契约一致：cookie/Bearer 头解 JWT，失败记 '-'。"""
|
||
token = request.cookies.get("access_token")
|
||
if not token:
|
||
auth = request.headers.get("authorization", "")
|
||
if auth.lower().startswith("bearer "):
|
||
token = auth[7:].strip()
|
||
if not token:
|
||
return "-"
|
||
try:
|
||
payload = verify_jwt_token(token)
|
||
except (JwtError, Exception): # 任何异常都吞
|
||
return "-"
|
||
sub = payload.get("sub")
|
||
return sub or "-"
|
||
|
||
|
||
class _AccessLogReplica(BaseHTTPMiddleware):
|
||
"""复制 main.py access_log 写审计行的契约（不 import 真实 main.py）。
|
||
|
||
测试只覆盖 access_log 的审计行为（success / exception 两条路径 +
|
||
user_id 解析 + 排除集守卫），避免触发 main.py 的 lifespan（MySQL /
|
||
路由初始化）。``excluded_paths`` 对应 main.py 的 ``_AUDIT_EXCLUDED``：
|
||
精确路径命中即跳过审计行（诊断 stderr 照常）。未来 main.py 改
|
||
access_log 字段时，这里同步改即可，测试不会假阳/假阴。
|
||
"""
|
||
|
||
def __init__(self, app, excluded_paths: frozenset[str] = frozenset()):
|
||
super().__init__(app)
|
||
self.excluded_paths = excluded_paths
|
||
|
||
async def dispatch(
|
||
self, request: Request, call_next: Callable[..., Awaitable[Response]]
|
||
) -> Response:
|
||
skip_audit = request.url.path in self.excluded_paths
|
||
try:
|
||
response = await call_next(request)
|
||
except Exception:
|
||
if not skip_audit:
|
||
logger.bind(
|
||
user_id=_audit_user_id(request),
|
||
method=request.method,
|
||
path=request.url.path,
|
||
status=500,
|
||
).info("audit")
|
||
raise
|
||
if not skip_audit:
|
||
logger.bind(
|
||
user_id=_audit_user_id(request),
|
||
method=request.method,
|
||
path=request.url.path,
|
||
status=response.status_code,
|
||
).info("audit")
|
||
return response
|
||
|
||
|
||
@pytest.fixture(autouse=True)
|
||
def _reset_audit_logging():
|
||
"""每个用例之间重置审计模块的幂等标志并卸掉上次挂上的审计 sink。
|
||
|
||
loguru 的全局 logger 会跨用例留存 sink，若不清理，后面的用例会往已
|
||
删除的临时目录继续写，且 ``_CONFIGURED`` 会让后续 configure 变空操作。
|
||
"""
|
||
audit._CONFIGURED = False
|
||
yield
|
||
if audit._HANDLER_ID is not None:
|
||
logger.remove(audit._HANDLER_ID)
|
||
audit._HANDLER_ID = None
|
||
audit._CONFIGURED = False
|
||
|
||
|
||
def _build_client(
|
||
log_dir: str,
|
||
*,
|
||
raise_server_exceptions: bool = True,
|
||
excluded_paths: frozenset[str] = frozenset(),
|
||
) -> TestClient:
|
||
"""配置审计日志并返回挂上 access_log 契约复刻中间件的测试 app 客户端。
|
||
|
||
只注册少量测试路由，不碰数据库与业务逻辑；中间件在路由之后注册，与
|
||
main.py 的 access_log 行为一致。``/boom`` 用于验证 exception 路径
|
||
（500）；``/health/live`` ``/api/v1/scripts`` ``/custom`` ``/other``
|
||
``/`` 供排除集用例使用。
|
||
"""
|
||
audit.configure_audit_logging(log_dir, retention_days=30)
|
||
|
||
app = FastAPI()
|
||
|
||
@app.get("/x")
|
||
def x() -> dict:
|
||
return {"ok": True}
|
||
|
||
@app.get("/boom")
|
||
def boom() -> dict:
|
||
raise RuntimeError("boom")
|
||
|
||
@app.get("/health/live")
|
||
def health_live() -> dict:
|
||
return {"ok": True}
|
||
|
||
@app.get("/api/v1/scripts")
|
||
def scripts() -> dict:
|
||
return {"ok": True}
|
||
|
||
@app.get("/custom")
|
||
def custom() -> dict:
|
||
return {"ok": True}
|
||
|
||
@app.get("/other")
|
||
def other() -> dict:
|
||
return {"ok": True}
|
||
|
||
@app.get("/")
|
||
def root() -> dict:
|
||
return {"ok": True}
|
||
|
||
app.add_middleware(_AccessLogReplica, excluded_paths=excluded_paths)
|
||
return TestClient(app, raise_server_exceptions=raise_server_exceptions)
|
||
|
||
|
||
def _local_today() -> str:
|
||
return datetime.now(UTC).astimezone().strftime("%Y-%m-%d")
|
||
|
||
|
||
def _today_file(log_dir: Path) -> Path:
|
||
return log_dir / f"audit-{_local_today()}.log"
|
||
|
||
|
||
def _audit_lines(log_dir: Path) -> list[str]:
|
||
"""等 enqueue 队列落盘后返回今天审计文件的全部非空行。"""
|
||
logger.complete()
|
||
path = _today_file(log_dir)
|
||
if not path.exists():
|
||
return []
|
||
return [line for line in path.read_text(encoding="utf-8").splitlines() if line]
|
||
|
||
|
||
def test_audit_log_file_is_created_with_date_suffix(tmp_path: Path) -> None:
|
||
client = _build_client(str(tmp_path))
|
||
client.get("/x")
|
||
|
||
lines = _audit_lines(tmp_path)
|
||
assert _today_file(tmp_path).exists()
|
||
assert len(lines) >= 1
|
||
|
||
|
||
def test_audit_log_line_contains_user_method_path_status(tmp_path: Path) -> None:
|
||
client = _build_client(str(tmp_path))
|
||
user_id = "01USER12345678901234567890"
|
||
token = issue_jwt(user_id)
|
||
client.cookies.set("access_token", token)
|
||
response = client.get("/x")
|
||
assert response.status_code == 200
|
||
|
||
lines = _audit_lines(tmp_path)
|
||
assert lines
|
||
assert f"| {user_id} | GET /x -> 200" in lines[0]
|
||
|
||
|
||
def test_access_log_writes_audit_line_on_success(tmp_path: Path) -> None:
|
||
"""access_log success 路径写审计行：user_id / status 正确。"""
|
||
client = _build_client(str(tmp_path))
|
||
user_id = "01SUCCESS0000000000000000"
|
||
token = issue_jwt(user_id)
|
||
client.cookies.set("access_token", token)
|
||
response = client.get("/x")
|
||
assert response.status_code == 200
|
||
|
||
lines = _audit_lines(tmp_path)
|
||
assert lines
|
||
assert f"| {user_id} | GET /x -> 200" in lines[0]
|
||
|
||
|
||
def test_access_log_writes_audit_line_on_5xx(tmp_path: Path) -> None:
|
||
"""access_log exception 路径也写审计行：status 记 500。"""
|
||
client = _build_client(str(tmp_path), raise_server_exceptions=False)
|
||
response = client.get("/boom")
|
||
assert response.status_code == 500
|
||
|
||
lines = _audit_lines(tmp_path)
|
||
assert lines
|
||
assert "GET /boom -> 500" in lines[0]
|
||
|
||
|
||
def test_access_log_writes_dash_user_when_no_jwt(tmp_path: Path) -> None:
|
||
"""无 cookie 无 header 时，审计行 user_id 列记 '-'。"""
|
||
client = _build_client(str(tmp_path))
|
||
response = client.get("/x")
|
||
assert response.status_code == 200
|
||
|
||
lines = _audit_lines(tmp_path)
|
||
assert lines
|
||
assert "| - | GET /x -> 200" in lines[0]
|
||
|
||
|
||
def test_audit_log_bearer_header_resolves_user(tmp_path: Path) -> None:
|
||
"""Authorization: Bearer 头同样能解析出 user_id。"""
|
||
client = _build_client(str(tmp_path))
|
||
user_id = "01BEARER000000000000000001"
|
||
token = issue_jwt(user_id)
|
||
response = client.get("/x", headers={"Authorization": f"Bearer {token}"})
|
||
assert response.status_code == 200
|
||
|
||
lines = _audit_lines(tmp_path)
|
||
assert lines
|
||
assert f"| {user_id} | GET /x -> 200" in lines[0]
|
||
|
||
|
||
def test_access_log_skips_audit_for_excluded_path(tmp_path: Path) -> None:
|
||
"""命中排除集（/health/live）：审计行不写，诊断 stderr 照常。
|
||
|
||
stderr 走 loguru default sink，难以用 caplog 抓取，这里直接断言
|
||
审计文件无新行即可。
|
||
"""
|
||
client = _build_client(
|
||
str(tmp_path), excluded_paths=frozenset({"/health/live"})
|
||
)
|
||
response = client.get("/health/live")
|
||
assert response.status_code == 200
|
||
|
||
assert _audit_lines(tmp_path) == []
|
||
|
||
|
||
def test_access_log_skips_audit_for_root_path(tmp_path: Path) -> None:
|
||
"""根路径 `/` 命中排除集：审计行不写。"""
|
||
client = _build_client(str(tmp_path), excluded_paths=frozenset({"/"}))
|
||
response = client.get("/")
|
||
assert response.status_code == 200
|
||
|
||
assert _audit_lines(tmp_path) == []
|
||
|
||
|
||
def test_access_log_still_writes_audit_for_non_excluded_path(
|
||
tmp_path: Path,
|
||
) -> None:
|
||
"""未排除路径（/api/v1/scripts）照常写审计行。"""
|
||
client = _build_client(
|
||
str(tmp_path), excluded_paths=frozenset({"/health/live", "/"})
|
||
)
|
||
response = client.get("/api/v1/scripts")
|
||
assert response.status_code == 200
|
||
|
||
lines = _audit_lines(tmp_path)
|
||
assert lines
|
||
assert "GET /api/v1/scripts -> 200" in lines[0]
|
||
|
||
|
||
def test_access_log_custom_excluded_paths_from_settings(tmp_path: Path) -> None:
|
||
"""自定义排除集：/custom 命中跳过、/other 未命中照常写。"""
|
||
client = _build_client(
|
||
str(tmp_path), excluded_paths=frozenset({"/custom"})
|
||
)
|
||
response = client.get("/custom")
|
||
assert response.status_code == 200
|
||
assert _audit_lines(tmp_path) == []
|
||
|
||
response = client.get("/other")
|
||
assert response.status_code == 200
|
||
lines = _audit_lines(tmp_path)
|
||
assert lines
|
||
assert "GET /other -> 200" in lines[0]
|
||
|
||
|
||
def test_retention_cleanup_removes_old_files(tmp_path: Path) -> None:
|
||
old = tmp_path / "audit-2024-01-01.log"
|
||
recent = tmp_path / "audit-2024-06-01.log"
|
||
old.write_text("old\n", encoding="utf-8")
|
||
recent.write_text("recent\n", encoding="utf-8")
|
||
|
||
old_mtime = datetime.now(UTC).timestamp() - 60 * 86_400 # 60 天前
|
||
recent_mtime = datetime.now(UTC).timestamp() - 5 * 86_400 # 5 天前
|
||
os.utime(old, (old_mtime, old_mtime))
|
||
os.utime(recent, (recent_mtime, recent_mtime))
|
||
|
||
audit.configure_audit_logging(str(tmp_path), retention_days=30)
|
||
|
||
assert not old.exists(), "超过保留期的旧审计文件应被启动清理删除"
|
||
assert recent.exists(), "保留期内(5 天前)的文件应保留"
|
||
|
||
|
||
def test_retention_cleanup_disabled_when_zero(tmp_path: Path) -> None:
|
||
"""AUDIT_LOG_RETENTION_DAYS=0 关闭清理：任何旧文件都不删除。"""
|
||
old = tmp_path / "audit-2024-01-01.log"
|
||
old.write_text("old\n", encoding="utf-8")
|
||
old_mtime = datetime.now(UTC).timestamp() - 400 * 86_400 # 400 天前
|
||
os.utime(old, (old_mtime, old_mtime))
|
||
|
||
audit.configure_audit_logging(str(tmp_path), retention_days=0)
|
||
|
||
assert old.exists(), "retention_days=0 时应跳过清理，保留所有旧文件"
|
||
|
||
|
||
def test_configure_audit_logging_is_idempotent(tmp_path: Path) -> None:
|
||
"""多次调用只有首次生效：日志只写到第一次给的目录。"""
|
||
first_dir = tmp_path / "first"
|
||
second_dir = tmp_path / "second"
|
||
|
||
audit.configure_audit_logging(str(first_dir), retention_days=30)
|
||
audit.configure_audit_logging(str(second_dir), retention_days=30)
|
||
|
||
app = FastAPI()
|
||
|
||
@app.get("/x")
|
||
def x() -> dict:
|
||
return {"ok": True}
|
||
|
||
app.add_middleware(_AccessLogReplica)
|
||
client = TestClient(app)
|
||
client.get("/x")
|
||
|
||
logger.complete()
|
||
assert (first_dir / f"audit-{_local_today()}.log").exists()
|
||
assert not (second_dir / f"audit-{_local_today()}.log").exists()
|