ctms/backend/tests/test_permission_monitoring_api.py

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

import inspect
import uuid

import pytest
from fastapi import HTTPException
from sqlalchemy import text

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


class FakeIpInfo:
    def __init__(self, province: str, city: str, isp: str = "电信", country: str = "中国") -> None:
        self.country = country
        self.province = province
        self.city = city
        self.isp = isp
        self.location = " / ".join(part for part in [country, province, city, isp] if part)


class AdminUserStub:
    is_admin = True


class ProjectPmUserStub:
    id = uuid.uuid4()
    is_admin = False


@pytest.mark.asyncio
async def test_resolve_monitoring_scope_rejects_project_pm(db_session):
    """系统监测模块仅允许系统管理员访问，项目 PM 不再具备监测范围。"""
    with pytest.raises(HTTPException) as exc_info:
        await permission_monitoring.resolve_monitoring_scope(db_session, ProjectPmUserStub())

    assert exc_info.value.status_code == 403


@pytest.mark.asyncio
async def test_system_permission_monitoring_definitions_are_admin_only():
    """系统级权限定义中的系统监测模块不应再标注 PM 角色。"""
    data = await list_system_permissions()
    monitoring_items = [
        item for item in data["permissions"]
        if item["module"] == "system_monitoring"
    ]

    assert monitoring_items
    assert data["module_labels"]["system_monitoring"] == "系统监测"
    assert all(item["roles"] == ["ADMIN"] for item in monitoring_items)
    assert all("PM" not in item["roles"] for item in monitoring_items)


