121 lines
4.8 KiB
Python
121 lines
4.8 KiB
Python
"""middleware/operation_log.py 操作日志中间件单元测试：日志写入延迟到响应后台任务，不阻塞响应。"""
|
|
import asyncio
|
|
import types
|
|
|
|
import pytest
|
|
from starlette.requests import Request
|
|
from starlette.responses import JSONResponse
|
|
|
|
import middleware.operation_log as op_middleware
|
|
|
|
|
|
def _make_request(method='PUT', path='/api/v1/demo', user_id=1):
|
|
scope = {
|
|
'type': 'http',
|
|
'method': method,
|
|
'path': path,
|
|
'headers': [(b'user-agent', b'test-agent')],
|
|
'query_string': b'',
|
|
'route': types.SimpleNamespace(summary='测试操作'),
|
|
}
|
|
request = Request(scope)
|
|
request.state.user_id = user_id
|
|
return request
|
|
|
|
|
|
async def _run_middleware(request):
|
|
async def call_next(request):
|
|
return JSONResponse(content={'code': 0, 'msg': 'ok'})
|
|
return await op_middleware.operation_log_middleware(request, call_next)
|
|
|
|
|
|
def _patch_to_thread(monkeypatch, submitted):
|
|
"""记录提交到线程池的函数并同步执行，隔离真实线程调度。"""
|
|
async def fake_to_thread(func, *args, **kwargs):
|
|
submitted.append(func)
|
|
return func(*args, **kwargs)
|
|
monkeypatch.setattr(asyncio, 'to_thread', fake_to_thread)
|
|
|
|
|
|
def test_operation_log_deferred_to_response_background(monkeypatch):
|
|
"""日志写入挂到 response.background，响应返回后异步落库，不阻塞响应。"""
|
|
import modules.operation_log.service as op_service
|
|
|
|
captured = {}
|
|
monkeypatch.setattr(op_service, 'record_operation_log',
|
|
lambda request, title, **kw: captured.update(kw) or None)
|
|
|
|
resp = asyncio.run(_run_middleware(_make_request()))
|
|
|
|
assert resp.status_code == 200
|
|
# 响应上挂载后台任务（而非在响应返回前同步执行日志写入）
|
|
assert resp.background is not None
|
|
# 后台任务执行时调用 record_operation_log 写入日志
|
|
asyncio.run(resp.background())
|
|
assert captured['status'] == 0
|
|
assert captured['consume_time'] >= 0
|
|
assert 'ok' in captured['result'] # result 采样非空
|
|
|
|
|
|
def test_operation_log_records_failure_status(monkeypatch):
|
|
"""业务失败（code=1）在后台任务中记录 error 与 status=1。"""
|
|
import modules.operation_log.service as op_service
|
|
|
|
captured = {}
|
|
monkeypatch.setattr(op_service, 'record_operation_log',
|
|
lambda request, title, **kw: captured.update(kw) or None)
|
|
|
|
async def call_next(request):
|
|
return JSONResponse(content={'code': 1, 'msg': '权限不足'})
|
|
|
|
resp = asyncio.run(op_middleware.operation_log_middleware(_make_request(), call_next))
|
|
asyncio.run(resp.background())
|
|
assert captured['status'] == 1
|
|
assert captured['error'] == '权限不足'
|
|
|
|
|
|
def test_operation_log_exception_path_writes_via_thread_pool(monkeypatch):
|
|
"""请求异常无响应可挂后台任务：走线程池同步写入（低频异常路径）。"""
|
|
import modules.operation_log.service as op_service
|
|
|
|
submitted = []
|
|
_patch_to_thread(monkeypatch, submitted)
|
|
captured = {}
|
|
monkeypatch.setattr(op_service, 'record_operation_log',
|
|
lambda request, title, **kw: captured.update(kw) or None)
|
|
|
|
async def call_next(request):
|
|
raise RuntimeError('boom')
|
|
|
|
with pytest.raises(RuntimeError):
|
|
asyncio.run(op_middleware.operation_log_middleware(_make_request(), call_next))
|
|
assert submitted == [op_service.record_operation_log]
|
|
assert captured['status'] == 1
|
|
assert captured['error'] == 'boom'
|
|
|
|
|
|
def test_operation_log_skipped_for_get_request(monkeypatch):
|
|
import modules.operation_log.service as op_service
|
|
|
|
submitted = []
|
|
_patch_to_thread(monkeypatch, submitted)
|
|
monkeypatch.setattr(op_service, 'record_operation_log', lambda *a, **k: None)
|
|
|
|
resp = asyncio.run(_run_middleware(_make_request(method='GET')))
|
|
|
|
assert resp.status_code == 200
|
|
assert submitted == []
|
|
|
|
|
|
def test_exclusion_uses_path_boundary_not_substring():
|
|
"""排除匹配按路径段边界，避免子串误伤其他模块路由。"""
|
|
# /api/v1/index/userInfo 是个人中心写操作，不应被子串 '/api/v1/index/user' 误排除
|
|
assert op_middleware._should_log(_make_request(method='PUT', path='/api/v1/index/userInfo'))
|
|
# 挂载前缀下的写操作应排除
|
|
assert not op_middleware._should_log(_make_request(method='POST', path='/api/v1/login'))
|
|
assert not op_middleware._should_log(_make_request(method='POST', path='/api/v1/upload/uploadFile'))
|
|
assert not op_middleware._should_log(_make_request(method='DELETE', path='/api/v1/operationlog/delete/1'))
|
|
assert not op_middleware._should_log(_make_request(method='DELETE', path='/api/v1/loginlog/batchDelete'))
|
|
# 业务模块写操作应记录
|
|
assert op_middleware._should_log(_make_request(method='PUT', path='/api/v1/user/update/1'))
|