Roog (顾新培)
88cfe91c56
变更内容： - 删除 `src` 子目录，将模块引用路径从 `src.functional_scaffold` 更新为 `functional_scaffold`。 - 修改相关代码、文档、测试用例及配置文件中的路径引用，包括 `README.md`、`Dockerfile`、`uvicorn` 启动命令等。 - 优化项目目录结构，提升代码维护性和可读性。
274 lines
9.1 KiB
Python
274 lines
9.1 KiB
Python
"""metrics_unified 模块单元测试"""
|
||
|
||
import pytest
|
||
from unittest.mock import MagicMock, patch
|
||
|
||
|
||
class TestMetricsManager:
|
||
"""MetricsManager 类测试"""
|
||
|
||
@pytest.fixture
|
||
def mock_redis(self):
|
||
"""模拟 Redis 客户端"""
|
||
with patch("redis.Redis") as mock:
|
||
mock_instance = MagicMock()
|
||
mock_instance.ping.return_value = True
|
||
mock_instance.hincrbyfloat.return_value = 1.0
|
||
mock_instance.hset.return_value = True
|
||
mock_instance.hgetall.return_value = {}
|
||
mock_instance.hget.return_value = "0"
|
||
mock_instance.keys.return_value = []
|
||
mock_instance.pipeline.return_value = MagicMock()
|
||
mock.return_value = mock_instance
|
||
yield mock_instance
|
||
|
||
@pytest.fixture
|
||
def manager(self, mock_redis):
|
||
"""创建测试用的 MetricsManager"""
|
||
from functional_scaffold.core.metrics_unified import (
|
||
MetricsManager,
|
||
reset_metrics_manager,
|
||
)
|
||
|
||
reset_metrics_manager()
|
||
manager = MetricsManager()
|
||
return manager
|
||
|
||
def test_init_loads_default_config(self, manager):
|
||
"""测试初始化加载默认配置"""
|
||
assert manager.config is not None
|
||
assert "builtin_metrics" in manager.config or len(manager.metrics_definitions) > 0
|
||
|
||
def test_metrics_definitions_registered(self, manager):
|
||
"""测试指标定义已注册"""
|
||
assert "http_requests_total" in manager.metrics_definitions
|
||
assert "http_request_duration_seconds" in manager.metrics_definitions
|
||
assert "algorithm_executions_total" in manager.metrics_definitions
|
||
|
||
def test_incr_counter(self, manager, mock_redis):
|
||
"""测试计数器增加"""
|
||
manager.incr("http_requests_total", {"method": "GET", "endpoint": "/", "status": "success"})
|
||
mock_redis.hincrbyfloat.assert_called()
|
||
|
||
def test_incr_with_invalid_metric_type(self, manager, mock_redis):
|
||
"""测试对非计数器类型调用 incr"""
|
||
# http_request_duration_seconds 是 histogram 类型
|
||
manager.incr("http_request_duration_seconds", {})
|
||
# 不应该调用 Redis（因为类型不匹配）
|
||
# 验证没有调用 hincrbyfloat（或者调用次数没有增加）
|
||
|
||
def test_set_gauge(self, manager, mock_redis):
|
||
"""测试设置仪表盘"""
|
||
manager.set("http_requests_in_progress", {}, 5)
|
||
mock_redis.hset.assert_called()
|
||
|
||
def test_gauge_incr(self, manager, mock_redis):
|
||
"""测试增加仪表盘"""
|
||
manager.gauge_incr("http_requests_in_progress", {}, 1)
|
||
mock_redis.hincrbyfloat.assert_called()
|
||
|
||
def test_gauge_decr(self, manager, mock_redis):
|
||
"""测试减少仪表盘"""
|
||
manager.gauge_decr("http_requests_in_progress", {}, 1)
|
||
mock_redis.hincrbyfloat.assert_called()
|
||
|
||
def test_observe_histogram(self, manager, mock_redis):
|
||
"""测试直方图观测"""
|
||
mock_pipeline = MagicMock()
|
||
mock_redis.pipeline.return_value = mock_pipeline
|
||
|
||
manager.observe("http_request_duration_seconds", {"method": "GET", "endpoint": "/"}, 0.05)
|
||
|
||
mock_redis.pipeline.assert_called()
|
||
mock_pipeline.execute.assert_called()
|
||
|
||
def test_labels_to_key(self, manager):
|
||
"""测试标签转换为 key"""
|
||
labels = {"method": "GET", "endpoint": "/api"}
|
||
key = manager._labels_to_key(labels)
|
||
assert "method=GET" in key
|
||
assert "endpoint=/api" in key
|
||
|
||
def test_labels_to_key_empty(self, manager):
|
||
"""测试空标签转换"""
|
||
key = manager._labels_to_key(None)
|
||
assert key == ""
|
||
|
||
key = manager._labels_to_key({})
|
||
assert key == ""
|
||
|
||
def test_is_available(self, manager):
|
||
"""测试 Redis 可用性检查"""
|
||
assert manager.is_available() is True
|
||
|
||
|
||
class TestConvenienceFunctions:
|
||
"""便捷函数测试"""
|
||
|
||
@pytest.fixture(autouse=True)
|
||
def setup(self):
|
||
"""每个测试前重置管理器"""
|
||
from functional_scaffold.core.metrics_unified import reset_metrics_manager
|
||
|
||
reset_metrics_manager()
|
||
|
||
@patch("redis.Redis")
|
||
def test_incr_function(self, mock_redis_class):
|
||
"""测试 incr 便捷函数"""
|
||
mock_instance = MagicMock()
|
||
mock_instance.ping.return_value = True
|
||
mock_redis_class.return_value = mock_instance
|
||
|
||
from functional_scaffold.core.metrics_unified import incr, reset_metrics_manager
|
||
|
||
reset_metrics_manager()
|
||
incr("http_requests_total", {"method": "GET", "endpoint": "/", "status": "success"})
|
||
|
||
mock_instance.hincrbyfloat.assert_called()
|
||
|
||
@patch("redis.Redis")
|
||
def test_set_function(self, mock_redis_class):
|
||
"""测试 set 便捷函数"""
|
||
mock_instance = MagicMock()
|
||
mock_instance.ping.return_value = True
|
||
mock_redis_class.return_value = mock_instance
|
||
|
||
from functional_scaffold.core.metrics_unified import reset_metrics_manager, set
|
||
|
||
reset_metrics_manager()
|
||
set("http_requests_in_progress", {}, 10)
|
||
|
||
mock_instance.hset.assert_called()
|
||
|
||
@patch("redis.Redis")
|
||
def test_observe_function(self, mock_redis_class):
|
||
"""测试 observe 便捷函数"""
|
||
mock_instance = MagicMock()
|
||
mock_instance.ping.return_value = True
|
||
mock_pipeline = MagicMock()
|
||
mock_instance.pipeline.return_value = mock_pipeline
|
||
mock_redis_class.return_value = mock_instance
|
||
|
||
from functional_scaffold.core.metrics_unified import observe, reset_metrics_manager
|
||
|
||
reset_metrics_manager()
|
||
observe("http_request_duration_seconds", {"method": "GET", "endpoint": "/"}, 0.1)
|
||
|
||
mock_instance.pipeline.assert_called()
|
||
|
||
|
||
class TestExport:
|
||
"""导出功能测试"""
|
||
|
||
@patch("redis.Redis")
|
||
def test_export_counter(self, mock_redis_class):
|
||
"""测试导出计数器"""
|
||
mock_instance = MagicMock()
|
||
mock_instance.ping.return_value = True
|
||
mock_instance.hgetall.return_value = {"method=GET,endpoint=/,status=success": "10"}
|
||
mock_redis_class.return_value = mock_instance
|
||
|
||
from functional_scaffold.core.metrics_unified import export, reset_metrics_manager
|
||
|
||
reset_metrics_manager()
|
||
output = export()
|
||
|
||
assert "http_requests_total" in output
|
||
assert "HELP" in output
|
||
assert "TYPE" in output
|
||
|
||
@patch("redis.Redis")
|
||
def test_export_histogram(self, mock_redis_class):
|
||
"""测试导出直方图"""
|
||
mock_instance = MagicMock()
|
||
mock_instance.ping.return_value = True
|
||
mock_instance.hgetall.side_effect = lambda key: (
|
||
{"method=GET,endpoint=/": "5"}
|
||
if "count" in key
|
||
else {"method=GET,endpoint=/": "0.5"}
|
||
if "sum" in key
|
||
else {}
|
||
)
|
||
mock_instance.hget.return_value = "3"
|
||
mock_redis_class.return_value = mock_instance
|
||
|
||
from functional_scaffold.core.metrics_unified import export, reset_metrics_manager
|
||
|
||
reset_metrics_manager()
|
||
output = export()
|
||
|
||
assert "http_request_duration_seconds" in output
|
||
|
||
|
||
class TestEnvVarSubstitution:
|
||
"""环境变量替换测试"""
|
||
|
||
def test_substitute_env_vars(self):
|
||
"""测试环境变量替换"""
|
||
import os
|
||
from functional_scaffold.core.metrics_unified import MetricsManager
|
||
|
||
# 设置测试环境变量
|
||
os.environ["TEST_VAR"] = "test_value"
|
||
|
||
manager = MetricsManager.__new__(MetricsManager)
|
||
result = manager._substitute_env_vars("${TEST_VAR:default}")
|
||
assert result == "test_value"
|
||
|
||
# 测试默认值
|
||
result = manager._substitute_env_vars("${NONEXISTENT_VAR:default_value}")
|
||
assert result == "default_value"
|
||
|
||
# 清理
|
||
del os.environ["TEST_VAR"]
|
||
|
||
|
||
class TestTrackAlgorithmExecution:
|
||
"""track_algorithm_execution 装饰器测试"""
|
||
|
||
@patch("redis.Redis")
|
||
def test_decorator_success(self, mock_redis_class):
|
||
"""测试装饰器成功执行"""
|
||
mock_instance = MagicMock()
|
||
mock_instance.ping.return_value = True
|
||
mock_pipeline = MagicMock()
|
||
mock_instance.pipeline.return_value = mock_pipeline
|
||
mock_redis_class.return_value = mock_instance
|
||
|
||
from functional_scaffold.core.metrics_unified import (
|
||
reset_metrics_manager,
|
||
track_algorithm_execution,
|
||
)
|
||
|
||
reset_metrics_manager()
|
||
|
||
@track_algorithm_execution("test_algo")
|
||
def test_func():
|
||
return "result"
|
||
|
||
result = test_func()
|
||
assert result == "result"
|
||
|
||
@patch("redis.Redis")
|
||
def test_decorator_error(self, mock_redis_class):
|
||
"""测试装饰器错误处理"""
|
||
mock_instance = MagicMock()
|
||
mock_instance.ping.return_value = True
|
||
mock_pipeline = MagicMock()
|
||
mock_instance.pipeline.return_value = mock_pipeline
|
||
mock_redis_class.return_value = mock_instance
|
||
|
||
from functional_scaffold.core.metrics_unified import (
|
||
reset_metrics_manager,
|
||
track_algorithm_execution,
|
||
)
|
||
|
||
reset_metrics_manager()
|
||
|
||
@track_algorithm_execution("test_algo")
|
||
def test_func():
|
||
raise ValueError("test error")
|
||
|
||
with pytest.raises(ValueError):
|
||
test_func()
|