async def _seed_permission_log(db_session, study_id: uuid.UUID, user_id: uuid.UUID, *, allowed: bool, elapsed_ms: float) -> None:
    study_exists = (
        await db_session.execute(text("SELECT id FROM studies WHERE id = :id"), {"id": str(study_id)})
    ).scalar_one_or_none()
    if not study_exists:
        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": str(study_id),
                "code": f"PERM-MON-{study_id.hex[:8]}",
                "name": "Permission Monitoring Study",
                "status": "ACTIVE",
                "is_locked": False,
                "visit_schedule": "[]",
                "active_roles": "[]",
            },
        )

    user_exists = (
        await db_session.execute(text("SELECT id FROM users WHERE id = :id"), {"id": str(user_id)})
    ).scalar_one_or_none()
    if not user_exists:
        await db_session.execute(
            text(
                """
                INSERT INTO users (id, email, password_hash, full_name, clinical_department, is_admin, status)
                VALUES (:id, :email, :password_hash, :full_name, :clinical_department, :is_admin, :status)
                """
            ),
            {
                "id": str(user_id),
                "email": f"{user_id.hex}@example.com",
                "password_hash": "hash",
                "full_name": "Permission Monitoring User",
                "clinical_department": "临床运营",
                "is_admin": False,
                "status": "ACTIVE",
            },
        )

    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
                (:id, :study_id, :user_id, :endpoint_key, :role, :allowed, :elapsed_ms, :ip_address, CURRENT_TIMESTAMP)
            """
        ),
        {
            "id": str(uuid.uuid4()),
            "study_id": str(study_id),
            "user_id": str(user_id),
            "endpoint_key": "admin.permissions.read",
            "role": "PM",
            "allowed": allowed,
            "elapsed_ms": elapsed_ms,
            "ip_address": "127.0.0.1",
        },
    )
    await db_session.commit()


@pytest.mark.asyncio
async def test_get_permission_metrics(db_session):
    """测试获取权限系统指标"""
    monitor = PermissionMonitor()
    set_permission_monitor(monitor)

    study_id = uuid.uuid4()
    user_id = uuid.uuid4()
    await _seed_permission_log(db_session, study_id, user_id, allowed=True, elapsed_ms=5)
    await _seed_permission_log(db_session, study_id, user_id, allowed=False, elapsed_ms=3)

    data = await permission_monitoring.get_permission_metrics(db=db_session, _=AdminUserStub(), hours=24)
    assert "check_metrics" in data
    assert "cache_metrics" in data
    assert data["check_metrics"]["total_checks"] == 2
    assert data["check_metrics"]["allowed_checks"] == 1
    assert data["check_metrics"]["denied_checks"] == 1


@pytest.mark.asyncio
async def test_get_cache_statistics(db_session):
    """测试获取缓存统计"""
    monitor = PermissionMonitor()
    set_permission_monitor(monitor)

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

    data = await permission_monitoring.get_cache_statistics(_=AdminUserStub(), db=db_session)
    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(db_session):
    """测试获取告警列表"""
    monitor = PermissionMonitor()
    set_permission_monitor(monitor)

    monitor.record_slow_check_alert(100)

    data = await permission_monitoring.get_alerts(_=AdminUserStub(), db=db_session)
    assert "alerts" in data
    assert data["total"] > 0


@pytest.mark.asyncio
async def test_get_alerts_with_level_filter(db_session):
    """测试按级别过滤告警"""
    monitor = PermissionMonitor()
    set_permission_monitor(monitor)

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

    data = await permission_monitoring.get_alerts(level="warning", _=AdminUserStub(), db=db_session)
    assert "alerts" in data
    assert all(alert["level"] == "warning" for alert in data["alerts"])


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

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

    data = await permission_monitoring.get_alerts(limit=5, _=AdminUserStub(), db=db_session)
    assert len(data["alerts"]) <= 5


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

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

    result = await permission_monitoring.reset_metrics(_=AdminUserStub(), db=db_session)

    assert result["message"] == "指标已重置"
    assert monitor.metrics.cache_metrics.total_accesses == 0


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

    monitor.record_slow_check_alert(100)
    assert len(monitor.get_alerts()) > 0

    result = await permission_monitoring.clear_alerts(_=AdminUserStub(), db=db_session)

    assert result["message"] == "告警已清除"
    assert len(monitor.get_alerts()) == 0


@pytest.mark.asyncio
async def test_permission_system_health_healthy(db_session):
    """测试权限系统健康检查（健康状态）"""
    monitor = PermissionMonitor()
    set_permission_monitor(monitor)

    for _ in range(100):
        monitor.record_cache_hit()

    data = await permission_monitoring.permission_system_health(db=db_session, _=AdminUserStub())
    assert data["status"] == "healthy"
    assert data["health_score"] >= 80


@pytest.mark.asyncio
async def test_permission_system_health_degraded(db_session):
    """测试权限系统健康检查（降级状态）"""
    monitor = PermissionMonitor()
    set_permission_monitor(monitor)

    study_id = uuid.uuid4()
    user_id = uuid.uuid4()
    for _ in range(10):
        await _seed_permission_log(db_session, study_id, user_id, allowed=False, elapsed_ms=100)
    for _ in range(100):
        monitor.record_cache_miss()

    data = await permission_monitoring.permission_system_health(db=db_session, _=AdminUserStub())
    assert "status" in data
    assert "health_score" in data
    assert "issues" in data
    assert any("权限检查平均响应时间过长" in issue for issue in data["issues"])
    assert any("权限拒绝率偏高" in issue for issue in data["issues"])
    assert any("缓存命中率偏低" in issue for issue in data["issues"])
    assert data["status"] == "unhealthy"
    assert data["health_score"] == 30
    assert data["score_breakdown"]["performance"]["deduction"] == 25
    assert data["score_breakdown"]["deny_rate"]["deduction"] == 20
    assert data["score_breakdown"]["cache"]["deduction"] == 25


@pytest.mark.asyncio
async def test_permission_system_health_includes_metrics(db_session):
    """测试健康检查包含详细指标"""
    monitor = PermissionMonitor()
    set_permission_monitor(monitor)

    data = await permission_monitoring.permission_system_health(db=db_session, _=AdminUserStub())
    assert "last_hour" in data
    assert "cache_stats" in data
    assert "score_breakdown" in data
    assert "sample_sufficient" in data
    assert "total_checks" in data["last_hour"]
    assert "cache_metrics" in data["cache_stats"]


@pytest.mark.asyncio
async def test_permission_system_health_ignores_small_cache_sample(db_session):
    """缓存样本不足时不应因为低命中率扣分"""
    monitor = PermissionMonitor()
    set_permission_monitor(monitor)

    for _ in range(3):
        monitor.record_cache_miss()

    data = await permission_monitoring.permission_system_health(db=db_session, _=AdminUserStub())

    assert data["score_breakdown"]["cache"]["sample_sufficient"] is False
    assert data["score_breakdown"]["cache"]["deduction"] == 0
    non_cache_deductions = sum(
        item["deduction"]
        for name, item in data["score_breakdown"].items()
        if name != "cache"
    )
    assert data["health_score"] == 100 - non_cache_deductions


@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, clinical_department, is_admin, status)
                VALUES (:id, :email, :password_hash, :full_name, :clinical_department, :is_admin, :status)
                """
            ),
            {
                "id": user_id,
                "email": email,
                "password_hash": "hash",
                "full_name": email,
                "clinical_department": "临床运营",
                "is_admin": False,
                "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, _=AdminUserStub(), 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_ip_locations_merges_same_region_with_different_isp(db_session, monkeypatch):
    """同一省市的 IP 属地统计不应因运营商不同拆分。"""
    await db_session.execute(text("DELETE FROM permission_access_logs"))
    await db_session.commit()

    study_id = "00000000-0000-0000-0000-000000000401"
    user_id = "00000000-0000-0000-0000-000000000501"

    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-MERGE-STUDY",
            "name": "IP Location Merge Study",
            "status": "ACTIVE",
            "is_locked": False,
            "visit_schedule": "[]",
            "active_roles": "[]",
        },
    )
    await db_session.execute(
        text(
            """
            INSERT INTO users (id, email, password_hash, full_name, clinical_department, is_admin, status)
            VALUES (:id, :email, :password_hash, :full_name, :clinical_department, :is_admin, :status)
            """
        ),
        {
            "id": user_id,
            "email": "region-merge@example.com",
            "password_hash": "hash",
            "full_name": "属地合并用户",
            "clinical_department": "临床运营",
            "is_admin": False,
            "status": "ACTIVE",
        },
    )
    for ip_address in ["10.2.1.1", "10.2.1.2"]:
        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
                    (:id, :study_id, :user_id, :endpoint_key, :role, :allowed, :elapsed_ms, :ip_address, CURRENT_TIMESTAMP)
                """
            ),
            {
                "id": str(uuid.uuid4()),
                "study_id": study_id,
                "user_id": user_id,
                "endpoint_key": "admin.permissions.read",
                "role": "PM",
                "allowed": True,
                "elapsed_ms": 3.2,
                "ip_address": ip_address,
            },
        )
    await db_session.commit()

    ip_info_by_address = {
        "10.2.1.1": FakeIpInfo("重庆", "重庆市", "电信"),
        "10.2.1.2": FakeIpInfo("重庆", "重庆市", "联通"),
    }
    monkeypatch.setattr(
        permission_monitoring,
        "resolve_ip_location",
        lambda ip: ip_info_by_address[ip],
    )

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

    assert len(result["items"]) == 1
    assert result["items"][0]["province"] == "重庆"
    assert result["items"][0]["city"] == "重庆市"
    assert result["items"][0]["total_count"] == 2
    assert result["items"][0]["unique_ip_count"] == 2


@pytest.mark.asyncio
async def test_ip_locations_keeps_private_network_location_label(db_session, monkeypatch):
    """局域网 IP 无省市信息时，属地统计仍应保留可展示标签。"""
    await db_session.execute(text("DELETE FROM permission_access_logs"))
    await db_session.commit()

    study_id = "00000000-0000-0000-0000-000000000601"
    user_id = "00000000-0000-0000-0000-000000000701"

    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-PRIVATE-STUDY",
            "name": "IP Location Private Study",
            "status": "ACTIVE",
            "is_locked": False,
            "visit_schedule": "[]",
            "active_roles": "[]",
        },
    )
    await db_session.execute(
        text(
            """
            INSERT INTO users (id, email, password_hash, full_name, clinical_department, is_admin, status)
            VALUES (:id, :email, :password_hash, :full_name, :clinical_department, :is_admin, :status)
            """
        ),
        {
            "id": user_id,
            "email": "private-network@example.com",
            "password_hash": "hash",
            "full_name": "局域网用户",
            "clinical_department": "临床运营",
            "is_admin": False,
            "status": "ACTIVE",
        },
    )
    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
                (:id, :study_id, :user_id, :endpoint_key, :role, :allowed, :elapsed_ms, :ip_address, CURRENT_TIMESTAMP)
            """
        ),
        {
            "id": str(uuid.uuid4()),
            "study_id": study_id,
            "user_id": user_id,
            "endpoint_key": "admin.permissions.read",
            "role": "PM",
            "allowed": True,
            "elapsed_ms": 3.2,
            "ip_address": "192.168.1.10",
        },
    )
    await db_session.commit()

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

    assert result["items"][0]["location"] == "局域网"
    assert result["items"][0]["province"] == ""
    assert result["items"][0]["city"] == ""


