ctms/backend/tests/test_permission_monitoring_api.py

"""监控API测试：权限系统监控API端点验证

测试权限系统监控API的功能。
"""

import pytest
from fastapi.testclient import TestClient
from sqlalchemy import text

from app.core.permission_monitor import get_permission_monitor, set_permission_monitor, PermissionMonitor
from app.api.v1 import permission_monitoring


class FakeIpInfo:
    def __init__(self, province: str, city: str) -> None:
        self.country = "中国"
        self.province = province
        self.city = city
        self.isp = "电信"
        self.location = f"中国 / {province} / {city} / 电信"


@pytest.mark.asyncio
async def test_get_permission_metrics(client: TestClient, auth_headers: dict):
    """测试获取权限系统指标"""
    # 清除并重置监控器
    monitor = PermissionMonitor()
    set_permission_monitor(monitor)

    # 记录一些指标
    monitor.record_permission_check(allowed=True, elapsed_time=0.005)
    monitor.record_permission_check(allowed=False, elapsed_time=0.003)

    response = client.get("/api/v1/permission-monitoring/metrics", headers=auth_headers)
    assert response.status_code == 200

    data = response.json()
    assert "check_metrics" in data
    assert "cache_metrics" in data
    assert data["check_metrics"]["total_checks"] == 2


@pytest.mark.asyncio
async def test_get_cache_statistics(client: TestClient, auth_headers: dict):
    """测试获取缓存统计"""
    # 清除并重置监控器
    monitor = PermissionMonitor()
    set_permission_monitor(monitor)

    # 记录缓存访问
    monitor.record_cache_hit()
    monitor.record_cache_hit()
    monitor.record_cache_miss()

    response = client.get("/api/v1/permission-monitoring/cache-stats", headers=auth_headers)
    assert response.status_code == 200

    data = response.json()
    assert "cache_metrics" in data
    assert data["cache_metrics"]["total_accesses"] == 3
    assert data["cache_metrics"]["cache_hits"] == 2


@pytest.mark.asyncio
async def test_get_alerts(client: TestClient, auth_headers: dict):
    """测试获取告警列表"""
    # 清除并重置监控器
    monitor = PermissionMonitor()
    set_permission_monitor(monitor)

    # 生成告警
    monitor.record_permission_check(allowed=True, elapsed_time=0.1)

    response = client.get("/api/v1/permission-monitoring/alerts", headers=auth_headers)
    assert response.status_code == 200

    data = response.json()
    assert "alerts" in data
    assert data["total"] > 0


@pytest.mark.asyncio
async def test_get_alerts_with_level_filter(client: TestClient, auth_headers: dict):
    """测试按级别过滤告警"""
    # 清除并重置监控器
    monitor = PermissionMonitor()
    set_permission_monitor(monitor)

    # 生成告警
    monitor.record_permission_check(allowed=True, elapsed_time=0.1)

    response = client.get(
        "/api/v1/permission-monitoring/alerts?level=warning",
        headers=auth_headers,
    )
    assert response.status_code == 200

    data = response.json()
    assert "alerts" in data


@pytest.mark.asyncio
async def test_get_alerts_with_limit(client: TestClient, auth_headers: dict):
    """测试限制告警数量"""
    # 清除并重置监控器
    monitor = PermissionMonitor()
    set_permission_monitor(monitor)

    # 生成多个告警
    for _ in range(20):
        monitor.record_permission_check(allowed=True, elapsed_time=0.1)

    response = client.get(
        "/api/v1/permission-monitoring/alerts?limit=5",
        headers=auth_headers,
    )
    assert response.status_code == 200

    data = response.json()
    assert len(data["alerts"]) <= 5


