"""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()