@pytest.mark.asyncio
async def test_ip_locations_includes_abnormal_security_ips(db_session, monkeypatch):
    """来源分析应同时统计安全事件中的异常来源 IP。"""
    await db_session.execute(text("DELETE FROM permission_access_logs"))
    await db_session.execute(text("DELETE FROM security_access_logs"))
    await db_session.commit()

    study_id = "00000000-0000-0000-0000-000000000801"
    user_id = "00000000-0000-0000-0000-000000000901"

    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-SECURITY-STUDY",
            "name": "IP Location Security Study",
            "status": "ACTIVE",
            "is_locked": False,
            "visit_schedule": "[]",
            "active_roles": "[]",
        },
    )
    await db_session.execute(
        text(
            """
            INSERT INTO users (id, email, password_hash, full_name, clinical_department, is_admin, status)
            VALUES (:id, :email, :password_hash, :full_name, :clinical_department, :is_admin, :status)
            """
        ),
        {
            "id": user_id,
            "email": "source-analysis@example.com",
            "password_hash": "hash",
            "full_name": "来源分析用户",
            "clinical_department": "临床运营",
            "is_admin": False,
            "status": "ACTIVE",
        },
    )
    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
                (:id, :study_id, :user_id, :endpoint_key, :role, :allowed, :elapsed_ms, :ip_address, CURRENT_TIMESTAMP)
            """
        ),
        {
            "id": str(uuid.uuid4()),
            "study_id": study_id,
            "user_id": user_id,
            "endpoint_key": "admin.permissions.read",
            "role": "PM",
            "allowed": True,
            "elapsed_ms": 3.2,
            "ip_address": "10.3.1.1",
        },
    )
    for ip_address in ["8.8.8.8", "1.1.1.1"]:
        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
                    (:id, :method, :path, :status_code, :elapsed_ms, :client_ip, :user_agent, :auth_status, :user_identifier, CURRENT_TIMESTAMP)
                """
            ),
            {
                "id": str(uuid.uuid4()),
                "method": "GET",
                "path": "/api/v1/admin",
                "status_code": 403,
                "elapsed_ms": 8.5,
                "client_ip": ip_address,
                "user_agent": "pytest",
                "auth_status": "ANONYMOUS",
                "user_identifier": None,
            },
        )
    await db_session.commit()

    ip_info_by_address = {
        "10.3.1.1": FakeIpInfo("", "", "", "局域网"),
        "8.8.8.8": FakeIpInfo("California", "Mountain View", "Google", "United States"),
        "1.1.1.1": FakeIpInfo("Queensland", "Brisbane", "Cloudflare", "Australia"),
    }
    monkeypatch.setattr(
        permission_monitoring,
        "resolve_ip_location",
        lambda ip: ip_info_by_address[ip],
    )

    result = await permission_monitoring.get_ip_locations(db=db_session, _=AdminUserStub(), days=7, limit=10)
    locations = {item["country"]: item for item in result["items"]}

    assert result["summary"]["total_count"] == 3
    assert result["summary"]["denied_count"] == 2
    assert result["summary"]["unique_ip_count"] == 3
    assert result["summary"]["unique_user_count"] == 1
    assert locations["United States"]["total_count"] == 1
    assert locations["United States"]["denied_count"] == 1
    assert locations["Australia"]["total_count"] == 1


