ctms/backend/tests/test_api_permissions.py

"""单元测试：API权限检查函数"""

import pytest
import uuid
from pathlib import Path
import ast
from sqlalchemy import delete
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.api_permissions import API_ENDPOINT_PERMISSIONS, PROJECT_PERMISSION_ROLES
from app.core.project_permissions import role_has_api_permission
from app.core.project_permissions import get_api_endpoint_permissions, replace_api_endpoint_permissions
from app.core.permission_cache import PermissionCache, set_permission_cache
from app.core.permission_monitor import PermissionMonitor, set_permission_monitor
from app.models.api_endpoint_permission import ApiEndpointPermission
from app.models.study import Study
from app.models.user import UserRole
from app.schemas.member import StudyMemberCreate


@pytest.mark.asyncio
async def test_api_permission_check_allowed(db_session: AsyncSession):
    """测试接口级权限检查 - 允许"""
    study_id = uuid.uuid4()

    # 创建权限记录
    perm = ApiEndpointPermission(
        study_id=study_id,
        role="CRA",
        endpoint_key="subjects:create",
        allowed=True,
    )
    db_session.add(perm)
    await db_session.commit()

    # 验证权限（禁用前置权限检查）
    result = await role_has_api_permission(
        db_session, study_id, "CRA", "subjects:create", check_prerequisites=False
    )
    assert result is True


def test_all_backend_permission_guards_are_configurable():
    """确保后端实际鉴权使用的 operation key 都能在权限管理中配置。"""
    api_dir = Path(__file__).resolve().parents[1] / "app" / "api" / "v1"
    used_keys: set[str] = set()
    for path in api_dir.rglob("*.py"):
        source = path.read_text()
        tree = ast.parse(source)
        for node in ast.walk(tree):
            if not isinstance(node, ast.Call):
                continue
            if isinstance(node.func, ast.Name) and node.func.id == "require_api_permission":
                if node.args and isinstance(node.args[0], ast.Constant) and isinstance(node.args[0].value, str):
                    used_keys.add(node.args[0].value)
            if isinstance(node.func, ast.Name) and node.func.id == "_ensure_project_access":
                if len(node.args) >= 4 and isinstance(node.args[3], ast.Constant) and isinstance(node.args[3].value, str):
                    used_keys.add(node.args[3].value)

    assert used_keys
    assert used_keys <= set(API_ENDPOINT_PERMISSIONS)


def test_document_service_uses_specific_document_permission_keys():
    """文档模块应使用 create/update/delete 细粒度权限，而不是全部退化为 update。"""
    service_path = Path(__file__).resolve().parents[1] / "app" / "services" / "document_service.py"
    source = service_path.read_text()

    assert '"create_document": "documents:create"' in source
    assert '"create_version": "documents:update"' in source
    assert '"delete_document": "documents:delete"' in source
    assert 'else "documents:update"' not in source


def test_user_role_no_longer_contains_qa():
    assert "QA" not in {role.value for role in UserRole}


@pytest.mark.asyncio
async def test_default_matrix_covers_every_role_and_permission(db_session: AsyncSession):
    """逐一验证预设项目权限角色在每个权限上的默认矩阵。"""
    study_id = uuid.uuid4()
    matrix = await get_api_endpoint_permissions(db_session, study_id)

    assert "QA" not in PROJECT_PERMISSION_ROLES
    assert all("QA" not in config["default_roles"] for config in API_ENDPOINT_PERMISSIONS.values())
    assert set(matrix) == set(PROJECT_PERMISSION_ROLES)
    for role in PROJECT_PERMISSION_ROLES:
        assert set(matrix[role]) == set(API_ENDPOINT_PERMISSIONS)
        for endpoint_key, config in API_ENDPOINT_PERMISSIONS.items():
            assert matrix[role][endpoint_key]["allowed"] is (role in config["default_roles"])


@pytest.mark.asyncio
async def test_default_matrix_round_trips_to_backend_checks(db_session: AsyncSession):
    """默认权限矩阵应与后端实际鉴权结果一致。"""
    study_id = uuid.uuid4()

    for role in PROJECT_PERMISSION_ROLES:
        for endpoint_key, config in API_ENDPOINT_PERMISSIONS.items():
            allowed = await role_has_api_permission(
                db_session,
                study_id,
                role,
                endpoint_key,
                check_prerequisites=False,
            )
            assert allowed is (role in config["default_roles"])


