ctms/backend/tests/test_api_permissions.py

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

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

from app.core.project_permissions import role_has_api_permission
from app.models.api_endpoint_permission import ApiEndpointPermission
from app.models.study_role_permission import StudyRolePermission


@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


@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_api_permission_fallback_to_module_level(db_session: AsyncSession):
    """测试权限回退 - 接口级权限未配置时回退到模块级权限"""
    study_id = uuid.uuid4()

    # 创建模块级权限
    module_perm = StudyRolePermission(
        study_id=study_id,
        role="CRA",
        module="subjects",
        can_write=True,
    )
    db_session.add(module_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


@pytest.mark.asyncio
async def test_api_permission_fallback_denied(db_session: AsyncSession):
    """测试权限回退 - 模块级权限被拒绝"""
    study_id = uuid.uuid4()

    # 创建模块级权限（拒绝）
    module_perm = StudyRolePermission(
        study_id=study_id,
        role="PV",
        module="subjects",
        can_write=False,
    )
    db_session.add(module_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_priority_over_module(db_session: AsyncSession):
    """测试接口级权限优先于模块级权限"""
    study_id = uuid.uuid4()

    # 创建模块级权限（允许）
    module_perm = StudyRolePermission(
        study_id=study_id,
        role="CRA",
        module="subjects",
        can_write=True,
    )
    db_session.add(module_perm)

    # 创建接口级权限（拒绝）- 应该优先使用这个
    api_perm = ApiEndpointPermission(
        study_id=study_id,
        role="CRA",
        endpoint_key="subjects:create",
        allowed=False,
    )
    db_session.add(api_perm)
    await db_session.commit()

    # 应该返回接口级权限的结果（False）（禁用前置权限检查）
    result = await role_has_api_permission(
        db_session, study_id, "CRA", "subjects:create", check_prerequisites=False
    )
    assert result is False


@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