@pytest.mark.asyncio
async def test_reset_metrics(client: TestClient, auth_headers: dict):
    """测试重置指标"""
    # 清除并重置监控器
    monitor = PermissionMonitor()
    set_permission_monitor(monitor)

    # 记录指标
    monitor.record_permission_check(allowed=True, elapsed_time=0.005)
    assert monitor.metrics.check_metrics.total_checks == 1

    # 重置指标
    response = client.post("/api/v1/permission-monitoring/reset-metrics", headers=auth_headers)
    assert response.status_code == 200

    # 验证指标已重置
    assert monitor.metrics.check_metrics.total_checks == 0


@pytest.mark.asyncio
async def test_clear_alerts(client: TestClient, auth_headers: dict):
    """测试清除告警"""
    # 清除并重置监控器
    monitor = PermissionMonitor()
    set_permission_monitor(monitor)

    # 生成告警
    monitor.record_permission_check(allowed=True, elapsed_time=0.1)
    assert len(monitor.get_alerts()) > 0

    # 清除告警
    response = client.post("/api/v1/permission-monitoring/clear-alerts", headers=auth_headers)
    assert response.status_code == 200

    # 验证告警已清除
    assert len(monitor.get_alerts()) == 0


@pytest.mark.asyncio
async def test_permission_system_health_healthy(client: TestClient, auth_headers: dict):
    """测试权限系统健康检查（健康状态）"""
    # 清除并重置监控器
    monitor = PermissionMonitor()
    set_permission_monitor(monitor)

    # 记录良好的指标
    for _ in range(100):
        monitor.record_permission_check(allowed=True, elapsed_time=0.001)
    for _ in range(100):
        monitor.record_cache_hit()

    response = client.get("/api/v1/permission-monitoring/health", headers=auth_headers)
    assert response.status_code == 200

    data = response.json()
    assert data["status"] in ["healthy", "degraded"]
    assert data["health_score"] > 50


@pytest.mark.asyncio
async def test_permission_system_health_degraded(client: TestClient, auth_headers: dict):
    """测试权限系统健康检查（降级状态）"""
    # 清除并重置监控器
    monitor = PermissionMonitor()
    set_permission_monitor(monitor)

    # 记录不良的指标
    for _ in range(100):
        monitor.record_permission_check(allowed=False, elapsed_time=0.1)
    for _ in range(100):
        monitor.record_cache_miss()

    response = client.get("/api/v1/permission-monitoring/health", headers=auth_headers)
    assert response.status_code == 200

    data = response.json()
    assert "status" in data
    assert "health_score" in data
    assert "issues" in data


@pytest.mark.asyncio
async def test_permission_system_health_includes_metrics(client: TestClient, auth_headers: dict):
    """测试健康检查包含详细指标"""
    # 清除并重置监控器
    monitor = PermissionMonitor()
    set_permission_monitor(monitor)

    response = client.get("/api/v1/permission-monitoring/health", headers=auth_headers)
    assert response.status_code == 200

    data = response.json()
    assert "metrics" in data
    assert "cache_stats" in data
    assert "check_metrics" in data["metrics"]
    assert "cache_metrics" in data["metrics"]


