ctms/backend/tests/test_api_permissions_endpoints.py

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

import pytest
import uuid
from fastapi.testclient import TestClient
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.project_permissions import (
    get_api_endpoint_permissions,
    preview_api_permission_prerequisite_adjustments,
    replace_api_endpoint_permissions,
)
from app.models.api_endpoint_permission import ApiEndpointPermission


@pytest.mark.asyncio
async def test_get_api_endpoint_permissions_empty(db_session: AsyncSession):
    """测试获取空的权限矩阵"""
    study_id = uuid.uuid4()

    result = await get_api_endpoint_permissions(db_session, study_id)

    # 应该返回所有角色和端点的默认权限
    assert isinstance(result, dict)
    assert len(result) > 0


@pytest.mark.asyncio
async def test_get_api_endpoint_permissions_with_custom(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 get_api_endpoint_permissions(db_session, study_id)

    assert isinstance(result, dict)
    assert "CRA" in result
    assert "subjects:create" in result["CRA"]
    assert result["CRA"]["subjects:create"]["allowed"] is True


@pytest.mark.asyncio
async def test_replace_api_endpoint_permissions_single_role(db_session: AsyncSession):
    """测试替换单个角色的权限"""
    study_id = uuid.uuid4()

    payload = {
        "CRA": {
            "subjects:create": True,
            "subjects:list": True,
            "subjects:update": True,
        }
    }

    result = await replace_api_endpoint_permissions(db_session, study_id, payload)

    assert isinstance(result, dict)
    assert "CRA" in result
    assert result["CRA"]["subjects:create"]["allowed"] is True
    assert "subjects:list" not in result["CRA"]
    assert result["CRA"]["subjects:read"]["allowed"] is True
    assert result["CRA"]["subjects:update"]["allowed"] is True


@pytest.mark.asyncio
async def test_replace_api_endpoint_permissions_multiple_roles(db_session: AsyncSession):
    """测试替换多个角色的权限"""
    study_id = uuid.uuid4()

    payload = {
        "CRA": {
            "subjects:create": True,
            "subjects:list": True,
        },
        "PV": {
            "subjects:list": True,
            "subjects:read": True,
        }
    }

    result = await replace_api_endpoint_permissions(db_session, study_id, payload)

    assert result["CRA"]["subjects:create"]["allowed"] is True
    assert "subjects:list" not in result["CRA"]
    assert result["CRA"]["subjects:read"]["allowed"] is True
    assert "subjects:list" not in result["PV"]
    assert result["PV"]["subjects:read"]["allowed"] is True


@pytest.mark.asyncio
async def test_replace_api_endpoint_permissions_accepts_legacy_read_alias_denials(db_session: AsyncSession):
    """历史列表读取 key 写入时应归并到当前读取权限，拒绝值不能被默认权限吞掉。"""
    study_id = uuid.uuid4()

    result = await replace_api_endpoint_permissions(
        db_session,
        study_id,
        {
            "CRA": {
                "subjects:list": False,
            }
        },
        confirm_prerequisite_adjustments=True,
    )

    assert "subjects:list" not in result["CRA"]
    assert result["CRA"]["subjects:read"]["allowed"] is False

    stored = await get_api_endpoint_permissions(db_session, study_id)
    assert "subjects:list" not in stored["CRA"]
    assert stored["CRA"]["subjects:read"]["allowed"] is False


@pytest.mark.asyncio
async def test_replace_api_endpoint_permissions_deny(db_session: AsyncSession):
    """测试替换权限为拒绝"""
    study_id = uuid.uuid4()

    payload = {
        "PV": {
            "subjects:create": False,
        }
    }

    result = await replace_api_endpoint_permissions(db_session, study_id, payload)

    assert result["PV"]["subjects:create"]["allowed"] is False


@pytest.mark.asyncio
async def test_replace_api_endpoint_permissions_overwrites_existing(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()

    # 替换权限
    payload = {
        "CRA": {
            "subjects:create": False,
        }
    }

    result = await replace_api_endpoint_permissions(db_session, study_id, payload)

    assert result["CRA"]["subjects:create"]["allowed"] is False


@pytest.mark.asyncio
async def test_replace_api_endpoint_permissions_multiple_endpoints(db_session: AsyncSession):
    """测试替换多个端点的权限"""
    study_id = uuid.uuid4()

    payload = {
        "CRA": {
            "subjects:create": True,
            "subjects:list": True,
            "subjects:update": True,
            "subjects:delete": False,
            "subject_aes:create": True,
            "subject_aes:list": True,
        }
    }

    result = await replace_api_endpoint_permissions(db_session, study_id, payload)

    assert result["CRA"]["subjects:create"]["allowed"] is True
    assert result["CRA"]["subjects:read"]["allowed"] is True
    assert result["CRA"]["subjects:update"]["allowed"] is True
    assert result["CRA"]["subjects:delete"]["allowed"] is False
    assert result["CRA"]["subject_aes:create"]["allowed"] is True
    assert result["CRA"]["subject_aes:read"]["allowed"] is True


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

    payload_1 = {
        "CRA": {
            "subjects:create": True,
        }
    }

    payload_2 = {
        "CRA": {
            "subjects:create": False,
        }
    }

    await replace_api_endpoint_permissions(db_session, study_id_1, payload_1)
    await replace_api_endpoint_permissions(db_session, study_id_2, payload_2)

    result_1 = await get_api_endpoint_permissions(db_session, study_id_1)
    result_2 = await get_api_endpoint_permissions(db_session, study_id_2)

    assert result_1["CRA"]["subjects:create"]["allowed"] is True
    assert result_2["CRA"]["subjects:create"]["allowed"] is False


@pytest.mark.asyncio
async def test_replace_api_endpoint_permissions_empty_payload(db_session: AsyncSession):
    """测试空的权限替换"""
    study_id = uuid.uuid4()

    payload = {}

    result = await replace_api_endpoint_permissions(db_session, study_id, payload)

    # 应该返回空结果或默认权限
    assert isinstance(result, dict)


@pytest.mark.asyncio
async def test_replace_api_endpoint_permissions_partial_update(db_session: AsyncSession):
    """测试部分权限更新"""
    study_id = uuid.uuid4()

    # 创建初始权限
    perm1 = ApiEndpointPermission(
        study_id=study_id,
        role="CRA",
        endpoint_key="subjects:create",
        allowed=True,
    )
    perm2 = ApiEndpointPermission(
        study_id=study_id,
        role="CRA",
        endpoint_key="subjects:list",
        allowed=True,
    )
    db_session.add(perm1)
    db_session.add(perm2)
    await db_session.commit()

    # 只更新一个权限
    payload = {
        "CRA": {
            "subjects:create": False,
        }
    }

    result = await replace_api_endpoint_permissions(db_session, study_id, payload)

    # POST权限应该被更新
    assert result["CRA"]["subjects:create"]["allowed"] is False
    # GET权限应该保持不变
    assert result["CRA"]["subjects:read"]["allowed"] is True


@pytest.mark.asyncio
async def test_replace_api_endpoint_permissions_preserves_unsubmitted_permissions_for_same_role(
    db_session: AsyncSession,
):
    """保存角色的部分权限时，不应清空该角色未提交的权限项。"""
    study_id = uuid.uuid4()

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

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

    assert result["CRA"]["subjects:create"]["allowed"] is False
    assert result["CRA"]["subjects:delete"]["allowed"] is True


@pytest.mark.asyncio
async def test_replace_api_endpoint_permissions_preserves_unsubmitted_roles(db_session: AsyncSession):
    """保存单个角色权限时，不应清空其他角色的已配置权限。"""
    study_id = uuid.uuid4()

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

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

    assert result["CRA"]["subjects:create"]["allowed"] is False
    assert result["PV"]["subjects:create"]["allowed"] is True


@pytest.mark.asyncio
async def test_preview_reports_prerequisites_to_add_for_custom_role(db_session: AsyncSession):
    """授权子权限前应提示需要补齐的前置权限。"""
    study_id = uuid.uuid4()

    preview = await preview_api_permission_prerequisite_adjustments(
        db_session,
        study_id,
        {"DATA_MANAGER": {"subjects:create": True}},
    )

    assert preview["requires_confirmation"] is True
    assert preview["roles"]["DATA_MANAGER"]["add"] == {
        "subjects:create": ["subjects:read", "sites:read"],
    }
    assert preview["roles"]["DATA_MANAGER"]["remove"] == {}


@pytest.mark.asyncio
async def test_replace_rejects_unconfirmed_missing_prerequisites(db_session: AsyncSession):
    """未确认时不应保存缺失前置权限的授权组合。"""
    study_id = uuid.uuid4()

    with pytest.raises(ValueError, match="需要确认前置权限调整"):
        await replace_api_endpoint_permissions(
            db_session,
            study_id,
            {"DATA_MANAGER": {"subjects:create": True}},
        )


@pytest.mark.asyncio
async def test_replace_confirmed_adds_missing_prerequisites(db_session: AsyncSession):
    """确认后应补齐授权子权限需要的前置权限。"""
    study_id = uuid.uuid4()

    result = await replace_api_endpoint_permissions(
        db_session,
        study_id,
        {"DATA_MANAGER": {"subjects:create": True}},
        confirm_prerequisite_adjustments=True,
    )

    assert result["DATA_MANAGER"]["subjects:create"]["allowed"] is True
    assert result["DATA_MANAGER"]["subjects:read"]["allowed"] is True
    assert result["DATA_MANAGER"]["sites:read"]["allowed"] is True


@pytest.mark.asyncio
async def test_replace_confirmed_removes_dependents_when_prerequisite_denied(db_session: AsyncSession):
    """确认取消前置权限时，应消除依赖它的下游权限。"""
    study_id = uuid.uuid4()

    await replace_api_endpoint_permissions(
        db_session,
        study_id,
        {
            "DATA_MANAGER": {
                "subjects:read": True,
                "sites:read": True,
                "subjects:create": True,
            }
        },
        confirm_prerequisite_adjustments=True,
    )

    result = await replace_api_endpoint_permissions(
        db_session,
        study_id,
        {"DATA_MANAGER": {"sites:read": False}},
        confirm_prerequisite_adjustments=True,
    )

    assert result["DATA_MANAGER"]["sites:read"]["allowed"] is False
    assert result["DATA_MANAGER"]["subjects:create"]["allowed"] is False
    assert result["DATA_MANAGER"]["subjects:read"]["allowed"] is True


@pytest.mark.asyncio
async def test_get_api_endpoint_permissions_structure(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 get_api_endpoint_permissions(db_session, study_id)

    # 验证结构
    assert isinstance(result, dict)
    for role, endpoints in result.items():
        assert isinstance(role, str)
        assert isinstance(endpoints, dict)
        for endpoint_key, permission in endpoints.items():
            assert isinstance(endpoint_key, str)
            assert isinstance(permission, dict)
            assert "allowed" in permission
            assert isinstance(permission["allowed"], bool)