ctms/backend/tests/test_permission_monitoring.py

"""监控测试：权限系统内存监控职责验证。"""

import time

import pytest

from app.core.permission_monitor import (
    CacheMetrics,
    PermissionMonitor,
    get_permission_monitor,
    set_permission_monitor,
)


@pytest.mark.asyncio
async def test_cache_metrics():
    """测试缓存指标"""
    monitor = PermissionMonitor()

    monitor.record_cache_hit()
    monitor.record_cache_hit()
    monitor.record_cache_miss()
    monitor.record_cache_hit()

    metrics = monitor.metrics.cache_metrics
    assert metrics.total_accesses == 4
    assert metrics.cache_hits == 3
    assert metrics.cache_misses == 1
    assert metrics.hit_rate == pytest.approx(75.0, 0.1)
    assert metrics.miss_rate == pytest.approx(25.0, 0.1)


@pytest.mark.asyncio
async def test_cache_metrics_one_hour_window_and_cache_types():
    monitor = PermissionMonitor()
    monitor.record_cache_hit("project_permissions")
    monitor.record_cache_miss("project_permissions")
    monitor.record_cache_hit("member_roles")

    stats = monitor.metrics.cache_metrics.window_stats(3600)

    assert stats["total_accesses"] == 3
    assert stats["hit_rate"] == pytest.approx(66.67, 0.01)
    assert stats["scope"] == "sliding_window"
    assert stats["by_cache"]["project_permissions"]["total_accesses"] == 2
    assert stats["by_cache"]["project_permissions"]["hit_rate"] == 50
    assert stats["by_cache"]["member_roles"]["total_accesses"] == 1
    assert stats["by_cache"]["member_roles"]["hit_rate"] == 100


@pytest.mark.asyncio
async def test_cache_invalidation_tracking():
    """测试缓存失效追踪"""
    monitor = PermissionMonitor()

    monitor.record_cache_invalidation()
    monitor.record_cache_invalidation()
    monitor.record_cache_invalidation()

    assert monitor.metrics.cache_metrics.cache_invalidations == 3


@pytest.mark.asyncio
async def test_slow_check_alert_generation():
    """测试慢权限检查告警生成"""
    monitor = PermissionMonitor()

    monitor.record_slow_check_alert(100)

    alerts = monitor.get_alerts()
    assert len(alerts) == 1
    assert alerts[0]["level"] == "warning"
    assert alerts[0]["type"] == "slow_permission_check"
    assert alerts[0]["data"]["elapsed_ms"] == 100


@pytest.mark.asyncio
async def test_slow_check_alert_ignores_fast_checks():
    """未超过阈值的权限检查不应生成告警"""
    monitor = PermissionMonitor()

    monitor.record_slow_check_alert(10)

    assert monitor.get_alerts() == []


@pytest.mark.asyncio
async def test_error_alert_generation():
    """测试错误告警生成"""
    monitor = PermissionMonitor()

    monitor.record_error_alert(ValueError("test error"))

    alerts = monitor.get_alerts()
    assert len(alerts) == 1
    assert alerts[0]["level"] == "error"
    assert alerts[0]["type"] == "permission_check_error"


@pytest.mark.asyncio
async def test_alert_filtering():
    """测试告警过滤"""
    monitor = PermissionMonitor()

    monitor.record_slow_check_alert(100)
    monitor.record_error_alert(ValueError("test error"))

    warning_alerts = monitor.get_alerts(level="warning")
    assert len(warning_alerts) == 1
    assert all(alert["level"] == "warning" for alert in warning_alerts)

    error_alerts = monitor.get_alerts(level="error")
    assert len(error_alerts) == 1
    assert all(alert["level"] == "error" for alert in error_alerts)


@pytest.mark.asyncio
async def test_alert_limit():
    """测试告警数量限制"""
    monitor = PermissionMonitor()

    for _ in range(50):
        monitor.record_slow_check_alert(100)

    assert len(monitor.get_alerts(limit=10)) == 10


@pytest.mark.asyncio
async def test_metrics_reset():
    """测试指标重置"""
    monitor = PermissionMonitor()

    monitor.record_cache_hit()
    assert monitor.metrics.cache_metrics.total_accesses == 1

    monitor.reset_metrics()

    assert monitor.metrics.cache_metrics.total_accesses == 0


@pytest.mark.asyncio
async def test_alerts_clear():
    """测试告警清除"""
    monitor = PermissionMonitor()

    monitor.record_slow_check_alert(100)
    assert len(monitor.get_alerts()) == 1

    monitor.clear_alerts()

    assert monitor.get_alerts() == []


@pytest.mark.asyncio
async def test_metrics_to_dict():
    """测试指标转换为字典"""
    monitor = PermissionMonitor()

    monitor.record_cache_hit()
    monitor.record_cache_miss()

    metrics_dict = monitor.get_metrics()

    assert "cache_metrics" in metrics_dict
    assert "uptime_seconds" in metrics_dict
    assert metrics_dict["cache_metrics"]["total_accesses"] == 2
    assert metrics_dict["cache_metrics"]["cache_hits"] == 1
    assert metrics_dict["cache_metrics"]["cache_misses"] == 1


@pytest.mark.asyncio
async def test_global_monitor_instance():
    """测试全局监控器实例"""
    monitor1 = get_permission_monitor()
    monitor2 = get_permission_monitor()

    assert monitor1 is monitor2


@pytest.mark.asyncio
async def test_set_global_monitor():
    """测试设置全局监控器实例"""
    new_monitor = PermissionMonitor()
    set_permission_monitor(new_monitor)

    assert get_permission_monitor() is new_monitor


@pytest.mark.asyncio
async def test_alert_timestamp():
    """测试告警时间戳"""
    monitor = PermissionMonitor()

    before_time = time.time()
    monitor.record_slow_check_alert(100)
    after_time = time.time()

    alerts = monitor.get_alerts()
    assert before_time <= alerts[0]["timestamp"] <= after_time


@pytest.mark.asyncio
async def test_cache_metrics_dataclass():
    """测试缓存指标数据类"""
    metrics = CacheMetrics()

    assert metrics.total_accesses == 0
    assert metrics.hit_rate == 0.0

    metrics.total_accesses = 100
    metrics.cache_hits = 80
    metrics.cache_misses = 20

    assert metrics.hit_rate == pytest.approx(80.0, 0.1)
    assert metrics.miss_rate == pytest.approx(20.0, 0.1)