main:重构指标系统并切换为 Redis 方案
变更内容： - 重构指标系统实现，支持基于 Redis 的多实例指标管理。 - 替换原有的 Pushgateway 和 Redis Exporter 方案。 - 更新 Prometheus 配置，适配新的指标抓取方式。 - 添加 Redis 指标相关配置和告警规则文件。 - 更新 Dockerfile 和 docker-compose 文件，移除多余服务，精简配置。 - 编写 `metrics_unified.py` 模块及单元测试。 - 修复部分代码中的冗余和格式问题。
This commit is contained in:
273
tests/test_metrics_unified.py
Normal file
273
tests/test_metrics_unified.py
Normal file
@@ -0,0 +1,273 @@
|
||||
"""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 src.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 src.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 src.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 src.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 src.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 src.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 src.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 src.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 src.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 src.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()
|
||||
Reference in New Issue
Block a user