@pytest.mark.asyncio
async def test_full_permission_matrix_round_trips_to_backend_checks(db_session: AsyncSession):
    """逐一验证前端提交格式会落库，并被后端鉴权函数按相同结果读取。"""
    cache = PermissionCache()
    monitor = PermissionMonitor()
    set_permission_cache(cache)
    set_permission_monitor(monitor)

    study_id = uuid.uuid4()
    configurable_roles = [role for role in PROJECT_PERMISSION_ROLES if role != "PM"]
    payload = {
        role: {
            endpoint_key: index % 2 == role_index % 2
            for index, endpoint_key in enumerate(API_ENDPOINT_PERMISSIONS)
        }
        for role_index, role in enumerate(configurable_roles)
    }

    matrix = await replace_api_endpoint_permissions(db_session, study_id, payload)

    for role in configurable_roles:
        for endpoint_key, expected in payload[role].items():
            assert matrix[role][endpoint_key]["allowed"] is expected
            allowed = await role_has_api_permission(
                db_session,
                study_id,
                role,
                endpoint_key,
                check_prerequisites=False,
            )
            assert allowed is expected


@pytest.mark.asyncio
async def test_custom_active_role_permissions_take_effect(db_session: AsyncSession):
    """验证项目自定义角色配置权限后能被后端鉴权读取。"""
    study_id = uuid.uuid4()
    db_session.add(
        Study(
            id=study_id,
            code=f"CUSTOM-{study_id.hex[:8]}",
            name="自定义角色权限测试",
            status="ACTIVE",
            is_locked=False,
            visit_schedule=[],
            active_roles=["DATA_MANAGER"],
        )
    )
    await db_session.commit()

    matrix = await get_api_endpoint_permissions(db_session, study_id)
    assert "DATA_MANAGER" in matrix
    assert matrix["DATA_MANAGER"]["subjects:read"]["allowed"] is False

    await replace_api_endpoint_permissions(
        db_session,
        study_id,
        {"DATA_MANAGER": {"subjects:read": True, "subjects:update": False}},
    )

    assert await role_has_api_permission(
        db_session,
        study_id,
        "DATA_MANAGER",
        "subjects:read",
        check_prerequisites=False,
    ) is True
    assert await role_has_api_permission(
        db_session,
        study_id,
        "DATA_MANAGER",
        "subjects:update",
        check_prerequisites=False,
    ) is False


def test_admin_cannot_be_used_as_project_custom_role():
    """避免项目角色 ADMIN 触发后端系统管理员绕过逻辑。"""
    with pytest.raises(ValueError):
        StudyMemberCreate(user_id=uuid.uuid4(), role_in_study="ADMIN")


@pytest.mark.asyncio
async def test_api_permission_check_denied(db_session: AsyncSession):
    """测试接口级权限检查 - 拒绝"""
    study_id = uuid.uuid4()

    # 创建权限记录
    perm = ApiEndpointPermission(
        study_id=study_id,
        role="PV",
        endpoint_key="subjects:create",
        allowed=False,
    )
    db_session.add(perm)
    await db_session.commit()

    # 验证权限（禁用前置权限检查）
    result = await role_has_api_permission(
        db_session, study_id, "PV", "subjects:create", check_prerequisites=False
    )
    assert result is False


@pytest.mark.asyncio
async def test_admin_always_allowed(db_session: AsyncSession):
    """测试ADMIN角色总是被允许"""
    study_id = uuid.uuid4()

    result = await role_has_api_permission(
        db_session, study_id, "ADMIN", "POST:/subjects"
    )
    assert result is True


@pytest.mark.asyncio
async def test_api_permission_read_endpoint(db_session: AsyncSession):
    """测试读取端点权限"""
    study_id = uuid.uuid4()

    # 创建读取权限
    perm = ApiEndpointPermission(
        study_id=study_id,
        role="PV",
        endpoint_key="subjects:read",
        allowed=True,
    )
    db_session.add(perm)
    await db_session.commit()

    # 验证权限（禁用前置权限检查）
    result = await role_has_api_permission(
        db_session, study_id, "PV", "subjects:read", check_prerequisites=False
    )
    assert result is True


@pytest.mark.asyncio
async def test_api_permission_different_endpoints(db_session: AsyncSession):
    """测试不同端点的权限独立"""
    study_id = uuid.uuid4()

    # 创建权限：允许GET，拒绝POST
    get_perm = ApiEndpointPermission(
        study_id=study_id,
        role="CRA",
        endpoint_key="subjects:read",
        allowed=True,
    )
    post_perm = ApiEndpointPermission(
        study_id=study_id,
        role="CRA",
        endpoint_key="subjects:create",
        allowed=False,
    )
    db_session.add(get_perm)
    db_session.add(post_perm)
    await db_session.commit()

    # 验证权限（禁用前置权限检查）
    get_result = await role_has_api_permission(
        db_session, study_id, "CRA", "subjects:read", check_prerequisites=False
    )
    post_result = await role_has_api_permission(
        db_session, study_id, "CRA", "subjects:create", check_prerequisites=False
    )

    assert get_result is True
    assert post_result is False


