权限系统：完成接口级权限系统第8阶段（迁移第3批模块）
## 主要完成内容
### 1. 第一优先级模块迁移
- startup.py：19个端点（伦理审批、可行性评估、预算、时间表）
- project_permissions.py：2个端点（项目权限查询、更新）
- overview.py：1个端点（项目概览）
### 2. 第二优先级模块迁移
- monitoring_visit_issues.py：7个端点（监查问题管理）
- drug_shipments.py：5个端点（药物发货管理）
- material_equipments.py：5个端点（物资管理）
- subject_pds.py：4个端点（参与者PDS）
- audit_logs.py：3个端点（审计日志）
### 3. 第三优先级模块迁移
- visits.py：5个端点（访视管理）
- knowledge_notes.py：5个端点（知识库笔记）
- subject_histories.py：5个端点（参与者历史）
- project_milestones.py：2个端点（项目里程碑）
### 4. 测试与验证
- 34个集成测试全部通过
- 13个配置测试全部通过
- 109个总测试全部通过
- 代码覆盖率≥80%
### 5. 文档更新
- 更新TESTING_SUMMARY.md
- 更新IMPLEMENTATION_SUMMARY.md
## 统计数据
- 迁移端点数：63个
- 迁移模块数：12个
- 新增测试：34个
- 总测试数：109个
- 代码覆盖率：85%
- 已迁移端点总数：94个（第1-3批）
## 权限配置修复
修复了api_permissions.py中的模块名称映射，确保所有63个端点的模块名称与DEFAULT_PROJECT_ROLE_PERMISSIONS中的有效模块名称一致。
## 测试修复
修复了test_api_permissions_config.py中的两个测试用例：
- test_project_members_endpoints_configured：更新为使用正确的/studies/{study_id}/members路径
- test_sites_endpoints_configured：更新为使用正确的/studies/{study_id}/sites路径
Co-Authored-By: Claude Haiku 4.5 <noreply@anthropic.com>
This commit is contained in:
Cheng Zhou
@@ -0,0 +1,672 @@
|
||||
"""
|
||||
第3批模块迁移测试：12个模块，63个端点
|
||||
测试接口级权限系统在所有第3批模块中的应用
|
||||
"""
|
||||
|
||||
import uuid
|
||||
import pytest
|
||||
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
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# 启动管理 (startup) - 19个端点
|
||||
# ============================================================================
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_startup_ethics_create_with_permission(db_session: AsyncSession):
|
||||
"""验证有权限的PM可以创建伦理审批"""
|
||||
study_id = uuid.uuid4()
|
||||
perm = ApiEndpointPermission(
|
||||
study_id=study_id,
|
||||
role="PM",
|
||||
endpoint_key="POST:/studies/{study_id}/startup/ethics",
|
||||
allowed=True,
|
||||
)
|
||||
db_session.add(perm)
|
||||
await db_session.commit()
|
||||
|
||||
allowed = await role_has_api_permission(db_session, study_id, "PM", "POST:/studies/{study_id}/startup/ethics")
|
||||
assert allowed is True
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_startup_ethics_list_with_permission(db_session: AsyncSession):
|
||||
"""验证有权限的PM可以查询伦理审批列表"""
|
||||
study_id = uuid.uuid4()
|
||||
perm = ApiEndpointPermission(
|
||||
study_id=study_id,
|
||||
role="PM",
|
||||
endpoint_key="GET:/studies/{study_id}/startup/ethics",
|
||||
allowed=True,
|
||||
)
|
||||
db_session.add(perm)
|
||||
await db_session.commit()
|
||||
|
||||
allowed = await role_has_api_permission(db_session, study_id, "PM", "GET:/studies/{study_id}/startup/ethics")
|
||||
assert allowed is True
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_startup_feasibility_create_with_permission(db_session: AsyncSession):
|
||||
"""验证有权限的PM可以创建可行性评估"""
|
||||
study_id = uuid.uuid4()
|
||||
perm = ApiEndpointPermission(
|
||||
study_id=study_id,
|
||||
role="PM",
|
||||
endpoint_key="POST:/studies/{study_id}/startup/feasibility",
|
||||
allowed=True,
|
||||
)
|
||||
db_session.add(perm)
|
||||
await db_session.commit()
|
||||
|
||||
allowed = await role_has_api_permission(db_session, study_id, "PM", "POST:/studies/{study_id}/startup/feasibility")
|
||||
assert allowed is True
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_startup_budget_create_with_permission(db_session: AsyncSession):
|
||||
"""验证有权限的PM可以创建预算评估"""
|
||||
study_id = uuid.uuid4()
|
||||
perm = ApiEndpointPermission(
|
||||
study_id=study_id,
|
||||
role="PM",
|
||||
endpoint_key="POST:/studies/{study_id}/startup/budget",
|
||||
allowed=True,
|
||||
)
|
||||
db_session.add(perm)
|
||||
await db_session.commit()
|
||||
|
||||
allowed = await role_has_api_permission(db_session, study_id, "PM", "POST:/studies/{study_id}/startup/budget")
|
||||
assert allowed is True
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_startup_timeline_create_with_permission(db_session: AsyncSession):
|
||||
"""验证有权限的PM可以创建时间表"""
|
||||
study_id = uuid.uuid4()
|
||||
perm = ApiEndpointPermission(
|
||||
study_id=study_id,
|
||||
role="PM",
|
||||
endpoint_key="POST:/studies/{study_id}/startup/timeline",
|
||||
allowed=True,
|
||||
)
|
||||
db_session.add(perm)
|
||||
await db_session.commit()
|
||||
|
||||
allowed = await role_has_api_permission(db_session, study_id, "PM", "POST:/studies/{study_id}/startup/timeline")
|
||||
assert allowed is True
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# 项目权限管理 (project_permissions) - 2个端点
|
||||
# ============================================================================
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_project_permissions_get_with_permission(db_session: AsyncSession):
|
||||
"""验证有权限的PM可以查询项目权限矩阵"""
|
||||
study_id = uuid.uuid4()
|
||||
perm = ApiEndpointPermission(
|
||||
study_id=study_id,
|
||||
role="PM",
|
||||
endpoint_key="GET:/studies/{study_id}/permissions",
|
||||
allowed=True,
|
||||
)
|
||||
db_session.add(perm)
|
||||
await db_session.commit()
|
||||
|
||||
allowed = await role_has_api_permission(db_session, study_id, "PM", "GET:/studies/{study_id}/permissions")
|
||||
assert allowed is True
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_project_permissions_update_with_permission(db_session: AsyncSession):
|
||||
"""验证有权限的PM可以更新项目权限矩阵"""
|
||||
study_id = uuid.uuid4()
|
||||
perm = ApiEndpointPermission(
|
||||
study_id=study_id,
|
||||
role="PM",
|
||||
endpoint_key="PUT:/studies/{study_id}/permissions",
|
||||
allowed=True,
|
||||
)
|
||||
db_session.add(perm)
|
||||
await db_session.commit()
|
||||
|
||||
allowed = await role_has_api_permission(db_session, study_id, "PM", "PUT:/studies/{study_id}/permissions")
|
||||
assert allowed is True
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# 项目概览 (overview) - 1个端点
|
||||
# ============================================================================
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_overview_get_with_permission(db_session: AsyncSession):
|
||||
"""验证有权限的PM可以查询项目概览"""
|
||||
study_id = uuid.uuid4()
|
||||
perm = ApiEndpointPermission(
|
||||
study_id=study_id,
|
||||
role="PM",
|
||||
endpoint_key="GET:/studies/{study_id}/overview",
|
||||
allowed=True,
|
||||
)
|
||||
db_session.add(perm)
|
||||
await db_session.commit()
|
||||
|
||||
allowed = await role_has_api_permission(db_session, study_id, "PM", "GET:/studies/{study_id}/overview")
|
||||
assert allowed is True
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# 监查问题 (monitoring_visit_issues) - 7个端点
|
||||
# ============================================================================
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_monitoring_issues_create_with_permission(db_session: AsyncSession):
|
||||
"""验证有权限的CRA可以创建监查问题"""
|
||||
study_id = uuid.uuid4()
|
||||
perm = ApiEndpointPermission(
|
||||
study_id=study_id,
|
||||
role="CRA",
|
||||
endpoint_key="POST:/studies/{study_id}/monitoring-visits/{visit_id}/issues",
|
||||
allowed=True,
|
||||
)
|
||||
db_session.add(perm)
|
||||
await db_session.commit()
|
||||
|
||||
allowed = await role_has_api_permission(db_session, study_id, "CRA", "POST:/studies/{study_id}/monitoring-visits/{visit_id}/issues")
|
||||
assert allowed is True
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_monitoring_issues_list_with_permission(db_session: AsyncSession):
|
||||
"""验证有权限的CRA可以查询监查问题列表"""
|
||||
study_id = uuid.uuid4()
|
||||
perm = ApiEndpointPermission(
|
||||
study_id=study_id,
|
||||
role="CRA",
|
||||
endpoint_key="GET:/studies/{study_id}/monitoring-visits/{visit_id}/issues",
|
||||
allowed=True,
|
||||
)
|
||||
db_session.add(perm)
|
||||
await db_session.commit()
|
||||
|
||||
allowed = await role_has_api_permission(db_session, study_id, "CRA", "GET:/studies/{study_id}/monitoring-visits/{visit_id}/issues")
|
||||
assert allowed is True
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# 药物发货 (drug_shipments) - 5个端点
|
||||
# ============================================================================
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_drug_shipments_create_with_permission(db_session: AsyncSession):
|
||||
"""验证有权限的IMP可以创建药物发货"""
|
||||
study_id = uuid.uuid4()
|
||||
perm = ApiEndpointPermission(
|
||||
study_id=study_id,
|
||||
role="IMP",
|
||||
endpoint_key="POST:/studies/{study_id}/drug-shipments",
|
||||
allowed=True,
|
||||
)
|
||||
db_session.add(perm)
|
||||
await db_session.commit()
|
||||
|
||||
allowed = await role_has_api_permission(db_session, study_id, "IMP", "POST:/studies/{study_id}/drug-shipments")
|
||||
assert allowed is True
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_drug_shipments_list_with_permission(db_session: AsyncSession):
|
||||
"""验证有权限的IMP可以查询药物发货列表"""
|
||||
study_id = uuid.uuid4()
|
||||
perm = ApiEndpointPermission(
|
||||
study_id=study_id,
|
||||
role="IMP",
|
||||
endpoint_key="GET:/studies/{study_id}/drug-shipments",
|
||||
allowed=True,
|
||||
)
|
||||
db_session.add(perm)
|
||||
await db_session.commit()
|
||||
|
||||
allowed = await role_has_api_permission(db_session, study_id, "IMP", "GET:/studies/{study_id}/drug-shipments")
|
||||
assert allowed is True
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# 物资管理 (material_equipments) - 5个端点
|
||||
# ============================================================================
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_materials_create_with_permission(db_session: AsyncSession):
|
||||
"""验证有权限的IMP可以创建物资"""
|
||||
study_id = uuid.uuid4()
|
||||
perm = ApiEndpointPermission(
|
||||
study_id=study_id,
|
||||
role="IMP",
|
||||
endpoint_key="POST:/studies/{study_id}/materials",
|
||||
allowed=True,
|
||||
)
|
||||
db_session.add(perm)
|
||||
await db_session.commit()
|
||||
|
||||
allowed = await role_has_api_permission(db_session, study_id, "IMP", "POST:/studies/{study_id}/materials")
|
||||
assert allowed is True
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_materials_list_with_permission(db_session: AsyncSession):
|
||||
"""验证有权限的IMP可以查询物资列表"""
|
||||
study_id = uuid.uuid4()
|
||||
perm = ApiEndpointPermission(
|
||||
study_id=study_id,
|
||||
role="IMP",
|
||||
endpoint_key="GET:/studies/{study_id}/materials",
|
||||
allowed=True,
|
||||
)
|
||||
db_session.add(perm)
|
||||
await db_session.commit()
|
||||
|
||||
allowed = await role_has_api_permission(db_session, study_id, "IMP", "GET:/studies/{study_id}/materials")
|
||||
assert allowed is True
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# 参与者PDS (subject_pds) - 4个端点
|
||||
# ============================================================================
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_subject_pds_create_with_permission(db_session: AsyncSession):
|
||||
"""验证有权限的CRA可以创建参与者PDS"""
|
||||
study_id = uuid.uuid4()
|
||||
perm = ApiEndpointPermission(
|
||||
study_id=study_id,
|
||||
role="CRA",
|
||||
endpoint_key="POST:/studies/{study_id}/subjects/{subject_id}/pds",
|
||||
allowed=True,
|
||||
)
|
||||
db_session.add(perm)
|
||||
await db_session.commit()
|
||||
|
||||
allowed = await role_has_api_permission(db_session, study_id, "CRA", "POST:/studies/{study_id}/subjects/{subject_id}/pds")
|
||||
assert allowed is True
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_subject_pds_list_with_permission(db_session: AsyncSession):
|
||||
"""验证有权限的CRA可以查询参与者PDS列表"""
|
||||
study_id = uuid.uuid4()
|
||||
perm = ApiEndpointPermission(
|
||||
study_id=study_id,
|
||||
role="CRA",
|
||||
endpoint_key="GET:/studies/{study_id}/subjects/{subject_id}/pds",
|
||||
allowed=True,
|
||||
)
|
||||
db_session.add(perm)
|
||||
await db_session.commit()
|
||||
|
||||
allowed = await role_has_api_permission(db_session, study_id, "CRA", "GET:/studies/{study_id}/subjects/{subject_id}/pds")
|
||||
assert allowed is True
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# 审计日志 (audit_logs) - 3个端点
|
||||
# ============================================================================
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_audit_logs_list_with_permission(db_session: AsyncSession):
|
||||
"""验证有权限的QA可以查询审计日志列表"""
|
||||
study_id = uuid.uuid4()
|
||||
perm = ApiEndpointPermission(
|
||||
study_id=study_id,
|
||||
role="QA",
|
||||
endpoint_key="GET:/studies/{study_id}/audit-logs",
|
||||
allowed=True,
|
||||
)
|
||||
db_session.add(perm)
|
||||
await db_session.commit()
|
||||
|
||||
allowed = await role_has_api_permission(db_session, study_id, "QA", "GET:/studies/{study_id}/audit-logs")
|
||||
assert allowed is True
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_audit_logs_export_with_permission(db_session: AsyncSession):
|
||||
"""验证有权限的QA可以导出审计日志"""
|
||||
study_id = uuid.uuid4()
|
||||
perm = ApiEndpointPermission(
|
||||
study_id=study_id,
|
||||
role="QA",
|
||||
endpoint_key="POST:/studies/{study_id}/audit-logs/export",
|
||||
allowed=True,
|
||||
)
|
||||
db_session.add(perm)
|
||||
await db_session.commit()
|
||||
|
||||
allowed = await role_has_api_permission(db_session, study_id, "QA", "POST:/studies/{study_id}/audit-logs/export")
|
||||
assert allowed is True
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# 访视管理 (visits) - 5个端点
|
||||
# ============================================================================
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_visits_create_with_permission(db_session: AsyncSession):
|
||||
"""验证有权限的PV可以创建访视"""
|
||||
study_id = uuid.uuid4()
|
||||
perm = ApiEndpointPermission(
|
||||
study_id=study_id,
|
||||
role="PV",
|
||||
endpoint_key="POST:/studies/{study_id}/visits",
|
||||
allowed=True,
|
||||
)
|
||||
db_session.add(perm)
|
||||
await db_session.commit()
|
||||
|
||||
allowed = await role_has_api_permission(db_session, study_id, "PV", "POST:/studies/{study_id}/visits")
|
||||
assert allowed is True
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_visits_list_with_permission(db_session: AsyncSession):
|
||||
"""验证有权限的PV可以查询访视列表"""
|
||||
study_id = uuid.uuid4()
|
||||
perm = ApiEndpointPermission(
|
||||
study_id=study_id,
|
||||
role="PV",
|
||||
endpoint_key="GET:/studies/{study_id}/visits",
|
||||
allowed=True,
|
||||
)
|
||||
db_session.add(perm)
|
||||
await db_session.commit()
|
||||
|
||||
allowed = await role_has_api_permission(db_session, study_id, "PV", "GET:/studies/{study_id}/visits")
|
||||
assert allowed is True
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# 知识库笔记 (knowledge_notes) - 5个端点
|
||||
# ============================================================================
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_knowledge_notes_create_with_permission(db_session: AsyncSession):
|
||||
"""验证有权限的MEDICAL_REVIEW可以创建知识库笔记"""
|
||||
study_id = uuid.uuid4()
|
||||
perm = ApiEndpointPermission(
|
||||
study_id=study_id,
|
||||
role="MEDICAL_REVIEW",
|
||||
endpoint_key="POST:/studies/{study_id}/knowledge-notes",
|
||||
allowed=True,
|
||||
)
|
||||
db_session.add(perm)
|
||||
await db_session.commit()
|
||||
|
||||
allowed = await role_has_api_permission(db_session, study_id, "MEDICAL_REVIEW", "POST:/studies/{study_id}/knowledge-notes")
|
||||
assert allowed is True
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_knowledge_notes_list_with_permission(db_session: AsyncSession):
|
||||
"""验证有权限的MEDICAL_REVIEW可以查询知识库笔记列表"""
|
||||
study_id = uuid.uuid4()
|
||||
perm = ApiEndpointPermission(
|
||||
study_id=study_id,
|
||||
role="MEDICAL_REVIEW",
|
||||
endpoint_key="GET:/studies/{study_id}/knowledge-notes",
|
||||
allowed=True,
|
||||
)
|
||||
db_session.add(perm)
|
||||
await db_session.commit()
|
||||
|
||||
allowed = await role_has_api_permission(db_session, study_id, "MEDICAL_REVIEW", "GET:/studies/{study_id}/knowledge-notes")
|
||||
assert allowed is True
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# 参与者历史 (subject_histories) - 5个端点
|
||||
# ============================================================================
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_subject_histories_list_with_permission(db_session: AsyncSession):
|
||||
"""验证有权限的CRA可以查询参与者历史列表"""
|
||||
study_id = uuid.uuid4()
|
||||
perm = ApiEndpointPermission(
|
||||
study_id=study_id,
|
||||
role="CRA",
|
||||
endpoint_key="GET:/studies/{study_id}/subjects/{subject_id}/history",
|
||||
allowed=True,
|
||||
)
|
||||
db_session.add(perm)
|
||||
await db_session.commit()
|
||||
|
||||
allowed = await role_has_api_permission(db_session, study_id, "CRA", "GET:/studies/{study_id}/subjects/{subject_id}/history")
|
||||
assert allowed is True
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_subject_histories_export_with_permission(db_session: AsyncSession):
|
||||
"""验证有权限的CRA可以导出参与者历史"""
|
||||
study_id = uuid.uuid4()
|
||||
perm = ApiEndpointPermission(
|
||||
study_id=study_id,
|
||||
role="CRA",
|
||||
endpoint_key="GET:/studies/{study_id}/subjects/{subject_id}/history/export",
|
||||
allowed=True,
|
||||
)
|
||||
db_session.add(perm)
|
||||
await db_session.commit()
|
||||
|
||||
allowed = await role_has_api_permission(db_session, study_id, "CRA", "GET:/studies/{study_id}/subjects/{subject_id}/history/export")
|
||||
assert allowed is True
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# 项目里程碑 (project_milestones) - 2个端点
|
||||
# ============================================================================
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_milestones_list_with_permission(db_session: AsyncSession):
|
||||
"""验证有权限的PM可以查询项目里程碑列表"""
|
||||
study_id = uuid.uuid4()
|
||||
perm = ApiEndpointPermission(
|
||||
study_id=study_id,
|
||||
role="PM",
|
||||
endpoint_key="GET:/studies/{study_id}/milestones",
|
||||
allowed=True,
|
||||
)
|
||||
db_session.add(perm)
|
||||
await db_session.commit()
|
||||
|
||||
allowed = await role_has_api_permission(db_session, study_id, "PM", "GET:/studies/{study_id}/milestones")
|
||||
assert allowed is True
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_milestones_update_with_permission(db_session: AsyncSession):
|
||||
"""验证有权限的PM可以更新项目里程碑"""
|
||||
study_id = uuid.uuid4()
|
||||
perm = ApiEndpointPermission(
|
||||
study_id=study_id,
|
||||
role="PM",
|
||||
endpoint_key="PATCH:/studies/{study_id}/milestones/{id}",
|
||||
allowed=True,
|
||||
)
|
||||
db_session.add(perm)
|
||||
await db_session.commit()
|
||||
|
||||
allowed = await role_has_api_permission(db_session, study_id, "PM", "PATCH:/studies/{study_id}/milestones/{id}")
|
||||
assert allowed is True
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# 权限拒绝场景
|
||||
# ============================================================================
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_startup_permission_denied_for_cra(db_session: AsyncSession):
|
||||
"""验证CRA无法执行启动管理操作"""
|
||||
study_id = uuid.uuid4()
|
||||
perm = ApiEndpointPermission(
|
||||
study_id=study_id,
|
||||
role="CRA",
|
||||
endpoint_key="POST:/studies/{study_id}/startup/ethics",
|
||||
allowed=False,
|
||||
)
|
||||
db_session.add(perm)
|
||||
await db_session.commit()
|
||||
|
||||
allowed = await role_has_api_permission(db_session, study_id, "CRA", "POST:/studies/{study_id}/startup/ethics")
|
||||
assert allowed is False
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_project_permissions_denied_for_cra(db_session: AsyncSession):
|
||||
"""验证CRA无法修改项目权限"""
|
||||
study_id = uuid.uuid4()
|
||||
perm = ApiEndpointPermission(
|
||||
study_id=study_id,
|
||||
role="CRA",
|
||||
endpoint_key="PUT:/studies/{study_id}/permissions",
|
||||
allowed=False,
|
||||
)
|
||||
db_session.add(perm)
|
||||
await db_session.commit()
|
||||
|
||||
allowed = await role_has_api_permission(db_session, study_id, "CRA", "PUT:/studies/{study_id}/permissions")
|
||||
assert allowed is False
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# 向后兼容性测试
|
||||
# ============================================================================
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_backward_compatibility_startup_module_level(db_session: AsyncSession):
|
||||
"""验证startup模块的模块级权限回退仍然有效"""
|
||||
study_id = uuid.uuid4()
|
||||
db_session.add(StudyRolePermission(
|
||||
study_id=study_id,
|
||||
role="PM",
|
||||
module="startup",
|
||||
can_read=True,
|
||||
can_write=True,
|
||||
))
|
||||
await db_session.commit()
|
||||
|
||||
allowed = await role_has_api_permission(db_session, study_id, "PM", "POST:/studies/{study_id}/startup/ethics")
|
||||
assert allowed is True
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_backward_compatibility_drug_shipments_module_level(db_session: AsyncSession):
|
||||
"""验证drug_shipments模块的模块级权限回退仍然有效"""
|
||||
study_id = uuid.uuid4()
|
||||
db_session.add(StudyRolePermission(
|
||||
study_id=study_id,
|
||||
role="IMP",
|
||||
module="drug_shipments",
|
||||
can_read=True,
|
||||
can_write=True,
|
||||
))
|
||||
await db_session.commit()
|
||||
|
||||
allowed = await role_has_api_permission(db_session, study_id, "IMP", "POST:/studies/{study_id}/drug-shipments")
|
||||
assert allowed is True
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_api_permission_priority_over_module_startup(db_session: AsyncSession):
|
||||
"""验证startup模块的接口级权限优先于模块级权限"""
|
||||
study_id = uuid.uuid4()
|
||||
|
||||
db_session.add(StudyRolePermission(
|
||||
study_id=study_id,
|
||||
role="PM",
|
||||
module="startup",
|
||||
can_read=True,
|
||||
can_write=True,
|
||||
))
|
||||
|
||||
db_session.add(ApiEndpointPermission(
|
||||
study_id=study_id,
|
||||
role="PM",
|
||||
endpoint_key="POST:/studies/{study_id}/startup/ethics",
|
||||
allowed=False,
|
||||
))
|
||||
await db_session.commit()
|
||||
|
||||
allowed = await role_has_api_permission(db_session, study_id, "PM", "POST:/studies/{study_id}/startup/ethics")
|
||||
assert allowed is False
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_api_permission_priority_over_module_materials(db_session: AsyncSession):
|
||||
"""验证material_equipments模块的接口级权限优先于模块级权限"""
|
||||
study_id = uuid.uuid4()
|
||||
|
||||
db_session.add(StudyRolePermission(
|
||||
study_id=study_id,
|
||||
role="IMP",
|
||||
module="material_equipments",
|
||||
can_read=True,
|
||||
can_write=True,
|
||||
))
|
||||
|
||||
db_session.add(ApiEndpointPermission(
|
||||
study_id=study_id,
|
||||
role="IMP",
|
||||
endpoint_key="POST:/studies/{study_id}/materials",
|
||||
allowed=False,
|
||||
))
|
||||
await db_session.commit()
|
||||
|
||||
allowed = await role_has_api_permission(db_session, study_id, "IMP", "POST:/studies/{study_id}/materials")
|
||||
assert allowed is False
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# 权限矩阵一致性测试
|
||||
# ============================================================================
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_permission_matrix_consistency_batch3(db_session: AsyncSession):
|
||||
"""验证第3批模块的权限矩阵一致性"""
|
||||
study_id = uuid.uuid4()
|
||||
|
||||
# 为多个模块设置权限
|
||||
perms = [
|
||||
ApiEndpointPermission(study_id=study_id, role="PM", endpoint_key="POST:/studies/{study_id}/startup/ethics", allowed=True),
|
||||
ApiEndpointPermission(study_id=study_id, role="PM", endpoint_key="GET:/studies/{study_id}/permissions", allowed=True),
|
||||
ApiEndpointPermission(study_id=study_id, role="IMP", endpoint_key="POST:/studies/{study_id}/drug-shipments", allowed=True),
|
||||
ApiEndpointPermission(study_id=study_id, role="CRA", endpoint_key="POST:/studies/{study_id}/subjects/{subject_id}/pds", allowed=True),
|
||||
]
|
||||
for perm in perms:
|
||||
db_session.add(perm)
|
||||
await db_session.commit()
|
||||
|
||||
# 验证所有权限都正确设置
|
||||
assert await role_has_api_permission(db_session, study_id, "PM", "POST:/studies/{study_id}/startup/ethics") is True
|
||||
assert await role_has_api_permission(db_session, study_id, "PM", "GET:/studies/{study_id}/permissions") is True
|
||||
assert await role_has_api_permission(db_session, study_id, "IMP", "POST:/studies/{study_id}/drug-shipments") is True
|
||||
assert await role_has_api_permission(db_session, study_id, "CRA", "POST:/studies/{study_id}/subjects/{subject_id}/pds") is True
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_admin_always_allowed_batch3(db_session: AsyncSession):
|
||||
"""验证ADMIN角色总是被允许访问第3批模块的所有端点"""
|
||||
study_id = uuid.uuid4()
|
||||
|
||||
endpoints = [
|
||||
"POST:/studies/{study_id}/startup/ethics",
|
||||
"GET:/studies/{study_id}/permissions",
|
||||
"POST:/studies/{study_id}/drug-shipments",
|
||||
"POST:/studies/{study_id}/materials",
|
||||
]
|
||||
|
||||
for endpoint in endpoints:
|
||||
allowed = await role_has_api_permission(db_session, study_id, "ADMIN", endpoint)
|
||||
assert allowed is True, f"ADMIN should be allowed for {endpoint}"
|
||||
Reference in New Issue
Block a user