@pytest.mark.asyncio
async def test_ip_locations_counts_unique_users_per_location(db_session, monkeypatch):
    """IP 属地统计应按省市聚合访问次数、来源 IP 数和访问用户数。"""
    await db_session.execute(text("DELETE FROM permission_access_logs"))
    await db_session.commit()

    study_id = "00000000-0000-0000-0000-000000000001"
    user_a = "00000000-0000-0000-0000-000000000101"
    user_b = "00000000-0000-0000-0000-000000000102"

    await db_session.execute(
        text(
            """
            INSERT INTO studies (id, code, name, status, is_locked, visit_schedule, active_roles)
            VALUES (:id, :code, :name, :status, :is_locked, :visit_schedule, :active_roles)
            """
        ),
        {
            "id": study_id,
            "code": "IP-LOCATION-STUDY",
            "name": "IP Location Study",
            "status": "ACTIVE",
            "is_locked": False,
            "visit_schedule": "[]",
            "active_roles": "[]",
        },
    )
    for user_id, email in [(user_a, "user-a@example.com"), (user_b, "user-b@example.com")]:
        await db_session.execute(
            text(
                """
                INSERT INTO users (id, email, password_hash, full_name, role, clinical_department, status)
                VALUES (:id, :email, :password_hash, :full_name, :role, :clinical_department, :status)
                """
            ),
            {
                "id": user_id,
                "email": email,
                "password_hash": "hash",
                "full_name": email,
                "role": "PM",
                "clinical_department": "临床运营",
                "status": "ACTIVE",
            },
        )

    rows = [
        (study_id, user_a, "10.1.1.1", True),
        (study_id, user_a, "10.1.1.1", False),
        (study_id, user_b, "10.1.1.2", True),
    ]
    for study, user, ip_address, allowed in rows:
        await db_session.execute(
            text(
                """
                INSERT INTO permission_access_logs
                    (id, study_id, user_id, endpoint_key, role, allowed, elapsed_ms, ip_address, created_at)
                VALUES
                    (lower(hex(randomblob(4))) || '-' || lower(hex(randomblob(2))) || '-4' ||
                     substr(lower(hex(randomblob(2))),2) || '-' ||
                     substr('89ab', abs(random()) % 4 + 1, 1) ||
                     substr(lower(hex(randomblob(2))),2) || '-' || lower(hex(randomblob(6))),
                     :study_id, :user_id, :endpoint_key, :role, :allowed, :elapsed_ms, :ip_address, CURRENT_TIMESTAMP)
                """
            ),
            {
                "study_id": study,
                "user_id": user,
                "endpoint_key": "admin.permissions.read",
                "role": "PM",
                "allowed": allowed,
                "elapsed_ms": 3.2,
                "ip_address": ip_address,
            },
        )
    await db_session.commit()

    monkeypatch.setattr(
        permission_monitoring,
        "resolve_ip_location",
        lambda ip: FakeIpInfo("广东省", "深圳市"),
    )

    result = await permission_monitoring.get_ip_locations(db=db_session, _=object(), days=7, limit=10)

    assert result["items"][0]["province"] == "广东省"
    assert result["items"][0]["total_count"] == 3
    assert result["items"][0]["unique_ip_count"] == 2
    assert result["items"][0]["unique_user_count"] == 2
    assert result["summary"]["total_count"] == 3
    assert result["summary"]["unique_ip_count"] == 2
    assert result["summary"]["unique_user_count"] == 2