@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, clinical_department, is_admin, status)
                VALUES (:id, :email, :password_hash, :full_name, :clinical_department, :is_admin, :status)
                """
            ),
            {
                "id": user_id,
                "email": email,
                "password_hash": "hash",
                "full_name": name,
                "clinical_department": "临床运营",
                "is_admin": False,
                "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,
        _=AdminUserStub(),
        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"}


def test_access_logs_user_stats_query_avoids_postgresql_uuid_max():
    """访问日志用户统计不能对 UUID 字段使用 max，PostgreSQL 不支持 max(uuid)。"""
    source = inspect.getsource(permission_monitoring.get_access_logs)

    assert "func.max(PermissionAccessLog.user_id)" not in source


@pytest.mark.asyncio
async def test_access_logs_ip_ranking_aggregates_by_ip_total(db_session):
    """IP 访问排行应按同一 IP 的整体访问量聚合排序，而不是按用户/IP/角色拆分。"""
    await db_session.execute(text("DELETE FROM permission_access_logs"))
    await db_session.commit()

    study_id = "00000000-0000-0000-0000-000000001201"
    user_a = "00000000-0000-0000-0000-000000001301"
    user_b = "00000000-0000-0000-0000-000000001302"

    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-IP-RANK-STUDY",
            "name": "Access IP Rank Study",
            "status": "ACTIVE",
            "is_locked": False,
            "visit_schedule": "[]",
            "active_roles": "[]",
        },
    )
    for user_id, email, name in [
        (user_a, "ip-rank-a@example.com", "IP排行用户A"),
        (user_b, "ip-rank-b@example.com", "IP排行用户B"),
    ]:
        await db_session.execute(
            text(
                """
                INSERT INTO users (id, email, password_hash, full_name, clinical_department, is_admin, status)
                VALUES (:id, :email, :password_hash, :full_name, :clinical_department, :is_admin, :status)
                """
            ),
            {
                "id": user_id,
                "email": email,
                "password_hash": "hash",
                "full_name": name,
                "clinical_department": "临床运营",
                "is_admin": False,
                "status": "ACTIVE",
            },
        )

    rows = [
        (user_a, "183.230.169.20", True),
        (user_a, "183.230.169.20", True),
        (user_b, "183.230.169.20", False),
        (user_a, "45.148.10.95", False),
        (user_b, "52.53.218.145", False),
    ]
    for 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
                    (:id, :study_id, :user_id, :endpoint_key, :role, :allowed, :elapsed_ms, :ip_address, CURRENT_TIMESTAMP)
                """
            ),
            {
                "id": str(uuid.uuid4()),
                "study_id": study_id,
                "user_id": user,
                "endpoint_key": "admin.permissions.read",
                "role": "CRA",
                "allowed": allowed,
                "elapsed_ms": 4.0,
                "ip_address": ip_address,
            },
        )
    await db_session.commit()

    result = await permission_monitoring.get_access_logs(
        db=db_session,
        _=AdminUserStub(),
        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["user_stats"][0]["sample_ip_address"] == "183.230.169.20"
    assert result["user_stats"][0]["total_count"] == 3
    assert result["user_stats"][0]["denied_count"] == 1
    assert {stat["sample_ip_address"] for stat in result["user_stats"][1:]} == {"52.53.218.145", "45.148.10.95"}


@pytest.mark.asyncio
async def test_security_access_logs_include_anonymous_ip_attempts(db_session, monkeypatch):
    """安全访问日志应覆盖未登录或未知账号的底层访问尝试。"""
    await db_session.execute(text("DELETE FROM security_access_logs"))
    monkeypatch.setattr(
        permission_monitoring,
        "resolve_ip_location",
        lambda ip: FakeIpInfo("上海市", "上海市", "联通"),
    )
    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,
        _=AdminUserStub(),
        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]["ip_location"] == "中国 / 上海市 / 上海市 / 联通"
    assert result["items"][0]["ip_country"] == "中国"
    assert result["items"][0]["ip_province"] == "上海市"
    assert result["items"][0]["ip_city"] == "上海市"
    assert result["items"][0]["ip_isp"] == "联通"
    assert result["items"][0]["account_label"] == "未知账号"
    assert result["items"][0]["status_code"] == 401


