Roog (顾新培)
8ca2f64f7e
变更内容： - 删除 `src` 子目录，将模块引用路径从 `src.functional_scaffold` 更新为 `functional_scaffold`。 - 修改相关代码、文档、测试用例及配置文件中的路径引用，包括 `README.md`、`Dockerfile`、`uvicorn` 启动命令等。 - 优化项目目录结构，提升代码维护性和可读性。
98 lines
3.5 KiB
Python
98 lines
3.5 KiB
Python
"""中间件测试"""
|
||
|
||
import pytest
|
||
from unittest.mock import patch, MagicMock
|
||
from fastapi.testclient import TestClient
|
||
|
||
from 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("functional_scaffold.main.incr")
|
||
@patch("functional_scaffold.main.observe")
|
||
@patch("functional_scaffold.main.gauge_incr")
|
||
@patch("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("functional_scaffold.main.incr")
|
||
@patch("functional_scaffold.main.observe")
|
||
@patch("functional_scaffold.main.gauge_incr")
|
||
@patch("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("functional_scaffold.main.incr")
|
||
@patch("functional_scaffold.main.observe")
|
||
@patch("functional_scaffold.main.gauge_incr")
|
||
@patch("functional_scaffold.main.gauge_decr")
|
||
@patch("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}"
|