229 lines
6.7 KiB
Python
229 lines
6.7 KiB
Python
"""监控模块测试"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
from minenasai.core.monitoring import (
|
|
ComponentHealth,
|
|
HealthChecker,
|
|
HealthStatus,
|
|
SystemMetrics,
|
|
get_health_checker,
|
|
get_metrics,
|
|
)
|
|
|
|
|
|
class TestHealthStatus:
|
|
"""HealthStatus 测试"""
|
|
|
|
def test_health_status_values(self):
|
|
"""测试健康状态值"""
|
|
assert HealthStatus.HEALTHY.value == "healthy"
|
|
assert HealthStatus.DEGRADED.value == "degraded"
|
|
assert HealthStatus.UNHEALTHY.value == "unhealthy"
|
|
|
|
|
|
class TestSystemMetrics:
|
|
"""SystemMetrics 测试"""
|
|
|
|
def test_initial_metrics(self):
|
|
"""测试初始指标"""
|
|
metrics = SystemMetrics()
|
|
|
|
assert metrics.total_requests == 0
|
|
assert metrics.successful_requests == 0
|
|
assert metrics.failed_requests == 0
|
|
assert metrics.active_connections == 0
|
|
|
|
def test_record_request(self):
|
|
"""测试记录请求"""
|
|
metrics = SystemMetrics()
|
|
|
|
metrics.record_request(100.0, success=True)
|
|
metrics.record_request(200.0, success=True)
|
|
metrics.record_request(50.0, success=False)
|
|
|
|
assert metrics.total_requests == 3
|
|
assert metrics.successful_requests == 2
|
|
assert metrics.failed_requests == 1
|
|
assert metrics.min_response_time_ms == 50.0
|
|
assert metrics.max_response_time_ms == 200.0
|
|
|
|
def test_avg_response_time(self):
|
|
"""测试平均响应时间"""
|
|
metrics = SystemMetrics()
|
|
|
|
metrics.record_request(100.0, success=True)
|
|
metrics.record_request(200.0, success=True)
|
|
|
|
assert metrics.avg_response_time_ms == 150.0
|
|
|
|
def test_success_rate(self):
|
|
"""测试成功率"""
|
|
metrics = SystemMetrics()
|
|
|
|
metrics.record_request(100.0, success=True)
|
|
metrics.record_request(100.0, success=True)
|
|
metrics.record_request(100.0, success=False)
|
|
|
|
assert metrics.success_rate == pytest.approx(2 / 3)
|
|
|
|
def test_record_error(self):
|
|
"""测试记录错误"""
|
|
metrics = SystemMetrics()
|
|
|
|
metrics.record_error("ValueError")
|
|
metrics.record_error("ValueError")
|
|
metrics.record_error("TypeError")
|
|
|
|
assert metrics.errors_by_type["ValueError"] == 2
|
|
assert metrics.errors_by_type["TypeError"] == 1
|
|
|
|
def test_to_dict(self):
|
|
"""测试转换为字典"""
|
|
metrics = SystemMetrics()
|
|
metrics.record_request(100.0, success=True)
|
|
|
|
result = metrics.to_dict()
|
|
|
|
assert "requests" in result
|
|
assert "response_time_ms" in result
|
|
assert "connections" in result
|
|
assert "uptime_seconds" in result
|
|
|
|
|
|
class TestComponentHealth:
|
|
"""ComponentHealth 测试"""
|
|
|
|
def test_component_health(self):
|
|
"""测试组件健康状态"""
|
|
health = ComponentHealth(
|
|
name="test",
|
|
status=HealthStatus.HEALTHY,
|
|
message="OK",
|
|
latency_ms=10.0,
|
|
)
|
|
|
|
assert health.name == "test"
|
|
assert health.status == HealthStatus.HEALTHY
|
|
assert health.message == "OK"
|
|
assert health.latency_ms == 10.0
|
|
|
|
|
|
class TestHealthChecker:
|
|
"""HealthChecker 测试"""
|
|
|
|
def setup_method(self):
|
|
"""初始化"""
|
|
self.checker = HealthChecker()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_register_and_check(self):
|
|
"""测试注册和检查"""
|
|
|
|
async def check_ok() -> ComponentHealth:
|
|
return ComponentHealth(
|
|
name="test",
|
|
status=HealthStatus.HEALTHY,
|
|
message="OK",
|
|
)
|
|
|
|
self.checker.register("test", check_ok)
|
|
result = await self.checker.check_component("test")
|
|
|
|
assert result.status == HealthStatus.HEALTHY
|
|
assert result.latency_ms > 0
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_check_unknown_component(self):
|
|
"""测试检查未知组件"""
|
|
result = await self.checker.check_component("unknown")
|
|
|
|
assert result.status == HealthStatus.UNHEALTHY
|
|
assert "未知组件" in result.message
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_check_all(self):
|
|
"""测试检查所有组件"""
|
|
|
|
async def check_a() -> ComponentHealth:
|
|
return ComponentHealth(name="a", status=HealthStatus.HEALTHY)
|
|
|
|
async def check_b() -> ComponentHealth:
|
|
return ComponentHealth(name="b", status=HealthStatus.HEALTHY)
|
|
|
|
self.checker.register("a", check_a)
|
|
self.checker.register("b", check_b)
|
|
|
|
results = await self.checker.check_all()
|
|
|
|
assert len(results) == 2
|
|
assert "a" in results
|
|
assert "b" in results
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_overall_status_healthy(self):
|
|
"""测试总体状态 - 健康"""
|
|
|
|
async def check_ok() -> ComponentHealth:
|
|
return ComponentHealth(name="test", status=HealthStatus.HEALTHY)
|
|
|
|
self.checker.register("test", check_ok)
|
|
await self.checker.check_all()
|
|
|
|
assert self.checker.get_overall_status() == HealthStatus.HEALTHY
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_overall_status_degraded(self):
|
|
"""测试总体状态 - 降级"""
|
|
|
|
async def check_degraded() -> ComponentHealth:
|
|
return ComponentHealth(name="test", status=HealthStatus.DEGRADED)
|
|
|
|
self.checker.register("test", check_degraded)
|
|
await self.checker.check_all()
|
|
|
|
assert self.checker.get_overall_status() == HealthStatus.DEGRADED
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_overall_status_unhealthy(self):
|
|
"""测试总体状态 - 不健康"""
|
|
|
|
async def check_unhealthy() -> ComponentHealth:
|
|
return ComponentHealth(name="test", status=HealthStatus.UNHEALTHY)
|
|
|
|
self.checker.register("test", check_unhealthy)
|
|
await self.checker.check_all()
|
|
|
|
assert self.checker.get_overall_status() == HealthStatus.UNHEALTHY
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_check_timeout(self):
|
|
"""测试检查超时"""
|
|
import asyncio
|
|
|
|
async def slow_check() -> ComponentHealth:
|
|
await asyncio.sleep(10) # 超过5秒超时
|
|
return ComponentHealth(name="slow", status=HealthStatus.HEALTHY)
|
|
|
|
self.checker.register("slow", slow_check)
|
|
result = await self.checker.check_component("slow")
|
|
|
|
assert result.status == HealthStatus.UNHEALTHY
|
|
assert "超时" in result.message
|
|
|
|
|
|
class TestGlobalInstances:
|
|
"""全局实例测试"""
|
|
|
|
def test_get_metrics(self):
|
|
"""测试获取全局指标"""
|
|
metrics = get_metrics()
|
|
assert isinstance(metrics, SystemMetrics)
|
|
|
|
def test_get_health_checker(self):
|
|
"""测试获取健康检查器"""
|
|
checker = get_health_checker()
|
|
assert isinstance(checker, HealthChecker)
|