main:新增中间件测试用例

变更内容:
- 为路径规范化函数添加单元测试,验证 /jobs 等路径的行为。
- 为指标中间件编写测试,包括健康检查端点跳过和普通端点的指标记录。
- 检查任务路径规范化逻辑并验证规范化后的路径是否正确。
This commit is contained in:
2026-02-02 17:12:07 +08:00
parent 87ed8c071c
commit 9b6642635b

97
tests/test_middleware.py Normal file
View File

@@ -0,0 +1,97 @@
"""中间件测试"""
import pytest
from unittest.mock import patch, MagicMock
from fastapi.testclient import TestClient
from src.functional_scaffold.main import app, normalize_path
class TestNormalizePath:
"""测试路径规范化函数"""
def test_normalize_jobs_path(self):
"""测试 /jobs/{job_id} 路径规范化"""
assert normalize_path("/jobs/a1b2c3d4e5f6") == "/jobs/{job_id}"
assert normalize_path("/jobs/123456789012") == "/jobs/{job_id}"
assert normalize_path("/jobs/xyz") == "/jobs/{job_id}"
def test_normalize_other_paths(self):
"""测试其他路径保持不变"""
assert normalize_path("/invoke") == "/invoke"
assert normalize_path("/healthz") == "/healthz"
assert normalize_path("/readyz") == "/readyz"
assert normalize_path("/metrics") == "/metrics"
assert normalize_path("/docs") == "/docs"
def test_normalize_jobs_root(self):
"""测试 /jobs 根路径"""
assert normalize_path("/jobs") == "/jobs"
class TestMetricsMiddleware:
"""测试指标中间件"""
@patch("src.functional_scaffold.main.incr")
@patch("src.functional_scaffold.main.observe")
@patch("src.functional_scaffold.main.gauge_incr")
@patch("src.functional_scaffold.main.gauge_decr")
def test_skip_health_endpoints(self, mock_gauge_decr, mock_gauge_incr, mock_observe, mock_incr):
"""测试跳过健康检查端点"""
client = TestClient(app)
# 访问健康检查端点
client.get("/healthz")
client.get("/readyz")
client.get("/metrics")
# 验证没有记录指标
mock_incr.assert_not_called()
mock_observe.assert_not_called()
mock_gauge_incr.assert_not_called()
mock_gauge_decr.assert_not_called()
@patch("src.functional_scaffold.main.incr")
@patch("src.functional_scaffold.main.observe")
@patch("src.functional_scaffold.main.gauge_incr")
@patch("src.functional_scaffold.main.gauge_decr")
def test_record_normal_endpoints(self, mock_gauge_decr, mock_gauge_incr, mock_observe, mock_incr):
"""测试记录普通端点"""
client = TestClient(app)
# 访问普通端点
client.post("/invoke", json={"number": 17})
# 验证记录了指标
mock_gauge_incr.assert_called_once()
mock_gauge_decr.assert_called_once()
mock_incr.assert_called_once()
mock_observe.assert_called_once()
# 验证使用了正确的端点路径
incr_call_args = mock_incr.call_args
assert incr_call_args[0][1]["endpoint"] == "/invoke"
@patch("src.functional_scaffold.main.incr")
@patch("src.functional_scaffold.main.observe")
@patch("src.functional_scaffold.main.gauge_incr")
@patch("src.functional_scaffold.main.gauge_decr")
@patch("src.functional_scaffold.core.job_manager.get_job_manager")
def test_normalize_job_path(self, mock_get_manager, mock_gauge_decr, mock_gauge_incr, mock_observe, mock_incr):
"""测试规范化任务路径"""
# Mock job manager
mock_manager = MagicMock()
mock_manager.get_job.return_value = None
mock_get_manager.return_value = mock_manager
client = TestClient(app)
# 访问任务端点(会返回 404但中间件应该记录指标
client.get("/jobs/a1b2c3d4e5f6")
# 验证记录了指标
mock_incr.assert_called_once()
# 验证使用了规范化后的路径
incr_call_args = mock_incr.call_args
assert incr_call_args[0][1]["endpoint"] == "/jobs/{job_id}"