ctms/backend/tests/test_api_permissions_prerequisites.py

"""集成测试：权限管理API - 前置权限端点"""

import pytest
import uuid
from sqlalchemy.ext.asyncio import AsyncSession

from app.api.v1.api_permissions import (
    check_operation_prerequisites,
    list_api_operations,
    list_operation_prerequisites,
)
from app.models.api_endpoint_permission import ApiEndpointPermission


@pytest.mark.asyncio
async def test_list_operations_with_prerequisites():
    """测试获取所有权限操作及其前置权限"""
    data = await list_api_operations()
    assert "operations" in data

    # 验证返回的操作包含前置权限字段
    operations = data["operations"]
    assert len(operations) > 0

    # 查找 subjects:create 操作
    subjects_create = next(
        (op for op in operations if op["operation_key"] == "subjects:create"),
        None
    )
    assert subjects_create is not None
    assert "prerequisite_permissions" in subjects_create
    assert "sites:read" in subjects_create["prerequisite_permissions"]


@pytest.mark.asyncio
async def test_list_operation_prerequisites_endpoint():
    """测试获取所有操作的前置权限依赖"""
    data = await list_operation_prerequisites()
    assert "prerequisites" in data

    prerequisites = data["prerequisites"]

    # 验证 subjects:create 的前置权限
    assert "subjects:create" in prerequisites
    assert "sites:read" in prerequisites["subjects:create"]["prerequisites"]

    # 验证 visits:create 的前置权限
    assert "visits:create" in prerequisites
    assert "subjects:read" in prerequisites["visits:create"]["prerequisites"]
    assert "sites:read" in prerequisites["visits:create"]["prerequisites"]


@pytest.mark.asyncio
async def test_check_operation_prerequisites_satisfied(
    db_session: AsyncSession
):
    """测试检查操作前置权限 - 满足"""
    study_id = uuid.uuid4()
    # 创建权限：主权限 + 前置权限都允许
    main_perm = ApiEndpointPermission(
        study_id=study_id,
        role="CRA",
        endpoint_key="subjects:create",
        allowed=True,
    )
    prereq_perm = ApiEndpointPermission(
        study_id=study_id,
        role="CRA",
        endpoint_key="sites:read",
        allowed=True,
    )
    db_session.add(main_perm)
    db_session.add(prereq_perm)
    await db_session.commit()

    data = await check_operation_prerequisites(
        study_id=study_id,
        endpoint_key="subjects:create",
        role="CRA",
        db=db_session,
    )

    assert data["endpoint_key"] == "subjects:create"
    assert data["role"] == "CRA"
    assert data["has_main_permission"] is True
    assert data["missing_prerequisites"] == []
    assert data["can_perform"] is True


@pytest.mark.asyncio
async def test_check_operation_prerequisites_missing(
    db_session: AsyncSession
):
    """测试检查操作前置权限 - 缺失"""
    study_id = uuid.uuid4()
    # 创建权限：主权限允许，前置权限不允许
    main_perm = ApiEndpointPermission(
        study_id=study_id,
        role="CRA",
        endpoint_key="subjects:create",
        allowed=True,
    )
    prereq_perm = ApiEndpointPermission(
        study_id=study_id,
        role="CRA",
        endpoint_key="sites:read",
        allowed=False,
    )
    db_session.add(main_perm)
    db_session.add(prereq_perm)
    await db_session.commit()

    data = await check_operation_prerequisites(
        study_id=study_id,
        endpoint_key="subjects:create",
        role="CRA",
        db=db_session,
    )

    assert data["endpoint_key"] == "subjects:create"
    assert data["role"] == "CRA"
    assert data["has_main_permission"] is True
    assert "sites:read" in data["missing_prerequisites"]
    assert data["can_perform"] is False


@pytest.mark.asyncio
async def test_check_operation_prerequisites_multiple_missing(
    db_session: AsyncSession
):
    """测试检查操作前置权限 - 多个缺失"""
    study_id = uuid.uuid4()
    # 创建权限：主权限允许，两个前置权限都不允许
    main_perm = ApiEndpointPermission(
        study_id=study_id,
        role="CRA",
        endpoint_key="visits:create",
        allowed=True,
    )
    prereq1 = ApiEndpointPermission(
        study_id=study_id,
        role="CRA",
        endpoint_key="subjects:read",
        allowed=False,
    )
    prereq2 = ApiEndpointPermission(
        study_id=study_id,
        role="CRA",
        endpoint_key="sites:read",
        allowed=False,
    )
    db_session.add(main_perm)
    db_session.add(prereq1)
    db_session.add(prereq2)
    await db_session.commit()

    data = await check_operation_prerequisites(
        study_id=study_id,
        endpoint_key="visits:create",
        role="CRA",
        db=db_session,
    )

    assert data["endpoint_key"] == "visits:create"
    assert data["role"] == "CRA"
    assert data["has_main_permission"] is True
    assert set(data["missing_prerequisites"]) == {"subjects:read", "sites:read"}
    assert data["can_perform"] is False


@pytest.mark.asyncio
async def test_check_operation_prerequisites_admin():
    """测试检查操作前置权限 - ADMIN角色"""
    data = await check_operation_prerequisites(
        study_id=uuid.uuid4(),
        endpoint_key="subjects:create",
        role="ADMIN",
        db=None,
    )

    assert data["role"] == "ADMIN"
    assert data["has_main_permission"] is True
    assert data["missing_prerequisites"] == []
    assert data["can_perform"] is True


@pytest.mark.asyncio
async def test_check_operation_prerequisites_no_main_permission(
    db_session: AsyncSession
):
    """测试检查操作前置权限 - 没有主权限"""
    study_id = uuid.uuid4()
    # 创建权限：主权限不允许
    main_perm = ApiEndpointPermission(
        study_id=study_id,
        role="CRA",
        endpoint_key="subjects:create",
        allowed=False,
    )
    db_session.add(main_perm)
    await db_session.commit()

    data = await check_operation_prerequisites(
        study_id=study_id,
        endpoint_key="subjects:create",
        role="CRA",
        db=db_session,
    )

    assert data["endpoint_key"] == "subjects:create"
    assert data["role"] == "CRA"
    assert data["has_main_permission"] is False
    assert data["can_perform"] is False