@pytest.mark.asyncio
async def test_access_logs_include_user_behavior_summary(db_session):
    """访问日志应返回用户行为审计汇总和用户排行。"""
    await db_session.execute(text("DELETE FROM permission_access_logs"))
    await db_session.commit()

    study_id = "00000000-0000-0000-0000-000000000201"
    user_a = "00000000-0000-0000-0000-000000000301"
    user_b = "00000000-0000-0000-0000-000000000302"

    await db_session.execute(
        text(
            """
            INSERT INTO studies (id, code, name, status, is_locked, visit_schedule, active_roles)
            VALUES (:id, :code, :name, :status, :is_locked, :visit_schedule, :active_roles)
            """
        ),
        {
            "id": study_id,
            "code": "ACCESS-LOG-STUDY",
            "name": "Access Log Study",
            "status": "ACTIVE",
            "is_locked": False,
            "visit_schedule": "[]",
            "active_roles": "[]",
        },
    )
    for user_id, email, name, role in [
        (user_a, "audit-a@example.com", "审计用户A", "PM"),
        (user_b, "audit-b@example.com", "审计用户B", "CRA"),
    ]:
        await db_session.execute(
            text(
                """
                INSERT INTO users (id, email, password_hash, full_name, role, clinical_department, status)
                VALUES (:id, :email, :password_hash, :full_name, :role, :clinical_department, :status)
                """
            ),
            {
                "id": user_id,
                "email": email,
                "password_hash": "hash",
                "full_name": name,
                "role": role,
                "clinical_department": "临床运营",
                "status": "ACTIVE",
            },
        )

    rows = [
        (study_id, user_a, "admin.permissions.read", "PM", True, 3.0, "10.1.1.1"),
        (study_id, user_a, "admin.users.delete", "PM", False, 9.0, "10.1.1.2"),
        (study_id, user_b, "admin.permissions.read", "CRA", True, 6.0, "10.1.1.3"),
    ]
    for study, user, endpoint, role, allowed, elapsed_ms, ip_address in rows:
        await db_session.execute(
            text(
                """
                INSERT INTO permission_access_logs
                    (id, study_id, user_id, endpoint_key, role, allowed, elapsed_ms, ip_address, created_at)
                VALUES
                    (lower(hex(randomblob(4))) || '-' || lower(hex(randomblob(2))) || '-4' ||
                     substr(lower(hex(randomblob(2))),2) || '-' ||
                     substr('89ab', abs(random()) % 4 + 1, 1) ||
                     substr(lower(hex(randomblob(2))),2) || '-' || lower(hex(randomblob(6))),
                     :study_id, :user_id, :endpoint_key, :role, :allowed, :elapsed_ms, :ip_address, CURRENT_TIMESTAMP)
                """
            ),
            {
                "study_id": study,
                "user_id": user,
                "endpoint_key": endpoint,
                "role": role,
                "allowed": allowed,
                "elapsed_ms": elapsed_ms,
                "ip_address": ip_address,
            },
        )
    await db_session.commit()

    result = await permission_monitoring.get_access_logs(
        db=db_session,
        _=object(),
        study_id=None,
        user_id=None,
        endpoint_key=None,
        role=None,
        allowed=None,
        start_time=None,
        end_time=None,
        page=1,
        page_size=50,
    )

    assert result["summary"]["total_count"] == 3
    assert result["summary"]["unique_user_count"] == 2
    assert result["summary"]["denied_count"] == 1
    assert result["summary"]["avg_elapsed_ms"] == 6.0
    assert result["user_stats"][0]["user_name"] == "审计用户A"
    assert result["user_stats"][0]["total_count"] == 1
    assert result["user_stats"][0]["denied_count"] in {0, 1}
    assert result["user_stats"][0]["unique_ip_count"] == 1
    assert result["user_stats"][0]["sample_ip_address"] in {"10.1.1.1", "10.1.1.2"}
    user_a_ips = {
        stat["sample_ip_address"]
        for stat in result["user_stats"]
        if stat["user_name"] == "审计用户A"
    }
    assert user_a_ips == {"10.1.1.1", "10.1.1.2"}


@pytest.mark.asyncio
async def test_security_access_logs_include_anonymous_ip_attempts(db_session):
    """安全访问日志应覆盖未登录或未知账号的底层访问尝试。"""
    await db_session.execute(text("DELETE FROM security_access_logs"))
    await db_session.execute(
        text(
            """
            INSERT INTO security_access_logs
                (id, method, path, status_code, elapsed_ms, client_ip, user_agent, auth_status, user_identifier, created_at)
            VALUES
                ('00000000-0000-4000-8000-000000000501', 'POST', '/api/v1/auth/login', 401, 12.5,
                 '203.0.113.10', 'attack-bot/1.0', 'ANONYMOUS', NULL, CURRENT_TIMESTAMP)
            """
        )
    )
    await db_session.commit()

    result = await permission_monitoring.get_security_access_logs(
        db=db_session,
        _=object(),
        status_min=400,
        auth_status=None,
        page=1,
        page_size=20,
    )

    assert result["summary"]["total_count"] == 1
    assert result["summary"]["anonymous_count"] == 1
    assert result["summary"]["error_count"] == 1
    assert result["items"][0]["client_ip"] == "203.0.113.10"
    assert result["items"][0]["account_label"] == "未知账号"
    assert result["items"][0]["status_code"] == 401