@pytest.mark.asyncio
async def test_api_permission_different_roles(db_session: AsyncSession):
    """测试不同角色的权限独立"""
    study_id = uuid.uuid4()

    # 创建权限：CRA允许，PV拒绝
    cra_perm = ApiEndpointPermission(
        study_id=study_id,
        role="CRA",
        endpoint_key="subjects:create",
        allowed=True,
    )
    pv_perm = ApiEndpointPermission(
        study_id=study_id,
        role="PV",
        endpoint_key="subjects:create",
        allowed=False,
    )
    db_session.add(cra_perm)
    db_session.add(pv_perm)
    await db_session.commit()

    # 验证权限（禁用前置权限检查）
    cra_result = await role_has_api_permission(
        db_session, study_id, "CRA", "subjects:create", check_prerequisites=False
    )
    pv_result = await role_has_api_permission(
        db_session, study_id, "PV", "subjects:create", check_prerequisites=False
    )

    assert cra_result is True
    assert pv_result is False


@pytest.mark.asyncio
async def test_api_permission_different_studies(db_session: AsyncSession):
    """测试不同项目的权限独立"""
    study_id_1 = uuid.uuid4()
    study_id_2 = uuid.uuid4()

    # 创建权限：项目1允许，项目2拒绝
    perm_1 = ApiEndpointPermission(
        study_id=study_id_1,
        role="CRA",
        endpoint_key="subjects:create",
        allowed=True,
    )
    perm_2 = ApiEndpointPermission(
        study_id=study_id_2,
        role="CRA",
        endpoint_key="subjects:create",
        allowed=False,
    )
    db_session.add(perm_1)
    db_session.add(perm_2)
    await db_session.commit()

    # 验证权限（禁用前置权限检查）
    result_1 = await role_has_api_permission(
        db_session, study_id_1, "CRA", "subjects:create", check_prerequisites=False
    )
    result_2 = await role_has_api_permission(
        db_session, study_id_2, "CRA", "subjects:create", check_prerequisites=False
    )

    assert result_1 is True
    assert result_2 is False


@pytest.mark.asyncio
async def test_api_permission_none_role(db_session: AsyncSession):
    """测试None角色的权限检查"""
    study_id = uuid.uuid4()

    result = await role_has_api_permission(
        db_session, study_id, None, "subjects:create", check_prerequisites=False
    )
    assert result is False


@pytest.mark.asyncio
async def test_api_permission_unknown_endpoint(db_session: AsyncSession):
    """测试未知端点的权限检查"""
    study_id = uuid.uuid4()

    result = await role_has_api_permission(
        db_session, study_id, "CRA", "unknown:endpoint", check_prerequisites=False
    )
    assert result is False


@pytest.mark.asyncio
async def test_api_permission_check_uses_project_permission_cache(db_session: AsyncSession):
    """测试接口权限检查会复用项目权限缓存并记录命中指标"""
    cache = PermissionCache()
    monitor = PermissionMonitor()
    set_permission_cache(cache)
    set_permission_monitor(monitor)
    study_id = uuid.uuid4()

    db_session.add(
        ApiEndpointPermission(
            study_id=study_id,
            role="CRA",
            endpoint_key="subjects:create",
            allowed=True,
        )
    )
    await db_session.commit()

    first_result = await role_has_api_permission(
        db_session, study_id, "CRA", "subjects:create", check_prerequisites=False
    )
    await db_session.execute(
        delete(ApiEndpointPermission).where(ApiEndpointPermission.study_id == study_id)
    )
    await db_session.commit()
    second_result = await role_has_api_permission(
        db_session, study_id, "CRA", "subjects:create", check_prerequisites=False
    )

    cache_metrics = monitor.get_metrics()["cache_metrics"]
    assert first_result is True
    assert second_result is True
    assert cache.get_cache_stats()["project_permissions_count"] == 1
    assert cache_metrics["cache_misses"] == 1
    assert cache_metrics["cache_hits"] == 1


@pytest.mark.asyncio
async def test_replace_api_endpoint_permissions_invalidates_project_permission_cache(
    db_session: AsyncSession,
):
    """测试替换权限后会失效旧缓存并记录失效指标"""
    from app.core.project_permissions import replace_api_endpoint_permissions

    cache = PermissionCache()
    monitor = PermissionMonitor()
    set_permission_cache(cache)
    set_permission_monitor(monitor)
    study_id = uuid.uuid4()

    await replace_api_endpoint_permissions(
        db_session,
        study_id,
        {"CRA": {"subjects:create": True}},
    )
    assert cache.get_cache_stats()["project_permissions_count"] == 1

    await replace_api_endpoint_permissions(
        db_session,
        study_id,
        {"CRA": {"subjects:create": False}},
    )

    result = await role_has_api_permission(
        db_session, study_id, "CRA", "subjects:create", check_prerequisites=False
    )
    cache_metrics = monitor.get_metrics()["cache_metrics"]
    assert result is False
    assert cache_metrics["cache_invalidations"] == 2