Cheng Zhou
ab4f0d93ed
1、合并列表与详情读取权限，统一使用 subjects:read、project_members:read、monitoring_issues:read 等业务读取 key。 2、增加历史读取权限 key 的兼容映射，确保存量项目权限覆盖值不会丢失。 3、同步前端权限工具、路由权限与权限管理页面，避免继续引用已合并的旧权限 key。 4、补充项目角色启停保护与抽屉未保存变更守卫，防止停用 PM 或仍有关联成员的角色。
344 lines
9.9 KiB
Python
344 lines
9.9 KiB
Python
"""集成测试：权限管理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, 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,
|
|
}
|
|
},
|
|
)
|
|
|
|
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_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)
|