统一项目角色与接口权限配置
This commit is contained in:
Cheng Zhou
@@ -7,7 +7,7 @@ import ast
|
||||
from sqlalchemy import delete
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from app.core.api_permissions import API_ENDPOINT_PERMISSIONS, PROJECT_PERMISSION_ROLES
|
||||
from app.core.api_permissions import API_ENDPOINT_PERMISSIONS, OPERATION_TO_ENDPOINTS, PROJECT_PERMISSION_ROLES
|
||||
from app.core.project_permissions import role_has_api_permission
|
||||
from app.core.project_permissions import get_api_endpoint_permissions, replace_api_endpoint_permissions
|
||||
from app.core.permission_cache import PermissionCache, set_permission_cache
|
||||
@@ -61,6 +61,326 @@ def test_all_backend_permission_guards_are_configurable():
|
||||
assert used_keys <= set(API_ENDPOINT_PERMISSIONS)
|
||||
|
||||
|
||||
def test_subject_history_permissions_are_labeled_as_medical_history_under_subjects():
|
||||
"""病史记录权限应归入参与者管理，而不是独立的历史模块。"""
|
||||
expected_descriptions = {
|
||||
"subject_histories:create": "创建病史记录",
|
||||
"subject_histories:list": "查询病史记录列表",
|
||||
"subject_histories:read": "查询病史记录详情",
|
||||
"subject_histories:update": "更新病史记录",
|
||||
"subject_histories:delete": "删除病史记录",
|
||||
}
|
||||
|
||||
for endpoint_key, description in expected_descriptions.items():
|
||||
assert API_ENDPOINT_PERMISSIONS[endpoint_key]["module"] == "subjects"
|
||||
assert API_ENDPOINT_PERMISSIONS[endpoint_key]["description"] == description
|
||||
|
||||
|
||||
def test_contract_fee_permissions_do_not_include_legacy_finance_contracts():
|
||||
"""旧合同基础信息权限应从当前权限矩阵移除。"""
|
||||
for endpoint_key in API_ENDPOINT_PERMISSIONS:
|
||||
assert not endpoint_key.startswith("finance_contracts:")
|
||||
assert not endpoint_key.startswith("fees_payments:")
|
||||
assert not endpoint_key.startswith("fees_attachments:")
|
||||
|
||||
expected_descriptions = {
|
||||
"fees_contracts:create": "创建合同费用",
|
||||
"fees_contracts:list": "查询合同费用列表",
|
||||
"fees_contracts:read": "查询合同费用详情",
|
||||
"fees_contracts:update": "更新合同费用",
|
||||
"fees_contracts:delete": "删除合同费用",
|
||||
}
|
||||
|
||||
for endpoint_key, description in expected_descriptions.items():
|
||||
assert API_ENDPOINT_PERMISSIONS[endpoint_key]["module"] == "fees"
|
||||
assert API_ENDPOINT_PERMISSIONS[endpoint_key]["description"] == description
|
||||
|
||||
|
||||
def test_generic_attachment_permissions_are_removed_from_matrix():
|
||||
"""前端和模板迁移后，项目权限矩阵不再暴露通用 attachments:*。"""
|
||||
assert not any(endpoint_key.startswith("attachments:") for endpoint_key in API_ENDPOINT_PERMISSIONS)
|
||||
|
||||
|
||||
def test_module_attachment_permissions_are_available():
|
||||
"""模块附件权限应接入权限矩阵。"""
|
||||
expected_keys = {
|
||||
"fees_contracts_attachments:create",
|
||||
"fees_contracts_attachments:read",
|
||||
"fees_contracts_attachments:delete",
|
||||
"startup_initiation_attachments:create",
|
||||
"startup_initiation_attachments:read",
|
||||
"startup_initiation_attachments:delete",
|
||||
"startup_ethics_attachments:create",
|
||||
"startup_ethics_attachments:read",
|
||||
"startup_ethics_attachments:delete",
|
||||
"startup_auth_attachments:create",
|
||||
"startup_auth_attachments:read",
|
||||
"startup_auth_attachments:delete",
|
||||
"drug_shipments_attachments:create",
|
||||
"drug_shipments_attachments:read",
|
||||
"drug_shipments_attachments:delete",
|
||||
"precautions_attachments:create",
|
||||
"precautions_attachments:read",
|
||||
"precautions_attachments:delete",
|
||||
"faq_attachments:create",
|
||||
"faq_attachments:read",
|
||||
"faq_attachments:delete",
|
||||
}
|
||||
|
||||
assert expected_keys <= set(API_ENDPOINT_PERMISSIONS)
|
||||
|
||||
|
||||
def test_precautions_permissions_use_business_aligned_keys():
|
||||
"""注意事项权限应使用业务一致的英文 key，并归入共享库父模块。"""
|
||||
expected_descriptions = {
|
||||
"precautions:create": "创建注意事项",
|
||||
"precautions:list": "查询注意事项列表",
|
||||
"precautions:read": "查询注意事项详情",
|
||||
"precautions:update": "更新注意事项",
|
||||
"precautions:delete": "删除注意事项",
|
||||
}
|
||||
|
||||
for endpoint_key, description in expected_descriptions.items():
|
||||
assert API_ENDPOINT_PERMISSIONS[endpoint_key]["module"] == "shared_library"
|
||||
assert API_ENDPOINT_PERMISSIONS[endpoint_key]["description"] == description
|
||||
|
||||
assert not any(key.startswith("knowledge_notes:") for key in API_ENDPOINT_PERMISSIONS)
|
||||
assert not any(key.startswith("knowledge_notes_attachments:") for key in API_ENDPOINT_PERMISSIONS)
|
||||
|
||||
|
||||
def test_faq_and_precautions_are_shared_library_sibling_sections():
|
||||
"""医学咨询与注意事项应同属共享库，由前端模块细分区分。"""
|
||||
shared_library_keys = OPERATION_TO_ENDPOINTS["shared_library"]
|
||||
|
||||
assert "faq:read" in shared_library_keys["read"]
|
||||
assert "precautions:read" in shared_library_keys["read"]
|
||||
assert "faq:create" in shared_library_keys["write"]
|
||||
assert "precautions:create" in shared_library_keys["write"]
|
||||
assert API_ENDPOINT_PERMISSIONS["faq:read"]["module"] == "shared_library"
|
||||
assert API_ENDPOINT_PERMISSIONS["precautions:read"]["module"] == "shared_library"
|
||||
|
||||
|
||||
def test_precautions_runtime_code_uses_business_entity_names():
|
||||
"""运行时代码不应继续使用 knowledge_note 作为注意事项实体命名。"""
|
||||
backend_root = Path(__file__).resolve().parents[1] / "app"
|
||||
ignored = {"__pycache__"}
|
||||
combined_source = "\n".join(
|
||||
path.read_text(encoding="utf-8")
|
||||
for path in backend_root.rglob("*.py")
|
||||
if not ignored.intersection(path.parts)
|
||||
)
|
||||
|
||||
assert "KnowledgeNote" not in combined_source
|
||||
assert "knowledge_note" not in combined_source
|
||||
assert "knowledge_notes" not in combined_source
|
||||
assert "Precaution" in combined_source
|
||||
|
||||
|
||||
def test_legacy_fee_attachment_router_is_not_registered():
|
||||
"""费用附件应统一走通用附件 API，不再注册专用 /fees/attachments 路由。"""
|
||||
router_source = (Path(__file__).resolve().parents[1] / "app" / "api" / "v1" / "router.py").read_text()
|
||||
|
||||
assert "fees_attachments" not in router_source
|
||||
|
||||
|
||||
def test_attachment_permissions_use_module_permissions():
|
||||
"""附件鉴权应使用模块附件权限，不再回退通用 attachments:*。"""
|
||||
attachments_path = Path(__file__).resolve().parents[1] / "app" / "api" / "v1" / "attachments.py"
|
||||
source = attachments_path.read_text()
|
||||
|
||||
expected_entity_modules = {
|
||||
'"contract_fee_contract": "fees_contracts_attachments"',
|
||||
'"contract_fee_voucher": "fees_contracts_attachments"',
|
||||
'"contract_fee_invoice": "fees_contracts_attachments"',
|
||||
'"startup_feasibility": "startup_initiation_attachments"',
|
||||
'"startup_ethics": "startup_ethics_attachments"',
|
||||
'"startup_kickoff": "startup_auth_attachments"',
|
||||
'"startup_kickoff_minutes": "startup_auth_attachments"',
|
||||
'"startup_kickoff_signin": "startup_auth_attachments"',
|
||||
'"startup_kickoff_ppt": "startup_auth_attachments"',
|
||||
'"training_authorization": "startup_auth_attachments"',
|
||||
'"drug_shipment": "drug_shipments_attachments"',
|
||||
'"precaution": "precautions_attachments"',
|
||||
'"faq_replies": "faq_attachments"',
|
||||
}
|
||||
|
||||
for expected in expected_entity_modules:
|
||||
assert expected in source
|
||||
|
||||
assert 'f"{module_prefix}:{operation}"' in source
|
||||
assert 'f"attachments:{operation}"' not in source
|
||||
assert "role_has_api_permission(" in source
|
||||
assert "parent_permission" in source
|
||||
assert source.count("_ensure_attachment_permission(") >= 8
|
||||
|
||||
|
||||
def test_attachment_parent_permissions_cover_business_modules():
|
||||
"""已知附件实体必须叠加对应父模块权限，避免只靠附件权限越权读取业务数据。"""
|
||||
attachments_path = Path(__file__).resolve().parents[1] / "app" / "api" / "v1" / "attachments.py"
|
||||
source = attachments_path.read_text()
|
||||
|
||||
expected_parent_permissions = {
|
||||
'"fees_contracts:read" if action == "read" else "fees_contracts:update"',
|
||||
'return f"startup_initiation:{parent_action}"',
|
||||
'return f"startup_ethics:{parent_action}"',
|
||||
'"startup_auth:read" if action == "read" else "startup_auth:update"',
|
||||
'"drug_shipments:read" if action == "read" else "drug_shipments:update"',
|
||||
'"precautions:read" if action == "read" else "precautions:update"',
|
||||
'return "faq:read"',
|
||||
'return "faq_reply:delete"',
|
||||
'return "faq_reply:create"',
|
||||
}
|
||||
|
||||
for expected in expected_parent_permissions:
|
||||
assert expected in source
|
||||
|
||||
|
||||
def test_startup_ethics_permissions_use_visible_business_language():
|
||||
"""立项与伦理权限矩阵应使用前台一致的业务名称。"""
|
||||
expected_descriptions = {
|
||||
"startup_initiation:create": "创建立项记录",
|
||||
"startup_initiation:list": "查询立项记录列表",
|
||||
"startup_initiation:read": "查询立项记录详情",
|
||||
"startup_initiation:update": "更新立项记录",
|
||||
"startup_initiation:delete": "删除立项记录",
|
||||
"startup_ethics:create": "创建伦理记录",
|
||||
"startup_ethics:list": "查询伦理记录列表",
|
||||
"startup_ethics:read": "查询伦理记录详情",
|
||||
"startup_ethics:update": "更新伦理记录",
|
||||
"startup_ethics:delete": "删除伦理记录",
|
||||
}
|
||||
|
||||
assert not any(key.startswith("feasibility:") for key in API_ENDPOINT_PERMISSIONS)
|
||||
assert not any(key.startswith("ethics:") for key in API_ENDPOINT_PERMISSIONS)
|
||||
for endpoint_key, description in expected_descriptions.items():
|
||||
assert API_ENDPOINT_PERMISSIONS[endpoint_key]["module"] == "startup_ethics"
|
||||
assert API_ENDPOINT_PERMISSIONS[endpoint_key]["description"] == description
|
||||
|
||||
|
||||
def test_startup_auth_permissions_only_include_wired_operations():
|
||||
"""启动与授权权限矩阵只应展示实际路由使用的操作。"""
|
||||
stale_keys = {
|
||||
"budget:create",
|
||||
"budget:list",
|
||||
"budget:read",
|
||||
"budget:update",
|
||||
"budget:delete",
|
||||
"timeline:create",
|
||||
"timeline:list",
|
||||
"timeline:read",
|
||||
"timeline:update",
|
||||
"timeline:delete",
|
||||
}
|
||||
|
||||
assert stale_keys.isdisjoint(API_ENDPOINT_PERMISSIONS)
|
||||
for endpoint_keys in OPERATION_TO_ENDPOINTS["startup_auth"].values():
|
||||
assert stale_keys.isdisjoint(endpoint_keys)
|
||||
|
||||
|
||||
def test_visit_list_permission_matches_wired_visit_endpoints():
|
||||
"""访视当前没有详情查询接口，列表接口应只使用 visits:list。"""
|
||||
visits_path = Path(__file__).resolve().parents[1] / "app" / "api" / "v1" / "visits.py"
|
||||
source = visits_path.read_text()
|
||||
|
||||
assert "visits:list" in API_ENDPOINT_PERMISSIONS
|
||||
assert "visits:read" not in API_ENDPOINT_PERMISSIONS
|
||||
assert 'require_api_permission("visits:list")' in source
|
||||
assert 'require_api_permission("visits:read")' not in source
|
||||
|
||||
|
||||
def test_ae_subject_and_summary_share_read_permission():
|
||||
"""风险问题AE汇总与参与者AE列表应使用同一个读取权限。"""
|
||||
aes_path = Path(__file__).resolve().parents[1] / "app" / "api" / "v1" / "aes.py"
|
||||
source = aes_path.read_text()
|
||||
summary_chunk = source[source.index('"/summary"') : source.index('@router.get(\n "/"')]
|
||||
list_chunk = source[source.index('@router.get(\n "/",') : source.index('@router.get(\n "/{ae_id}"')]
|
||||
detail_chunk = source[source.index('response_model=AERead,\n dependencies=[Depends(require_api_permission("subject_aes:read"))]') : source.index('@router.patch')]
|
||||
|
||||
assert 'require_api_permission("subject_aes:list")' in summary_chunk
|
||||
assert 'require_api_permission("subject_aes:list")' in list_chunk
|
||||
assert 'require_api_permission("subject_aes:read")' in detail_chunk
|
||||
|
||||
|
||||
def test_ae_mutation_permissions_are_under_subject_management():
|
||||
"""AE维护发生在参与者详情内，风险问题模块只保留汇总读取权限。"""
|
||||
expected_descriptions = {
|
||||
"subject_aes:create": "创建参与者AE",
|
||||
"subject_aes:list": "查询参与者AE列表",
|
||||
"subject_aes:read": "查询参与者AE详情",
|
||||
"subject_aes:update": "更新参与者AE",
|
||||
"subject_aes:delete": "删除参与者AE",
|
||||
}
|
||||
for endpoint_key, description in expected_descriptions.items():
|
||||
assert API_ENDPOINT_PERMISSIONS[endpoint_key]["module"] == "subjects"
|
||||
assert API_ENDPOINT_PERMISSIONS[endpoint_key]["description"] == description
|
||||
|
||||
for stale_key in ("risk_issues:create", "risk_issues:list", "risk_issues:read", "risk_issues:update", "risk_issues:delete"):
|
||||
assert stale_key not in API_ENDPOINT_PERMISSIONS
|
||||
for stale_key in ("risk_issue_aes:list", "risk_issue_pds:list"):
|
||||
assert stale_key not in API_ENDPOINT_PERMISSIONS
|
||||
|
||||
|
||||
def test_risk_issue_pd_summary_uses_subject_pd_permission():
|
||||
"""风险问题PD汇总与参与者PD列表应使用同一个读取权限。"""
|
||||
pds_path = Path(__file__).resolve().parents[1] / "app" / "api" / "v1" / "study_subject_pds.py"
|
||||
source = pds_path.read_text()
|
||||
|
||||
assert 'require_api_permission("subject_pds:list")' in source
|
||||
assert 'require_api_permission("risk_issues:list")' not in source
|
||||
|
||||
|
||||
def test_risk_issue_monitoring_visit_uses_monitoring_issue_permissions():
|
||||
"""监查访视问题是风险问题下唯一独立权限子模块。"""
|
||||
monitoring_path = Path(__file__).resolve().parents[1] / "app" / "api" / "v1" / "monitoring_visit_issues.py"
|
||||
source = monitoring_path.read_text()
|
||||
legacy_issue_prefix = "monitoring" + "_issues"
|
||||
tree = ast.parse(source)
|
||||
used_permission_keys = {
|
||||
node.args[0].value
|
||||
for node in ast.walk(tree)
|
||||
if isinstance(node, ast.Call)
|
||||
and isinstance(node.func, ast.Name)
|
||||
and node.func.id == "require_api_permission"
|
||||
and node.args
|
||||
and isinstance(node.args[0], ast.Constant)
|
||||
and isinstance(node.args[0].value, str)
|
||||
}
|
||||
list_chunk = source[source.index('@router.get(\n "/issues"') : source.index("@router.post")]
|
||||
create_chunk = source[source.index("@router.post") : source.index('@router.get(\n "/issues/export"')]
|
||||
export_chunk = source[source.index('@router.get(\n "/issues/export"') : source.index('@router.get(\n "/issues/{issue_id}"')]
|
||||
detail_chunk = source[source.index('@router.get(\n "/issues/{issue_id}"') : source.index("@router.patch")]
|
||||
update_chunk = source[source.index("@router.patch") : source.index("@router.delete")]
|
||||
delete_chunk = source[source.index("@router.delete") : source.index('@router.post(\n "/issues/import"')]
|
||||
import_chunk = source[source.index('@router.post(\n "/issues/import"') :]
|
||||
|
||||
expected_descriptions = {
|
||||
"monitoring_issues:list": "查询监查访视问题列表",
|
||||
"monitoring_issues:read": "查询监查访视问题详情",
|
||||
"monitoring_issues:create": "创建监查访视问题",
|
||||
"monitoring_issues:update": "更新监查访视问题",
|
||||
"monitoring_issues:delete": "删除监查访视问题",
|
||||
}
|
||||
for endpoint_key, description in expected_descriptions.items():
|
||||
assert API_ENDPOINT_PERMISSIONS[endpoint_key]["module"] == "risk_issues"
|
||||
assert API_ENDPOINT_PERMISSIONS[endpoint_key]["description"] == description
|
||||
|
||||
assert 'require_api_permission("monitoring_issues:list")' in list_chunk
|
||||
assert 'require_api_permission("monitoring_issues:list")' in export_chunk
|
||||
assert 'require_api_permission("monitoring_issues:read")' in detail_chunk
|
||||
assert 'require_api_permission("monitoring_issues:create")' in create_chunk
|
||||
assert 'require_api_permission("monitoring_issues:create")' in import_chunk
|
||||
assert 'require_api_permission("monitoring_issues:update")' in update_chunk
|
||||
assert 'require_api_permission("monitoring_issues:delete")' in delete_chunk
|
||||
assert used_permission_keys == set(expected_descriptions)
|
||||
|
||||
for stale_key in (
|
||||
f"{legacy_issue_prefix}:close",
|
||||
f"{legacy_issue_prefix}:history",
|
||||
):
|
||||
assert stale_key not in API_ENDPOINT_PERMISSIONS
|
||||
assert not any(config["module"] == "monitoring" + "_audit" for config in API_ENDPOINT_PERMISSIONS.values())
|
||||
|
||||
|
||||
def test_document_service_uses_specific_document_permission_keys():
|
||||
"""文档模块应使用 create/update/delete 细粒度权限，而不是全部退化为 update。"""
|
||||
service_path = Path(__file__).resolve().parents[1] / "app" / "services" / "document_service.py"
|
||||
@@ -72,8 +392,8 @@ def test_document_service_uses_specific_document_permission_keys():
|
||||
assert 'else "documents:update"' not in source
|
||||
|
||||
|
||||
def test_user_role_no_longer_contains_qa():
|
||||
assert "QA" not in {role.value for role in UserRole}
|
||||
def test_user_role_contains_current_project_roles():
|
||||
assert {"QA", "CTA"} <= {role.value for role in UserRole}
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@@ -82,8 +402,9 @@ async def test_default_matrix_covers_every_role_and_permission(db_session: Async
|
||||
study_id = uuid.uuid4()
|
||||
matrix = await get_api_endpoint_permissions(db_session, study_id)
|
||||
|
||||
assert "QA" not in PROJECT_PERMISSION_ROLES
|
||||
assert all("QA" not in config["default_roles"] for config in API_ENDPOINT_PERMISSIONS.values())
|
||||
assert {"QA", "CTA"} <= set(PROJECT_PERMISSION_ROLES)
|
||||
assert any("QA" in config["default_roles"] for config in API_ENDPOINT_PERMISSIONS.values())
|
||||
assert any("CTA" in config["default_roles"] for config in API_ENDPOINT_PERMISSIONS.values())
|
||||
assert set(matrix) == set(PROJECT_PERMISSION_ROLES)
|
||||
for role in PROJECT_PERMISSION_ROLES:
|
||||
assert set(matrix[role]) == set(API_ENDPOINT_PERMISSIONS)
|
||||
@@ -184,6 +505,69 @@ async def test_custom_active_role_permissions_take_effect(db_session: AsyncSessi
|
||||
) is False
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_permission_matrix_filters_stale_unknown_endpoint_overrides(db_session: AsyncSession):
|
||||
"""数据库历史残留的未知权限项不应重新出现在权限矩阵中。"""
|
||||
study_id = uuid.uuid4()
|
||||
db_session.add_all(
|
||||
(
|
||||
ApiEndpointPermission(
|
||||
study_id=study_id,
|
||||
role="CRA",
|
||||
endpoint_key="subject_history:export",
|
||||
allowed=True,
|
||||
),
|
||||
ApiEndpointPermission(
|
||||
study_id=study_id,
|
||||
role="CRA",
|
||||
endpoint_key="subject_pds:read",
|
||||
allowed=True,
|
||||
),
|
||||
ApiEndpointPermission(
|
||||
study_id=study_id,
|
||||
role="CRA",
|
||||
endpoint_key="visits:read",
|
||||
allowed=True,
|
||||
),
|
||||
ApiEndpointPermission(
|
||||
study_id=study_id,
|
||||
role="CRA",
|
||||
endpoint_key="risk_issues:create",
|
||||
allowed=True,
|
||||
),
|
||||
ApiEndpointPermission(
|
||||
study_id=study_id,
|
||||
role="CRA",
|
||||
endpoint_key="risk_issues:list",
|
||||
allowed=True,
|
||||
),
|
||||
ApiEndpointPermission(
|
||||
study_id=study_id,
|
||||
role="CRA",
|
||||
endpoint_key="risk_issue_aes:list",
|
||||
allowed=True,
|
||||
),
|
||||
ApiEndpointPermission(
|
||||
study_id=study_id,
|
||||
role="CRA",
|
||||
endpoint_key=("monitoring" + "_issues") + ":history",
|
||||
allowed=True,
|
||||
),
|
||||
)
|
||||
)
|
||||
await db_session.commit()
|
||||
|
||||
matrix = await get_api_endpoint_permissions(db_session, study_id)
|
||||
|
||||
assert "subject_history:export" not in matrix["CRA"]
|
||||
assert "subject_pds:read" not in matrix["CRA"]
|
||||
assert "visits:read" not in matrix["CRA"]
|
||||
assert "risk_issues:create" not in matrix["CRA"]
|
||||
assert "risk_issues:list" not in matrix["CRA"]
|
||||
assert "risk_issue_aes:list" not in matrix["CRA"]
|
||||
assert ("monitoring" + "_issues") + ":history" not in matrix["CRA"]
|
||||
|
||||
|
||||
def test_admin_cannot_be_used_as_project_custom_role():
|
||||
"""避免项目角色 ADMIN 触发后端系统管理员绕过逻辑。"""
|
||||
with pytest.raises(ValueError):
|
||||
@@ -218,7 +602,7 @@ async def test_admin_always_allowed(db_session: AsyncSession):
|
||||
study_id = uuid.uuid4()
|
||||
|
||||
result = await role_has_api_permission(
|
||||
db_session, study_id, "ADMIN", "POST:/subjects"
|
||||
db_session, study_id, "ADMIN", "subjects:create"
|
||||
)
|
||||
assert result is True
|
||||
|
||||
@@ -370,6 +754,59 @@ async def test_api_permission_unknown_endpoint(db_session: AsyncSession):
|
||||
assert result is False
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_api_permission_unknown_endpoint_ignores_stale_override(db_session: AsyncSession):
|
||||
"""即使数据库残留未知权限项，鉴权入口也不应放行。"""
|
||||
study_id = uuid.uuid4()
|
||||
db_session.add_all(
|
||||
(
|
||||
ApiEndpointPermission(
|
||||
study_id=study_id,
|
||||
role="CRA",
|
||||
endpoint_key="subject_history:export",
|
||||
allowed=True,
|
||||
),
|
||||
ApiEndpointPermission(
|
||||
study_id=study_id,
|
||||
role="CRA",
|
||||
endpoint_key="subject_pds:read",
|
||||
allowed=True,
|
||||
),
|
||||
ApiEndpointPermission(
|
||||
study_id=study_id,
|
||||
role="CRA",
|
||||
endpoint_key="risk_issues:list",
|
||||
allowed=True,
|
||||
),
|
||||
ApiEndpointPermission(
|
||||
study_id=study_id,
|
||||
role="CRA",
|
||||
endpoint_key="risk_issue_aes:list",
|
||||
allowed=True,
|
||||
),
|
||||
)
|
||||
)
|
||||
await db_session.commit()
|
||||
|
||||
history_result = await role_has_api_permission(
|
||||
db_session, study_id, "CRA", "subject_history:export", check_prerequisites=False
|
||||
)
|
||||
pds_result = await role_has_api_permission(
|
||||
db_session, study_id, "CRA", "subject_pds:read", check_prerequisites=False
|
||||
)
|
||||
risk_issue_result = await role_has_api_permission(
|
||||
db_session, study_id, "CRA", "risk_issues:list", check_prerequisites=False
|
||||
)
|
||||
risk_issue_ae_result = await role_has_api_permission(
|
||||
db_session, study_id, "CRA", "risk_issue_aes:list", check_prerequisites=False
|
||||
)
|
||||
|
||||
assert history_result is False
|
||||
assert pds_result is False
|
||||
assert risk_issue_result is False
|
||||
assert risk_issue_ae_result is False
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_api_permission_check_uses_project_permission_cache(db_session: AsyncSession):
|
||||
"""测试接口权限检查会复用项目权限缓存并记录命中指标"""
|
||||
|
||||
Reference in New Issue
Block a user