@pytest.mark.asyncio
async def test_security_access_logs_classify_sensitive_path_probe(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-000000000502', 'GET', '/api/.git/config', 404, 0.7,
                 '198.51.100.20', 'probe-bot/1.0', 'ANONYMOUS', NULL, CURRENT_TIMESTAMP)
            """
        )
    )
    await db_session.commit()

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

    assert result["items"][0]["category"] == "PROBE"
    assert result["items"][0]["severity"] == "CRITICAL"


@pytest.mark.asyncio
async def test_security_access_logs_prioritize_sensitive_probe_over_foreign_ip(db_session, monkeypatch):
    """敏感路径探测应优先于海外 IP 分类，避免降低风险等级。"""
    await db_session.execute(text("DELETE FROM security_access_logs"))
    monkeypatch.setattr(
        permission_monitoring,
        "resolve_ip_location",
        lambda ip: FakeIpInfo("South Holland", "", "", "Netherlands"),
    )
    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-000000000505', 'GET', '/api/.git/config', 404, 0.7,
                 '45.148.10.95', 'probe-bot/1.0', 'ANONYMOUS', NULL, CURRENT_TIMESTAMP)
            """
        )
    )
    await db_session.commit()

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

    assert result["items"][0]["category"] == "PROBE"
    assert result["items"][0]["severity"] == "CRITICAL"


@pytest.mark.asyncio
async def test_security_access_logs_classify_non_china_ip_as_abnormal(db_session, monkeypatch):
    """安全事件明细应把非中国公网 IP 归类为异常 IP。"""
    await db_session.execute(text("DELETE FROM security_access_logs"))
    monkeypatch.setattr(
        permission_monitoring,
        "resolve_ip_location",
        lambda ip: FakeIpInfo("California", "San Jose", "Google", "United States"),
    )
    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-000000000504', 'GET', '/api/v1/studies/', 401, 8.4,
                 '52.53.218.145', 'foreign-client/1.0', 'INVALID_TOKEN', NULL, CURRENT_TIMESTAMP)
            """
        )
    )
    await db_session.commit()

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

    item = result["items"][0]
    assert item["category"] == "ABNORMAL_IP"
    assert item["severity"] == "HIGH"
    assert item["ip_location"] == "United States / California / San Jose / Google"
    assert item["ip_country"] == "United States"


@pytest.mark.asyncio
async def test_security_access_logs_resolve_public_ip_with_packaged_ip2region(db_session):
    """安全中心公网 IP 应复用 ip2region 解析结果，避免前端只能显示未知位置。"""
    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-000000000503', 'GET', '/api/v1/auth/login', 404, 1.2,
                 '52.53.218.145', 'probe-bot/1.0', 'ANONYMOUS', NULL, CURRENT_TIMESTAMP)
            """
        )
    )
    await db_session.commit()

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

    item = result["items"][0]
    assert item["client_ip"] == "52.53.218.145"
    assert item["ip_location"] not in {"", "公网", "未知"}
    assert item["ip_country"] == "United States"
    assert item["ip_province"] == "California"
    assert item["ip_city"] == "San Jose"
    assert item["category"] == "ABNORMAL_IP"