完全移除模块级权限系统，迁移至接口级权限
后端： - 删除 StudyRolePermission 模型和 project_permissions API 文件 - 重写 project_permissions.py core，移除所有模块级权限函数 - 重写 permission_cache.py，移除模块级权限缓存逻辑 - 新增权限模板功能（PermissionTemplate 模型、API、服务层） - 新增 permission_templates 数据库迁移 - 迁移 8 个模块（attachments、audit_logs、dashboard、faqs、 faq_categories、fees_attachments、knowledge_notes、 material_equipments、overview、subject_histories、subject_pds） 至接口级权限检查 - 删除所有模块级权限相关测试文件，新增权限模板测试 前端： - 删除 ProjectPermissions.vue 和 ProjectPermissionsModule.vue - 重写 projectRoutePermissions.ts，改为基于接口级权限格式 - 更新 store/study.ts、router/index.ts、AuditLogs.vue、Projects.vue 中的权限 API 调用，从 fetchProjectRolePermissions 改为 fetchApiEndpointPermissions - 清理 types/api.ts 中的旧模块级权限类型定义 - 新增 PermissionTemplateSelector.vue 组件 - 更新权限管理页面，移除模块级权限 tab Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Cheng Zhou
@@ -73,7 +73,6 @@ async def _create_test_engine():
|
||||
# Import all models to register them with Base metadata
|
||||
from app.db.base_class import Base
|
||||
import app.models.study
|
||||
import app.models.study_role_permission
|
||||
import app.models.api_endpoint_permission
|
||||
import app.models.api_endpoint_registry
|
||||
import app.models.user
|
||||
@@ -83,6 +82,7 @@ async def _create_test_engine():
|
||||
import app.models.ae
|
||||
import app.models.monitoring_visit_issue
|
||||
import app.models.site
|
||||
import app.models.permission_template
|
||||
|
||||
# Replace PostgreSQL UUID type with custom GUID type for SQLite
|
||||
for table in Base.metadata.tables.values():
|
||||
|
||||
@@ -6,7 +6,6 @@ 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
|
||||
@@ -53,50 +52,6 @@ async def test_api_permission_check_denied(db_session: AsyncSession):
|
||||
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角色总是被允许"""
|
||||
@@ -108,37 +63,6 @@ async def test_admin_always_allowed(db_session: AsyncSession):
|
||||
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):
|
||||
"""测试读取端点权限"""
|
||||
|
||||
@@ -1,177 +0,0 @@
|
||||
"""单元测试：API权限配置"""
|
||||
|
||||
import pytest
|
||||
|
||||
from app.core.api_permissions import API_ENDPOINT_PERMISSIONS, MODULE_TO_ENDPOINTS
|
||||
|
||||
|
||||
def test_api_endpoint_permissions_structure():
|
||||
"""测试API端点权限配置结构"""
|
||||
for endpoint_key, config in API_ENDPOINT_PERMISSIONS.items():
|
||||
assert "module" in config, f"Missing 'module' in {endpoint_key}"
|
||||
assert "action" in config, f"Missing 'action' in {endpoint_key}"
|
||||
assert "description" in config, f"Missing 'description' in {endpoint_key}"
|
||||
assert "default_roles" in config, f"Missing 'default_roles' in {endpoint_key}"
|
||||
assert config["action"] in ["read", "write"], f"Invalid action in {endpoint_key}"
|
||||
assert isinstance(config["default_roles"], list), f"default_roles must be list in {endpoint_key}"
|
||||
assert len(config["default_roles"]) > 0, f"default_roles must not be empty in {endpoint_key}"
|
||||
|
||||
|
||||
def test_module_to_endpoints_mapping():
|
||||
"""测试模块到端点的映射"""
|
||||
for module, actions in MODULE_TO_ENDPOINTS.items():
|
||||
assert "read" in actions, f"Missing 'read' in {module}"
|
||||
assert "write" in actions, f"Missing 'write' in {module}"
|
||||
assert isinstance(actions["read"], list), f"read must be list in {module}"
|
||||
assert isinstance(actions["write"], list), f"write must be list in {module}"
|
||||
|
||||
# 验证所有端点都在API_ENDPOINT_PERMISSIONS中定义
|
||||
for endpoint_key in actions["read"] + actions["write"]:
|
||||
assert endpoint_key in API_ENDPOINT_PERMISSIONS, f"Endpoint {endpoint_key} not in API_ENDPOINT_PERMISSIONS"
|
||||
|
||||
|
||||
def test_subjects_endpoints_configured():
|
||||
"""测试subjects模块的端点配置"""
|
||||
expected_endpoints = [
|
||||
"POST:/subjects",
|
||||
"GET:/subjects",
|
||||
"GET:/subjects/{id}",
|
||||
"PATCH:/subjects/{id}",
|
||||
"DELETE:/subjects/{id}",
|
||||
"POST:/subjects/{subject_id}/visits",
|
||||
"GET:/subjects/{subject_id}/visits",
|
||||
"GET:/subjects/{subject_id}/visits/{visit_id}",
|
||||
"PATCH:/subjects/{subject_id}/visits/{visit_id}",
|
||||
]
|
||||
for endpoint_key in expected_endpoints:
|
||||
assert endpoint_key in API_ENDPOINT_PERMISSIONS, f"Missing endpoint {endpoint_key}"
|
||||
assert API_ENDPOINT_PERMISSIONS[endpoint_key]["module"] == "subjects"
|
||||
|
||||
|
||||
def test_risk_issues_endpoints_configured():
|
||||
"""测试risk_issues模块的端点配置"""
|
||||
expected_endpoints = [
|
||||
"POST:/risk-issues",
|
||||
"GET:/risk-issues",
|
||||
"GET:/risk-issues/{id}",
|
||||
"DELETE:/risk-issues/{id}",
|
||||
]
|
||||
for endpoint_key in expected_endpoints:
|
||||
assert endpoint_key in API_ENDPOINT_PERMISSIONS, f"Missing endpoint {endpoint_key}"
|
||||
assert API_ENDPOINT_PERMISSIONS[endpoint_key]["module"] == "risk_issues"
|
||||
|
||||
|
||||
def test_fees_endpoints_configured():
|
||||
"""测试fees模块的端点配置"""
|
||||
expected_endpoints = [
|
||||
"POST:/fees/contracts",
|
||||
"GET:/fees/contracts",
|
||||
"GET:/fees/contracts/{id}",
|
||||
"PATCH:/fees/contracts/{id}",
|
||||
"DELETE:/fees/contracts/{id}",
|
||||
"POST:/fees/contracts/{id}/payments",
|
||||
"PATCH:/fees/payments/{id}",
|
||||
"DELETE:/fees/payments/{id}",
|
||||
"POST:/finance/contracts",
|
||||
"GET:/finance/contracts",
|
||||
"GET:/finance/contracts/{id}",
|
||||
"PATCH:/finance/contracts/{id}",
|
||||
"DELETE:/finance/contracts/{id}",
|
||||
]
|
||||
for endpoint_key in expected_endpoints:
|
||||
assert endpoint_key in API_ENDPOINT_PERMISSIONS, f"Missing endpoint {endpoint_key}"
|
||||
assert API_ENDPOINT_PERMISSIONS[endpoint_key]["module"] == "fees"
|
||||
|
||||
|
||||
def test_project_members_endpoints_configured():
|
||||
"""测试project_members模块的端点配置"""
|
||||
expected_endpoints = [
|
||||
"POST:/studies/{study_id}/members",
|
||||
"GET:/studies/{study_id}/members",
|
||||
"GET:/studies/{study_id}/members/candidates",
|
||||
"PATCH:/studies/{study_id}/members/{member_id}",
|
||||
"DELETE:/studies/{study_id}/members/{member_id}",
|
||||
]
|
||||
for endpoint_key in expected_endpoints:
|
||||
assert endpoint_key in API_ENDPOINT_PERMISSIONS, f"Missing endpoint {endpoint_key}"
|
||||
assert API_ENDPOINT_PERMISSIONS[endpoint_key]["module"] == "project_members"
|
||||
|
||||
|
||||
def test_sites_endpoints_configured():
|
||||
"""测试sites模块的端点配置"""
|
||||
expected_endpoints = [
|
||||
"POST:/studies/{study_id}/sites",
|
||||
"GET:/studies/{study_id}/sites",
|
||||
"GET:/studies/{study_id}/sites/{site_id}",
|
||||
"PATCH:/studies/{study_id}/sites/{site_id}",
|
||||
"DELETE:/studies/{study_id}/sites/{site_id}",
|
||||
]
|
||||
for endpoint_key in expected_endpoints:
|
||||
assert endpoint_key in API_ENDPOINT_PERMISSIONS, f"Missing endpoint {endpoint_key}"
|
||||
assert API_ENDPOINT_PERMISSIONS[endpoint_key]["module"] == "sites"
|
||||
|
||||
|
||||
def test_endpoint_key_format():
|
||||
"""测试端点key格式"""
|
||||
for endpoint_key in API_ENDPOINT_PERMISSIONS.keys():
|
||||
# 格式应该是 "METHOD:/path"
|
||||
assert ":" in endpoint_key, f"Invalid endpoint_key format: {endpoint_key}"
|
||||
method, path = endpoint_key.split(":", 1)
|
||||
assert method in ["GET", "POST", "PATCH", "DELETE", "PUT"], f"Invalid method in {endpoint_key}"
|
||||
assert path.startswith("/"), f"Invalid path in {endpoint_key}"
|
||||
|
||||
|
||||
def test_default_roles_valid():
|
||||
"""测试默认角色有效"""
|
||||
valid_roles = {"PM", "CRA", "PV", "MEDICAL_REVIEW", "IMP", "QA", "ADMIN"}
|
||||
|
||||
for endpoint_key, config in API_ENDPOINT_PERMISSIONS.items():
|
||||
for role in config["default_roles"]:
|
||||
assert role in valid_roles, f"Invalid role {role} in {endpoint_key}"
|
||||
|
||||
|
||||
def test_module_to_endpoints_completeness():
|
||||
"""测试MODULE_TO_ENDPOINTS包含所有模块"""
|
||||
modules_in_config = set()
|
||||
for config in API_ENDPOINT_PERMISSIONS.values():
|
||||
modules_in_config.add(config["module"])
|
||||
|
||||
for module in modules_in_config:
|
||||
assert module in MODULE_TO_ENDPOINTS, f"Module {module} not in MODULE_TO_ENDPOINTS"
|
||||
|
||||
|
||||
def test_read_write_endpoints_consistency():
|
||||
"""测试read/write端点的一致性"""
|
||||
for module, actions in MODULE_TO_ENDPOINTS.items():
|
||||
read_endpoints = set(actions["read"])
|
||||
write_endpoints = set(actions["write"])
|
||||
|
||||
# read和write不应该有重叠
|
||||
overlap = read_endpoints & write_endpoints
|
||||
assert len(overlap) == 0, f"Overlap between read and write in {module}: {overlap}"
|
||||
|
||||
# 所有端点都应该在API_ENDPOINT_PERMISSIONS中
|
||||
all_endpoints = read_endpoints | write_endpoints
|
||||
for endpoint_key in all_endpoints:
|
||||
assert endpoint_key in API_ENDPOINT_PERMISSIONS, f"Endpoint {endpoint_key} not in API_ENDPOINT_PERMISSIONS"
|
||||
config = API_ENDPOINT_PERMISSIONS[endpoint_key]
|
||||
assert config["module"] == module, f"Module mismatch for {endpoint_key}"
|
||||
|
||||
# 验证action与read/write分类一致
|
||||
if endpoint_key in read_endpoints:
|
||||
assert config["action"] == "read", f"Action mismatch for {endpoint_key}"
|
||||
else:
|
||||
assert config["action"] == "write", f"Action mismatch for {endpoint_key}"
|
||||
|
||||
|
||||
def test_no_duplicate_endpoints():
|
||||
"""测试没有重复的端点"""
|
||||
endpoints = list(API_ENDPOINT_PERMISSIONS.keys())
|
||||
assert len(endpoints) == len(set(endpoints)), "Duplicate endpoints found"
|
||||
|
||||
|
||||
def test_endpoint_descriptions_not_empty():
|
||||
"""测试所有端点都有描述"""
|
||||
for endpoint_key, config in API_ENDPOINT_PERMISSIONS.items():
|
||||
assert config["description"], f"Empty description for {endpoint_key}"
|
||||
assert len(config["description"]) > 0, f"Empty description for {endpoint_key}"
|
||||
@@ -7,7 +7,6 @@ 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
|
||||
from app.models.study_role_permission import StudyRolePermission
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
|
||||
@@ -7,7 +7,6 @@ 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
|
||||
|
||||
@@ -9,7 +9,6 @@ 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
|
||||
@@ -271,98 +270,3 @@ async def test_sites_permission_denied_for_cra_write(db_session: AsyncSession):
|
||||
allowed = await role_has_api_permission(db_session, study_id, "CRA", "POST:/studies/{study_id}/sites")
|
||||
assert allowed is False
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_backward_compatibility_members_module_level(db_session: AsyncSession):
|
||||
"""验证members模块的模块级权限回退仍然有效"""
|
||||
study_id = uuid.uuid4()
|
||||
|
||||
# 设置模块级权限（不设置接口级权限）
|
||||
db_session.add(StudyRolePermission(
|
||||
study_id=study_id,
|
||||
role="PM",
|
||||
module="project_members",
|
||||
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}/members")
|
||||
assert allowed is True
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_backward_compatibility_sites_module_level(db_session: AsyncSession):
|
||||
"""验证sites模块的模块级权限回退仍然有效"""
|
||||
study_id = uuid.uuid4()
|
||||
|
||||
# 设置模块级权限（不设置接口级权限）
|
||||
db_session.add(StudyRolePermission(
|
||||
study_id=study_id,
|
||||
role="PM",
|
||||
module="sites",
|
||||
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}/sites")
|
||||
assert allowed is True
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_api_permission_priority_over_module_members(db_session: AsyncSession):
|
||||
"""验证members模块的接口级权限优先于模块级权限"""
|
||||
study_id = uuid.uuid4()
|
||||
|
||||
# 设置模块级权限为允许
|
||||
db_session.add(StudyRolePermission(
|
||||
study_id=study_id,
|
||||
role="PM",
|
||||
module="project_members",
|
||||
can_read=True,
|
||||
can_write=True,
|
||||
))
|
||||
|
||||
# 设置接口级权限为拒绝（优先级更高）
|
||||
db_session.add(ApiEndpointPermission(
|
||||
study_id=study_id,
|
||||
role="PM",
|
||||
endpoint_key="POST:/studies/{study_id}/members",
|
||||
allowed=False,
|
||||
))
|
||||
await db_session.commit()
|
||||
|
||||
# 验证接口级权限优先
|
||||
allowed = await role_has_api_permission(db_session, study_id, "PM", "POST:/studies/{study_id}/members")
|
||||
assert allowed is False
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_api_permission_priority_over_module_sites(db_session: AsyncSession):
|
||||
"""验证sites模块的接口级权限优先于模块级权限"""
|
||||
study_id = uuid.uuid4()
|
||||
|
||||
# 设置模块级权限为允许
|
||||
db_session.add(StudyRolePermission(
|
||||
study_id=study_id,
|
||||
role="PM",
|
||||
module="sites",
|
||||
can_read=True,
|
||||
can_write=True,
|
||||
))
|
||||
|
||||
# 设置接口级权限为拒绝（优先级更高）
|
||||
db_session.add(ApiEndpointPermission(
|
||||
study_id=study_id,
|
||||
role="PM",
|
||||
endpoint_key="POST:/studies/{study_id}/sites",
|
||||
allowed=False,
|
||||
))
|
||||
await db_session.commit()
|
||||
|
||||
# 验证接口级权限优先
|
||||
allowed = await role_has_api_permission(db_session, study_id, "PM", "POST:/studies/{study_id}/sites")
|
||||
assert allowed is False
|
||||
|
||||
@@ -9,7 +9,6 @@ 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
|
||||
|
||||
|
||||
# ============================================================================
|
||||
@@ -538,135 +537,3 @@ async def test_project_permissions_denied_for_cra(db_session: AsyncSession):
|
||||
|
||||
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}"
|
||||
|
||||
@@ -1,246 +0,0 @@
|
||||
"""缓存测试：权限缓存机制验证
|
||||
|
||||
测试权限缓存的功能，包括：
|
||||
- 缓存命中和未命中
|
||||
- 缓存过期
|
||||
- 缓存失效
|
||||
- 并发缓存访问
|
||||
- 缓存统计
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import time
|
||||
import uuid
|
||||
|
||||
import pytest
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from app.core.permission_cache import PermissionCache, get_permission_cache, set_permission_cache
|
||||
from app.core.project_permissions import get_project_role_permissions, get_member_role
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_cache_hit():
|
||||
"""测试缓存命中"""
|
||||
cache = PermissionCache()
|
||||
|
||||
# 第一次调用：缓存未命中
|
||||
cache._project_permissions_cache["test_key"] = ({"test": "data"}, time.time())
|
||||
|
||||
# 第二次调用：缓存命中
|
||||
cached_data, timestamp = cache._project_permissions_cache.get("test_key")
|
||||
assert cached_data == {"test": "data"}
|
||||
assert not cache._is_expired(timestamp, 300)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_cache_miss():
|
||||
"""测试缓存未命中"""
|
||||
cache = PermissionCache()
|
||||
|
||||
# 缓存中不存在该键
|
||||
assert "nonexistent_key" not in cache._project_permissions_cache
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_cache_expiration():
|
||||
"""测试缓存过期"""
|
||||
cache = PermissionCache(default_ttl=1)
|
||||
|
||||
# 添加一个即将过期的缓存项
|
||||
cache._project_permissions_cache["test_key"] = ({"test": "data"}, time.time() - 2)
|
||||
|
||||
# 验证缓存已过期
|
||||
_, timestamp = cache._project_permissions_cache["test_key"]
|
||||
assert cache._is_expired(timestamp, 1)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_cache_invalidation():
|
||||
"""测试缓存失效"""
|
||||
cache = PermissionCache()
|
||||
study_id = uuid.uuid4()
|
||||
|
||||
# 添加缓存
|
||||
cache._project_permissions_cache[cache._make_project_cache_key(study_id)] = (
|
||||
{"test": "data"},
|
||||
time.time(),
|
||||
)
|
||||
assert len(cache._project_permissions_cache) == 1
|
||||
|
||||
# 失效缓存
|
||||
cache.invalidate_project_permissions(study_id)
|
||||
assert len(cache._project_permissions_cache) == 0
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_member_role_cache_invalidation():
|
||||
"""测试成员角色缓存失效"""
|
||||
cache = PermissionCache()
|
||||
study_id = uuid.uuid4()
|
||||
user_id = uuid.uuid4()
|
||||
|
||||
# 添加缓存
|
||||
cache._member_role_cache[cache._make_member_cache_key(study_id, user_id)] = ("PM", time.time())
|
||||
assert len(cache._member_role_cache) == 1
|
||||
|
||||
# 失效缓存
|
||||
cache.invalidate_member_role(study_id, user_id)
|
||||
assert len(cache._member_role_cache) == 0
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_invalidate_all_member_roles():
|
||||
"""测试失效项目中所有成员的角色缓存"""
|
||||
cache = PermissionCache()
|
||||
study_id = uuid.uuid4()
|
||||
|
||||
# 添加多个成员的缓存
|
||||
for i in range(5):
|
||||
user_id = uuid.uuid4()
|
||||
cache._member_role_cache[cache._make_member_cache_key(study_id, user_id)] = ("PM", time.time())
|
||||
|
||||
assert len(cache._member_role_cache) == 5
|
||||
|
||||
# 失效项目中所有成员的缓存
|
||||
cache.invalidate_all_member_roles(study_id)
|
||||
assert len(cache._member_role_cache) == 0
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_concurrent_cache_access():
|
||||
"""测试并发缓存访问"""
|
||||
cache = PermissionCache()
|
||||
study_id = uuid.uuid4()
|
||||
|
||||
async def add_to_cache(i):
|
||||
user_id = uuid.uuid4()
|
||||
cache._member_role_cache[cache._make_member_cache_key(study_id, user_id)] = ("PM", time.time())
|
||||
|
||||
# 并发添加缓存
|
||||
tasks = [add_to_cache(i) for i in range(10)]
|
||||
await asyncio.gather(*tasks)
|
||||
|
||||
# 验证所有缓存都被添加
|
||||
assert len(cache._member_role_cache) == 10
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_cache_key_generation():
|
||||
"""测试缓存键生成"""
|
||||
cache = PermissionCache()
|
||||
study_id = uuid.uuid4()
|
||||
user_id = uuid.uuid4()
|
||||
|
||||
# 验证缓存键格式
|
||||
project_key = cache._make_project_cache_key(study_id)
|
||||
assert project_key.startswith("project_permissions:")
|
||||
assert str(study_id) in project_key
|
||||
|
||||
member_key = cache._make_member_cache_key(study_id, user_id)
|
||||
assert member_key.startswith("member_role:")
|
||||
assert str(study_id) in member_key
|
||||
assert str(user_id) in member_key
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_cache_stats():
|
||||
"""测试缓存统计"""
|
||||
cache = PermissionCache()
|
||||
study_id = uuid.uuid4()
|
||||
|
||||
# 添加缓存
|
||||
cache._project_permissions_cache[cache._make_project_cache_key(study_id)] = (
|
||||
{"test": "data"},
|
||||
time.time(),
|
||||
)
|
||||
cache._member_role_cache[cache._make_member_cache_key(study_id, uuid.uuid4())] = ("PM", time.time())
|
||||
|
||||
# 获取统计信息
|
||||
stats = cache.get_cache_stats()
|
||||
assert stats["project_permissions_count"] == 1
|
||||
assert stats["member_role_count"] == 1
|
||||
assert stats["total_count"] == 2
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_clear_all_cache():
|
||||
"""测试清除所有缓存"""
|
||||
cache = PermissionCache()
|
||||
study_id = uuid.uuid4()
|
||||
|
||||
# 添加缓存
|
||||
cache._project_permissions_cache[cache._make_project_cache_key(study_id)] = (
|
||||
{"test": "data"},
|
||||
time.time(),
|
||||
)
|
||||
cache._member_role_cache[cache._make_member_cache_key(study_id, uuid.uuid4())] = ("PM", time.time())
|
||||
|
||||
assert len(cache._project_permissions_cache) == 1
|
||||
assert len(cache._member_role_cache) == 1
|
||||
|
||||
# 清除所有缓存
|
||||
cache.clear_all()
|
||||
assert len(cache._project_permissions_cache) == 0
|
||||
assert len(cache._member_role_cache) == 0
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_cache_with_different_ttl(db_session: AsyncSession, study_id: uuid.UUID):
|
||||
"""测试不同 TTL 的缓存"""
|
||||
cache = PermissionCache()
|
||||
|
||||
# 使用不同的 TTL 获取权限
|
||||
permissions1 = await cache.get_project_role_permissions(db_session, study_id, ttl=1)
|
||||
permissions2 = await cache.get_project_role_permissions(db_session, study_id, ttl=300)
|
||||
|
||||
# 两次调用都应该返回相同的数据
|
||||
assert permissions1 == permissions2
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_global_cache_instance():
|
||||
"""测试全局缓存实例"""
|
||||
cache1 = get_permission_cache()
|
||||
cache2 = get_permission_cache()
|
||||
|
||||
# 应该是同一个实例
|
||||
assert cache1 is cache2
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_set_global_cache_instance():
|
||||
"""测试设置全局缓存实例"""
|
||||
new_cache = PermissionCache()
|
||||
set_permission_cache(new_cache)
|
||||
|
||||
cache = get_permission_cache()
|
||||
assert cache is new_cache
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_cache_performance_improvement(db_session: AsyncSession, study_id: uuid.UUID):
|
||||
"""测试缓存的性能改进
|
||||
|
||||
验证缓存确实提高了性能。
|
||||
"""
|
||||
cache = PermissionCache()
|
||||
set_permission_cache(cache)
|
||||
cache.clear_all()
|
||||
|
||||
# 第一次调用：从数据库查询
|
||||
start_time = time.time()
|
||||
permissions1 = await cache.get_project_role_permissions(db_session, study_id)
|
||||
first_call_time = time.time() - start_time
|
||||
|
||||
# 第二次调用：从缓存获取
|
||||
start_time = time.time()
|
||||
permissions2 = await cache.get_project_role_permissions(db_session, study_id)
|
||||
second_call_time = time.time() - start_time
|
||||
|
||||
# 验证结果相同
|
||||
assert permissions1 == permissions2
|
||||
|
||||
# 缓存调用应该快得多
|
||||
print(f"\n缓存性能改进：第一次 {first_call_time*1000:.2f}ms，第二次 {second_call_time*1000:.2f}ms")
|
||||
assert second_call_time < first_call_time / 2, "缓存性能改进不足"
|
||||
@@ -1,228 +0,0 @@
|
||||
"""性能测试：权限系统缓存效率
|
||||
|
||||
测试权限检查的性能，对比有缓存和无缓存的性能差异。
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import time
|
||||
import uuid
|
||||
from typing import Callable
|
||||
|
||||
import pytest
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from app.core.permission_cache import PermissionCache, get_permission_cache, set_permission_cache
|
||||
from app.core.project_permissions import (
|
||||
get_project_role_permissions,
|
||||
role_has_project_permission,
|
||||
role_has_api_permission,
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_permission_check_performance_baseline(db_session: AsyncSession, study_id: uuid.UUID):
|
||||
"""基准测试：权限检查性能（无缓存）"""
|
||||
# 清除缓存
|
||||
cache = get_permission_cache()
|
||||
cache.clear_all()
|
||||
|
||||
# 执行权限检查 100 次
|
||||
start_time = time.time()
|
||||
for _ in range(100):
|
||||
await role_has_project_permission(db_session, study_id, "PM", "subjects", "read")
|
||||
elapsed_time = time.time() - start_time
|
||||
|
||||
# 平均每次权限检查的时间
|
||||
avg_time = elapsed_time / 100
|
||||
print(f"\n基准测试（无缓存）：{elapsed_time:.3f}s，平均 {avg_time*1000:.2f}ms/次")
|
||||
|
||||
# 基准测试应该在合理的时间范围内
|
||||
assert elapsed_time < 10, "权限检查性能过低"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_permission_check_performance_with_cache(db_session: AsyncSession, study_id: uuid.UUID):
|
||||
"""性能测试：权限检查性能（有缓存）"""
|
||||
# 清除缓存
|
||||
cache = get_permission_cache()
|
||||
cache.clear_all()
|
||||
|
||||
# 第一次调用会触发缓存填充
|
||||
await role_has_project_permission(db_session, study_id, "PM", "subjects", "read")
|
||||
|
||||
# 执行权限检查 100 次（应该都命中缓存）
|
||||
start_time = time.time()
|
||||
for _ in range(100):
|
||||
await role_has_project_permission(db_session, study_id, "PM", "subjects", "read")
|
||||
elapsed_time = time.time() - start_time
|
||||
|
||||
# 平均每次权限检查的时间
|
||||
avg_time = elapsed_time / 100
|
||||
print(f"\n缓存测试（有缓存）：{elapsed_time:.3f}s，平均 {avg_time*1000:.2f}ms/次")
|
||||
|
||||
# 缓存命中应该非常快
|
||||
assert elapsed_time < 1, "缓存性能不足"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_cache_hit_rate(db_session: AsyncSession, study_id: uuid.UUID):
|
||||
"""缓存效率测试：缓存命中率"""
|
||||
cache = get_permission_cache()
|
||||
cache.clear_all()
|
||||
|
||||
# 执行权限检查，模拟实际场景
|
||||
# 第一次调用：缓存未命中
|
||||
await role_has_project_permission(db_session, study_id, "PM", "subjects", "read")
|
||||
|
||||
# 后续调用：缓存命中
|
||||
for _ in range(99):
|
||||
await role_has_project_permission(db_session, study_id, "PM", "subjects", "read")
|
||||
|
||||
# 缓存统计
|
||||
stats = cache.get_cache_stats()
|
||||
print(f"\n缓存统计：{stats}")
|
||||
|
||||
# 验证缓存已被使用
|
||||
assert stats["project_permissions_count"] > 0, "缓存未被使用"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_cache_invalidation(db_session: AsyncSession, study_id: uuid.UUID):
|
||||
"""缓存失效测试"""
|
||||
cache = get_permission_cache()
|
||||
cache.clear_all()
|
||||
|
||||
# 填充缓存
|
||||
await role_has_project_permission(db_session, study_id, "PM", "subjects", "read")
|
||||
assert cache.get_cache_stats()["project_permissions_count"] == 1
|
||||
|
||||
# 失效缓存
|
||||
cache.invalidate_project_permissions(study_id)
|
||||
assert cache.get_cache_stats()["project_permissions_count"] == 0
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_concurrent_permission_checks(db_session: AsyncSession, study_id: uuid.UUID):
|
||||
"""并发权限检查测试"""
|
||||
cache = get_permission_cache()
|
||||
cache.clear_all()
|
||||
|
||||
# 并发执行权限检查
|
||||
async def check_permission():
|
||||
return await role_has_project_permission(db_session, study_id, "PM", "subjects", "read")
|
||||
|
||||
start_time = time.time()
|
||||
tasks = [check_permission() for _ in range(100)]
|
||||
results = await asyncio.gather(*tasks)
|
||||
elapsed_time = time.time() - start_time
|
||||
|
||||
# 所有检查都应该返回相同的结果
|
||||
assert all(results), "并发权限检查失败"
|
||||
print(f"\n并发测试（100个并发请求）：{elapsed_time:.3f}s")
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_cache_expiration(db_session: AsyncSession, study_id: uuid.UUID):
|
||||
"""缓存过期测试"""
|
||||
# 创建一个 TTL 很短的缓存
|
||||
cache = PermissionCache(default_ttl=1)
|
||||
set_permission_cache(cache)
|
||||
|
||||
# 填充缓存
|
||||
permissions = await cache.get_project_role_permissions(db_session, study_id, ttl=1)
|
||||
assert permissions is not None
|
||||
|
||||
# 等待缓存过期
|
||||
await asyncio.sleep(1.1)
|
||||
|
||||
# 再次获取，应该重新从数据库查询
|
||||
permissions2 = await cache.get_project_role_permissions(db_session, study_id, ttl=1)
|
||||
assert permissions2 is not None
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_list_operation_performance(db_session: AsyncSession, study_id: uuid.UUID):
|
||||
"""列表操作性能测试
|
||||
|
||||
模拟列表操作中的权限检查，每个项目都需要进行权限检查。
|
||||
"""
|
||||
cache = get_permission_cache()
|
||||
cache.clear_all()
|
||||
|
||||
# 模拟列表操作：检查 50 个项目的权限
|
||||
start_time = time.time()
|
||||
for i in range(50):
|
||||
# 每个项目都需要检查权限
|
||||
await role_has_project_permission(db_session, study_id, "PM", "subjects", "read")
|
||||
elapsed_time = time.time() - start_time
|
||||
|
||||
print(f"\n列表操作性能（50个项目）：{elapsed_time:.3f}s，平均 {elapsed_time/50*1000:.2f}ms/项")
|
||||
|
||||
# 列表操作应该在合理的时间范围内
|
||||
assert elapsed_time < 5, "列表操作性能过低"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_api_permission_check_performance(db_session: AsyncSession, study_id: uuid.UUID):
|
||||
"""API权限检查性能测试"""
|
||||
cache = get_permission_cache()
|
||||
cache.clear_all()
|
||||
|
||||
# 执行 API 权限检查 100 次
|
||||
start_time = time.time()
|
||||
for _ in range(100):
|
||||
await role_has_api_permission(db_session, study_id, "PM", "POST:/subjects")
|
||||
elapsed_time = time.time() - start_time
|
||||
|
||||
avg_time = elapsed_time / 100
|
||||
print(f"\nAPI权限检查性能：{elapsed_time:.3f}s，平均 {avg_time*1000:.2f}ms/次")
|
||||
|
||||
assert elapsed_time < 10, "API权限检查性能过低"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_permission_matrix_caching(db_session: AsyncSession, study_id: uuid.UUID):
|
||||
"""权限矩阵缓存测试"""
|
||||
cache = get_permission_cache()
|
||||
cache.clear_all()
|
||||
|
||||
# 第一次调用：从数据库查询
|
||||
start_time = time.time()
|
||||
permissions1 = await cache.get_project_role_permissions(db_session, study_id)
|
||||
first_call_time = time.time() - start_time
|
||||
|
||||
# 第二次调用：从缓存获取
|
||||
start_time = time.time()
|
||||
permissions2 = await cache.get_project_role_permissions(db_session, study_id)
|
||||
second_call_time = time.time() - start_time
|
||||
|
||||
# 验证结果相同
|
||||
assert permissions1 == permissions2
|
||||
|
||||
# 缓存调用应该快得多
|
||||
print(f"\n权限矩阵缓存：第一次 {first_call_time*1000:.2f}ms，第二次 {second_call_time*1000:.2f}ms")
|
||||
assert second_call_time < first_call_time / 2, "缓存性能不足"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_member_role_caching(db_session: AsyncSession, study_id: uuid.UUID, user_id: uuid.UUID):
|
||||
"""成员角色缓存测试"""
|
||||
cache = get_permission_cache()
|
||||
cache.clear_all()
|
||||
|
||||
# 第一次调用：从数据库查询
|
||||
start_time = time.time()
|
||||
role1 = await cache.get_member_role(db_session, study_id, user_id)
|
||||
first_call_time = time.time() - start_time
|
||||
|
||||
# 第二次调用：从缓存获取
|
||||
start_time = time.time()
|
||||
role2 = await cache.get_member_role(db_session, study_id, user_id)
|
||||
second_call_time = time.time() - start_time
|
||||
|
||||
# 验证结果相同
|
||||
assert role1 == role2
|
||||
|
||||
# 缓存调用应该快得多
|
||||
print(f"\n成员角色缓存：第一次 {first_call_time*1000:.2f}ms，第二次 {second_call_time*1000:.2f}ms")
|
||||
assert second_call_time < first_call_time / 2, "缓存性能不足"
|
||||
@@ -1,245 +0,0 @@
|
||||
"""安全测试：权限系统安全性验证
|
||||
|
||||
测试权限系统的安全性，包括：
|
||||
- 权限检查遗漏检测
|
||||
- 权限配置完整性
|
||||
- 缓存失效场景
|
||||
- 成员停用后的权限检查
|
||||
- 权限更新后的立即生效
|
||||
- 并发权限检查一致性
|
||||
"""
|
||||
|
||||
import uuid
|
||||
|
||||
import pytest
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from app.core.api_permissions import API_ENDPOINT_PERMISSIONS
|
||||
from app.core.permission_cache import get_permission_cache
|
||||
from app.core.project_permissions import (
|
||||
role_has_api_permission,
|
||||
role_has_project_permission,
|
||||
replace_project_role_permissions,
|
||||
replace_api_endpoint_permissions,
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_all_endpoints_have_permission_checks():
|
||||
"""测试所有端点都配置了权限检查
|
||||
|
||||
这个测试验证 API_ENDPOINT_PERMISSIONS 中的所有端点都有完整的权限配置。
|
||||
"""
|
||||
for endpoint_key, config in API_ENDPOINT_PERMISSIONS.items():
|
||||
# 验证必需字段
|
||||
assert "module" in config, f"Missing 'module' in {endpoint_key}"
|
||||
assert "action" in config, f"Missing 'action' in {endpoint_key}"
|
||||
assert "description" in config, f"Missing 'description' in {endpoint_key}"
|
||||
assert "default_roles" in config, f"Missing 'default_roles' in {endpoint_key}"
|
||||
|
||||
# 验证字段值
|
||||
assert config["action"] in ["read", "write"], f"Invalid action in {endpoint_key}"
|
||||
assert isinstance(config["default_roles"], list), f"default_roles must be list in {endpoint_key}"
|
||||
assert len(config["default_roles"]) > 0, f"default_roles must not be empty in {endpoint_key}"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_all_endpoints_in_api_permissions():
|
||||
"""测试所有端点都在 API_ENDPOINT_PERMISSIONS 中定义"""
|
||||
# 这个测试确保没有端点被遗漏
|
||||
assert len(API_ENDPOINT_PERMISSIONS) > 0, "API_ENDPOINT_PERMISSIONS is empty"
|
||||
assert len(API_ENDPOINT_PERMISSIONS) >= 94, "Missing endpoints in API_ENDPOINT_PERMISSIONS"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_all_endpoints_have_default_roles():
|
||||
"""测试所有端点都有默认角色配置"""
|
||||
valid_roles = {"PM", "CRA", "PV", "MEDICAL_REVIEW", "IMP", "QA"}
|
||||
|
||||
for endpoint_key, config in API_ENDPOINT_PERMISSIONS.items():
|
||||
default_roles = config.get("default_roles", [])
|
||||
assert len(default_roles) > 0, f"No default roles for {endpoint_key}"
|
||||
|
||||
for role in default_roles:
|
||||
assert role in valid_roles, f"Invalid role {role} in {endpoint_key}"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_inactive_member_permission_denied(db_session: AsyncSession, study_id: uuid.UUID):
|
||||
"""测试成员停用后的权限检查
|
||||
|
||||
当成员被停用时，应该拒绝其权限请求。
|
||||
"""
|
||||
# 这个测试需要创建一个停用的成员
|
||||
# 然后验证权限检查返回 False
|
||||
# 具体实现取决于成员模型的设计
|
||||
|
||||
# 暂时跳过，等待成员管理的完整实现
|
||||
pass
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_permission_update_immediate_effect(db_session: AsyncSession, study_id: uuid.UUID):
|
||||
"""测试权限更新后的立即生效
|
||||
|
||||
当权限被更新时，新的权限应该立即生效。
|
||||
"""
|
||||
# 获取初始权限
|
||||
initial_allowed = await role_has_project_permission(db_session, study_id, "CRA", "subjects", "write")
|
||||
|
||||
# 更新权限：允许 CRA 写入 subjects
|
||||
new_permissions = {
|
||||
"PM": {"subjects": {"read": True, "write": True}},
|
||||
"CRA": {"subjects": {"read": True, "write": True}},
|
||||
}
|
||||
await replace_project_role_permissions(db_session, study_id, new_permissions)
|
||||
|
||||
# 验证权限立即生效
|
||||
updated_allowed = await role_has_project_permission(db_session, study_id, "CRA", "subjects", "write")
|
||||
assert updated_allowed is True, "权限更新未立即生效"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_cache_invalidation_on_permission_update(db_session: AsyncSession, study_id: uuid.UUID):
|
||||
"""测试权限更新时的缓存失效
|
||||
|
||||
当权限被更新时，相关的缓存应该被失效。
|
||||
"""
|
||||
cache = get_permission_cache()
|
||||
cache.clear_all()
|
||||
|
||||
# 填充缓存
|
||||
await role_has_project_permission(db_session, study_id, "PM", "subjects", "read")
|
||||
assert cache.get_cache_stats()["project_permissions_count"] == 1
|
||||
|
||||
# 更新权限
|
||||
new_permissions = {
|
||||
"PM": {"subjects": {"read": True, "write": True}},
|
||||
}
|
||||
await replace_project_role_permissions(db_session, study_id, new_permissions)
|
||||
|
||||
# 验证缓存已失效
|
||||
assert cache.get_cache_stats()["project_permissions_count"] == 0
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_cache_invalidation_on_member_update(db_session: AsyncSession, study_id: uuid.UUID, user_id: uuid.UUID):
|
||||
"""测试成员更新时的缓存失效
|
||||
|
||||
当成员角色被更新时，相关的缓存应该被失效。
|
||||
"""
|
||||
cache = get_permission_cache()
|
||||
cache.clear_all()
|
||||
|
||||
# 填充缓存
|
||||
await role_has_project_permission(db_session, study_id, "PM", "subjects", "read")
|
||||
initial_stats = cache.get_cache_stats()
|
||||
|
||||
# 失效成员角色缓存
|
||||
cache.invalidate_member_role(study_id, user_id)
|
||||
|
||||
# 验证成员角色缓存已失效
|
||||
# （项目权限缓存应该保留）
|
||||
assert cache.get_cache_stats()["member_role_count"] == 0
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_concurrent_permission_checks_consistency(db_session: AsyncSession, study_id: uuid.UUID):
|
||||
"""测试并发权限检查的一致性
|
||||
|
||||
多个并发请求检查权限时，结果应该一致。
|
||||
"""
|
||||
import asyncio
|
||||
|
||||
async def check_permission():
|
||||
return await role_has_project_permission(db_session, study_id, "PM", "subjects", "read")
|
||||
|
||||
# 并发执行权限检查
|
||||
tasks = [check_permission() for _ in range(10)]
|
||||
results = await asyncio.gather(*tasks)
|
||||
|
||||
# 所有结果应该相同
|
||||
assert all(r == results[0] for r in results), "并发权限检查结果不一致"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_api_permission_priority_over_module(db_session: AsyncSession, study_id: uuid.UUID):
|
||||
"""测试接口级权限优先于模块级权限
|
||||
|
||||
当同时配置了接口级和模块级权限时，接口级权限应该优先。
|
||||
"""
|
||||
# 设置模块级权限：允许 CRA 读取 subjects
|
||||
module_permissions = {
|
||||
"PM": {"subjects": {"read": True, "write": True}},
|
||||
"CRA": {"subjects": {"read": True, "write": False}},
|
||||
}
|
||||
await replace_project_role_permissions(db_session, study_id, module_permissions)
|
||||
|
||||
# 设置接口级权限：拒绝 CRA 读取 subjects
|
||||
api_permissions = {
|
||||
"PM": {"GET:/subjects": True},
|
||||
"CRA": {"GET:/subjects": False},
|
||||
}
|
||||
await replace_api_endpoint_permissions(db_session, study_id, api_permissions)
|
||||
|
||||
# 验证接口级权限优先
|
||||
allowed = await role_has_api_permission(db_session, study_id, "CRA", "GET:/subjects")
|
||||
assert allowed is False, "接口级权限未优先于模块级权限"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_admin_role_always_allowed(db_session: AsyncSession, study_id: uuid.UUID):
|
||||
"""测试 ADMIN 角色总是被允许
|
||||
|
||||
ADMIN 角色应该在所有权限检查中都被允许。
|
||||
"""
|
||||
# 即使没有配置权限，ADMIN 也应该被允许
|
||||
allowed = await role_has_project_permission(db_session, study_id, "ADMIN", "subjects", "write")
|
||||
assert allowed is True, "ADMIN 角色权限检查失败"
|
||||
|
||||
# API 权限检查也应该允许 ADMIN
|
||||
api_allowed = await role_has_api_permission(db_session, study_id, "ADMIN", "POST:/subjects")
|
||||
assert api_allowed is True, "ADMIN 角色 API 权限检查失败"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_none_role_permission_denied(db_session: AsyncSession, study_id: uuid.UUID):
|
||||
"""测试 None 角色被拒绝
|
||||
|
||||
当角色为 None 时，权限检查应该返回 False。
|
||||
"""
|
||||
allowed = await role_has_project_permission(db_session, study_id, None, "subjects", "read")
|
||||
assert allowed is False, "None 角色权限检查失败"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_unknown_endpoint_permission_denied(db_session: AsyncSession, study_id: uuid.UUID):
|
||||
"""测试未知端点被拒绝
|
||||
|
||||
当端点不存在时，权限检查应该返回 False。
|
||||
"""
|
||||
allowed = await role_has_api_permission(db_session, study_id, "PM", "GET:/unknown-endpoint")
|
||||
assert allowed is False, "未知端点权限检查失败"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_permission_matrix_consistency(db_session: AsyncSession, study_id: uuid.UUID):
|
||||
"""测试权限矩阵的一致性
|
||||
|
||||
权限矩阵应该包含所有角色和端点的权限信息。
|
||||
"""
|
||||
from app.core.project_permissions import get_project_role_permissions
|
||||
|
||||
permissions = await get_project_role_permissions(db_session, study_id)
|
||||
|
||||
# 验证矩阵结构
|
||||
assert isinstance(permissions, dict), "权限矩阵格式错误"
|
||||
|
||||
# 验证所有角色都存在
|
||||
for role in ["PM", "CRA", "PV", "MEDICAL_REVIEW", "IMP", "QA"]:
|
||||
assert role in permissions, f"缺少角色 {role}"
|
||||
|
||||
# 验证每个角色都有权限配置
|
||||
for role, role_permissions in permissions.items():
|
||||
assert isinstance(role_permissions, dict), f"角色 {role} 的权限配置格式错误"
|
||||
assert len(role_permissions) > 0, f"角色 {role} 没有权限配置"
|
||||
@@ -0,0 +1,230 @@
|
||||
"""权限模板服务单元测试"""
|
||||
from __future__ import annotations
|
||||
|
||||
import uuid
|
||||
|
||||
import pytest
|
||||
|
||||
from app.models.permission_template import PermissionTemplate, TemplateType
|
||||
from app.schemas.permission_template import PermissionTemplateCreate, PermissionTemplateUpdate
|
||||
from app.services.permission_template_service import PermissionTemplateService
|
||||
|
||||
|
||||
SAMPLE_PERMISSIONS = {
|
||||
"PM": {
|
||||
"subjects:create": True,
|
||||
"subjects:read": True,
|
||||
"subjects:update": True,
|
||||
"subjects:delete": False,
|
||||
},
|
||||
"CRA": {
|
||||
"subjects:create": False,
|
||||
"subjects:read": True,
|
||||
"subjects:update": False,
|
||||
"subjects:delete": False,
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def user_id() -> uuid.UUID:
|
||||
return uuid.uuid4()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 创建模板
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_create_template_success(db_session, user_id):
|
||||
payload = PermissionTemplateCreate(
|
||||
name="测试模板",
|
||||
description="用于测试",
|
||||
template_type=TemplateType.CUSTOM,
|
||||
permissions=SAMPLE_PERMISSIONS,
|
||||
)
|
||||
template = await PermissionTemplateService.create_template(db_session, payload, user_id)
|
||||
|
||||
assert template.id is not None
|
||||
assert template.name == "测试模板"
|
||||
assert template.is_system is False
|
||||
assert template.template_type == TemplateType.CUSTOM
|
||||
assert template.permissions == SAMPLE_PERMISSIONS
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_create_template_creates_version_1(db_session, user_id):
|
||||
payload = PermissionTemplateCreate(
|
||||
name="版本测试",
|
||||
template_type=TemplateType.CUSTOM,
|
||||
permissions=SAMPLE_PERMISSIONS,
|
||||
)
|
||||
template = await PermissionTemplateService.create_template(db_session, payload, user_id)
|
||||
|
||||
await db_session.refresh(template, ["versions"])
|
||||
assert len(template.versions) == 1
|
||||
assert template.versions[0].version == 1
|
||||
assert template.versions[0].change_log == "初始版本"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_create_template_invalid_endpoint_key(db_session, user_id):
|
||||
payload = PermissionTemplateCreate(
|
||||
name="无效模板",
|
||||
template_type=TemplateType.CUSTOM,
|
||||
permissions={"PM": {"nonexistent:action": True}},
|
||||
)
|
||||
with pytest.raises(ValueError, match="权限操作 nonexistent:action 不存在"):
|
||||
await PermissionTemplateService.create_template(db_session, payload, user_id)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_create_template_invalid_permission_value(db_session, user_id):
|
||||
payload = PermissionTemplateCreate(
|
||||
name="无效值模板",
|
||||
template_type=TemplateType.CUSTOM,
|
||||
permissions={"PM": {"subjects:read": "yes"}}, # type: ignore
|
||||
)
|
||||
with pytest.raises(ValueError, match="必须是布尔类型"):
|
||||
await PermissionTemplateService.create_template(db_session, payload, user_id)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 获取模板
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_template_exists(db_session, user_id):
|
||||
payload = PermissionTemplateCreate(
|
||||
name="获取测试",
|
||||
template_type=TemplateType.ROLE,
|
||||
permissions=SAMPLE_PERMISSIONS,
|
||||
)
|
||||
created = await PermissionTemplateService.create_template(db_session, payload, user_id)
|
||||
fetched = await PermissionTemplateService.get_template(db_session, created.id)
|
||||
|
||||
assert fetched is not None
|
||||
assert fetched.id == created.id
|
||||
assert fetched.name == "获取测试"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_template_not_found(db_session):
|
||||
result = await PermissionTemplateService.get_template(db_session, uuid.uuid4())
|
||||
assert result is None
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 列表模板
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_list_templates_filter_by_type(db_session, user_id):
|
||||
await PermissionTemplateService.create_template(
|
||||
db_session,
|
||||
PermissionTemplateCreate(name="角色模板", template_type=TemplateType.ROLE, permissions=SAMPLE_PERMISSIONS),
|
||||
user_id,
|
||||
)
|
||||
await PermissionTemplateService.create_template(
|
||||
db_session,
|
||||
PermissionTemplateCreate(name="场景模板", template_type=TemplateType.SCENARIO, permissions=SAMPLE_PERMISSIONS),
|
||||
user_id,
|
||||
)
|
||||
|
||||
role_templates = await PermissionTemplateService.list_templates(db_session, template_type=TemplateType.ROLE)
|
||||
assert all(t.template_type == TemplateType.ROLE for t in role_templates)
|
||||
assert len(role_templates) >= 1
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 更新模板
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_update_template_name(db_session, user_id):
|
||||
created = await PermissionTemplateService.create_template(
|
||||
db_session,
|
||||
PermissionTemplateCreate(name="旧名称", template_type=TemplateType.CUSTOM, permissions=SAMPLE_PERMISSIONS),
|
||||
user_id,
|
||||
)
|
||||
|
||||
updated = await PermissionTemplateService.update_template(
|
||||
db_session, created.id, PermissionTemplateUpdate(name="新名称")
|
||||
)
|
||||
assert updated.name == "新名称"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_update_template_permissions_creates_new_version(db_session, user_id):
|
||||
created = await PermissionTemplateService.create_template(
|
||||
db_session,
|
||||
PermissionTemplateCreate(name="版本测试2", template_type=TemplateType.CUSTOM, permissions=SAMPLE_PERMISSIONS),
|
||||
user_id,
|
||||
)
|
||||
|
||||
new_perms = {"PM": {"subjects:read": True}}
|
||||
await PermissionTemplateService.update_template(
|
||||
db_session, created.id, PermissionTemplateUpdate(permissions=new_perms)
|
||||
)
|
||||
|
||||
updated = await PermissionTemplateService.get_template(db_session, created.id)
|
||||
await db_session.refresh(updated, ["versions"])
|
||||
assert len(updated.versions) == 2
|
||||
assert updated.versions[-1].version == 2
|
||||
assert updated.permissions == new_perms
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_update_system_template_raises(db_session):
|
||||
system_template = PermissionTemplate(
|
||||
name="系统模板",
|
||||
template_type=TemplateType.ROLE,
|
||||
is_system=True,
|
||||
permissions=SAMPLE_PERMISSIONS,
|
||||
)
|
||||
db_session.add(system_template)
|
||||
await db_session.commit()
|
||||
await db_session.refresh(system_template)
|
||||
|
||||
with pytest.raises(ValueError, match="不允许修改系统预设模板"):
|
||||
await PermissionTemplateService.update_template(
|
||||
db_session, system_template.id, PermissionTemplateUpdate(name="改名")
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 删除模板
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_delete_template_success(db_session, user_id):
|
||||
created = await PermissionTemplateService.create_template(
|
||||
db_session,
|
||||
PermissionTemplateCreate(name="待删除", template_type=TemplateType.CUSTOM, permissions=SAMPLE_PERMISSIONS),
|
||||
user_id,
|
||||
)
|
||||
|
||||
await PermissionTemplateService.delete_template(db_session, created.id)
|
||||
assert await PermissionTemplateService.get_template(db_session, created.id) is None
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_delete_system_template_raises(db_session):
|
||||
system_template = PermissionTemplate(
|
||||
name="系统模板2",
|
||||
template_type=TemplateType.ROLE,
|
||||
is_system=True,
|
||||
permissions=SAMPLE_PERMISSIONS,
|
||||
)
|
||||
db_session.add(system_template)
|
||||
await db_session.commit()
|
||||
await db_session.refresh(system_template)
|
||||
|
||||
with pytest.raises(ValueError, match="不允许删除系统预设模板"):
|
||||
await PermissionTemplateService.delete_template(db_session, system_template.id)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_delete_nonexistent_template_raises(db_session):
|
||||
with pytest.raises(ValueError, match="不存在"):
|
||||
await PermissionTemplateService.delete_template(db_session, uuid.uuid4())
|
||||
@@ -9,7 +9,6 @@ from app.core.project_permissions import (
|
||||
get_missing_prerequisites,
|
||||
)
|
||||
from app.models.api_endpoint_permission import ApiEndpointPermission
|
||||
from app.models.study_role_permission import StudyRolePermission
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
|
||||
@@ -1,166 +0,0 @@
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
|
||||
|
||||
def test_project_setup_routes_keep_project_admin_and_pm_defaults():
|
||||
source = (ROOT / "app/api/v1/studies.py").read_text(encoding="utf-8")
|
||||
|
||||
assert 'require_study_roles(["PM"])' not in source
|
||||
assert 'require_study_roles(["ADMIN", "PM"])' not in source
|
||||
assert 'require_roles(["ADMIN"])' in source
|
||||
|
||||
|
||||
def test_project_permission_routes_are_registered():
|
||||
source = (ROOT / "app/api/v1/router.py").read_text(encoding="utf-8")
|
||||
|
||||
assert "project_permissions" in source
|
||||
assert 'prefix="/studies/{study_id}/permissions"' in source
|
||||
|
||||
|
||||
def test_project_permission_updates_use_project_member_write_permission():
|
||||
source = (ROOT / "app/api/v1/project_permissions.py").read_text(encoding="utf-8")
|
||||
|
||||
assert 'require_study_permission("project_members", "write")' in source
|
||||
assert 'require_roles(["ADMIN"])' not in source
|
||||
|
||||
|
||||
def test_project_member_writes_use_saved_permission_matrix():
|
||||
source = (ROOT / "app/api/v1/members.py").read_text(encoding="utf-8")
|
||||
|
||||
assert 'require_study_permission("project_members", "write")' in source
|
||||
assert 'require_study_roles(["ADMIN", "PM"])' not in source
|
||||
|
||||
|
||||
def test_project_member_candidates_use_project_write_permission():
|
||||
source = (ROOT / "app/api/v1/members.py").read_text(encoding="utf-8")
|
||||
user_crud_source = (ROOT / "app/crud/user.py").read_text(encoding="utf-8")
|
||||
|
||||
assert '"/candidates"' in source
|
||||
assert 'response_model=list[UserResponse]' in source
|
||||
assert 'dependencies=[Depends(require_study_permission("project_members", "write"))]' in source
|
||||
assert "list_active_member_candidates_for_study" in source
|
||||
assert "User.status == UserStatus.ACTIVE" in user_crud_source
|
||||
assert ".where(~existing_member)" in user_crud_source
|
||||
|
||||
|
||||
def test_project_member_mutation_blocks_self_and_higher_role_changes():
|
||||
source = (ROOT / "app/api/v1/members.py").read_text(encoding="utf-8")
|
||||
|
||||
assert "PROJECT_ROLE_RANK" in source
|
||||
assert 'if target_member and target_member.user_id == current_user.id:' in source
|
||||
assert "不能修改自己的项目成员权限" in source
|
||||
assert "不能修改系统管理员账号的项目权限" in source
|
||||
assert 'if _role_value(target_user) == "ADMIN":' in source
|
||||
assert "不能修改权限高于自己的项目成员" in source
|
||||
assert "不能授予高于自己的项目角色" in source
|
||||
assert "_ensure_member_mutation_allowed(" in source
|
||||
|
||||
|
||||
def test_business_modules_use_saved_permission_matrix():
|
||||
expected = {
|
||||
"overview.py": [
|
||||
'require_study_permission("project_overview", "read")',
|
||||
],
|
||||
"project_milestones.py": [
|
||||
'require_study_permission("project_milestones", "read")',
|
||||
'require_study_permission("project_milestones", "write")',
|
||||
],
|
||||
"subjects.py": [
|
||||
'require_study_permission("subjects", "read")',
|
||||
'require_study_permission("subjects", "write")',
|
||||
],
|
||||
"material_equipments.py": [
|
||||
'require_study_permission("materials", "read")',
|
||||
'require_study_permission("materials", "write")',
|
||||
],
|
||||
"startup.py": [
|
||||
'require_study_permission("startup_ethics", "read")',
|
||||
'require_study_permission("startup_ethics", "write")',
|
||||
'require_study_permission("startup_auth", "read")',
|
||||
'require_study_permission("startup_auth", "write")',
|
||||
],
|
||||
"aes.py": [
|
||||
'require_study_permission("risk_issues", "read")',
|
||||
'require_study_permission("risk_issues", "write")',
|
||||
],
|
||||
"monitoring_visit_issues.py": [
|
||||
'require_study_permission("monitoring_audit", "read")',
|
||||
'require_study_permission("monitoring_audit", "write")',
|
||||
],
|
||||
}
|
||||
|
||||
for filename, checks in expected.items():
|
||||
source = (ROOT / f"app/api/v1/{filename}").read_text(encoding="utf-8")
|
||||
for check in checks:
|
||||
assert check in source
|
||||
|
||||
ae_source = (ROOT / "app/api/v1/aes.py").read_text(encoding="utf-8")
|
||||
assert "ALLOWED_CREATE_ROLES" not in ae_source
|
||||
assert "ALLOWED_UPDATE_ROLES" not in ae_source
|
||||
assert 'member_role not in {"PM", "PV"}' not in ae_source
|
||||
|
||||
|
||||
def test_fee_contracts_use_saved_permission_matrix():
|
||||
source = (ROOT / "app/api/v1/fees_contracts.py").read_text(encoding="utf-8")
|
||||
|
||||
assert 'role_has_project_permission(db, project_id, membership.role_in_study, "fees", action)' in source
|
||||
assert 'if write and membership.role_in_study not in {"ADMIN", "PM"}' not in source
|
||||
|
||||
|
||||
def test_attachment_deletes_use_saved_permission_matrix():
|
||||
source = (ROOT / "app/api/v1/attachments.py").read_text(encoding="utf-8")
|
||||
fees_source = (ROOT / "app/api/v1/fees_attachments.py").read_text(encoding="utf-8")
|
||||
|
||||
assert "_permission_module_for_entity(entity_type)" in source
|
||||
assert '"knowledge_note": "shared_library"' in source
|
||||
assert "_permission_module_for_entity(attachment.entity_type)" in source
|
||||
assert 'getattr(membership, "role_in_study", None) == "PM"' not in source
|
||||
assert 'role_has_project_permission(db, project_id, membership.role_in_study, "fees", action)' in fees_source
|
||||
assert 'getattr(membership, "role_in_study", None) == "PM"' not in fees_source
|
||||
|
||||
|
||||
def test_faq_uses_dedicated_permission_matrix():
|
||||
source = (ROOT / "app/api/v1/faqs.py").read_text(encoding="utf-8")
|
||||
category_source = (ROOT / "app/api/v1/faq_categories.py").read_text(encoding="utf-8")
|
||||
|
||||
assert 'role_has_project_permission(db, study_id, member.role_in_study, "faq", action)' in source
|
||||
assert 'role_has_project_permission(db, study_id, member.role_in_study, "faq", action)' in category_source
|
||||
assert '"etmf"' not in source
|
||||
assert '"etmf"' not in category_source
|
||||
assert 'member_role != "PM"' not in source
|
||||
assert 'member_role != "PM"' not in category_source
|
||||
|
||||
|
||||
def test_audit_logs_use_saved_permission_matrix():
|
||||
source = (ROOT / "app/api/v1/audit_logs.py").read_text(encoding="utf-8")
|
||||
get_block = source[source.index("@router.get("):source.index("@router.delete(")]
|
||||
|
||||
assert 'require_study_permission("audit_export", "read")' in get_block
|
||||
assert 'require_study_member()' not in get_block
|
||||
|
||||
|
||||
def test_document_service_uses_saved_permission_matrix():
|
||||
source = (ROOT / "app/services/document_service.py").read_text(encoding="utf-8")
|
||||
|
||||
assert 'role_has_project_permission' in source
|
||||
assert '"file_versions"' in source
|
||||
assert 'rbac.is_allowed' not in source
|
||||
|
||||
|
||||
def test_remaining_business_reads_use_saved_permission_matrix():
|
||||
expected = {
|
||||
"finance_contracts.py": ['require_study_permission("fees", "read")'],
|
||||
"drug_shipments.py": ['require_study_permission("materials", "read")'],
|
||||
"knowledge_notes.py": ['require_study_permission("shared_library", "read")'],
|
||||
"subject_histories.py": ['require_study_permission("subjects", "read")'],
|
||||
"subject_pds.py": ['require_study_permission("risk_issues", "read")'],
|
||||
"visits.py": ['require_study_permission("subjects", "read")'],
|
||||
"attachments.py": ['_permission_module_for_entity(entity_type)'],
|
||||
}
|
||||
|
||||
for filename, checks in expected.items():
|
||||
source = (ROOT / f"app/api/v1/{filename}").read_text(encoding="utf-8")
|
||||
for check in checks:
|
||||
assert check in source
|
||||
@@ -1,189 +0,0 @@
|
||||
from app.core.project_permissions import PROJECT_PERMISSION_MODULES, normalize_permission_matrix
|
||||
|
||||
|
||||
def test_normalize_permission_matrix_makes_admin_fixed_full_access():
|
||||
matrix = normalize_permission_matrix({"ADMIN": {}})
|
||||
|
||||
for module in PROJECT_PERMISSION_MODULES:
|
||||
assert matrix["ADMIN"][module["key"]] == {
|
||||
"read": True,
|
||||
"write": module.get("writable") is not False,
|
||||
}
|
||||
|
||||
|
||||
def test_normalize_permission_matrix_keeps_admin_fixed_even_when_disabled():
|
||||
payload = {
|
||||
"ADMIN": {
|
||||
module["key"]: {"read": False, "write": False}
|
||||
for module in PROJECT_PERMISSION_MODULES
|
||||
}
|
||||
}
|
||||
|
||||
matrix = normalize_permission_matrix(payload)
|
||||
|
||||
for module in PROJECT_PERMISSION_MODULES:
|
||||
assert matrix["ADMIN"][module["key"]] == {
|
||||
"read": True,
|
||||
"write": module.get("writable") is not False,
|
||||
}
|
||||
|
||||
|
||||
def test_normalize_permission_matrix_write_implies_read():
|
||||
matrix = normalize_permission_matrix({
|
||||
"CRA": {
|
||||
"subjects": {"read": False, "write": True},
|
||||
}
|
||||
})
|
||||
|
||||
assert matrix["CRA"]["subjects"] == {"read": True, "write": True}
|
||||
|
||||
|
||||
def test_business_permission_modules_follow_project_menu():
|
||||
module_labels = [module["label"] for module in PROJECT_PERMISSION_MODULES]
|
||||
|
||||
assert module_labels == [
|
||||
"项目成员",
|
||||
"中心管理",
|
||||
"审计日志导出",
|
||||
"项目总览",
|
||||
"项目里程碑",
|
||||
"合同费用管理",
|
||||
"物资管理",
|
||||
"文件版本管理",
|
||||
"立项与伦理",
|
||||
"启动与授权",
|
||||
"参与者管理",
|
||||
"风险问题",
|
||||
"监查稽查",
|
||||
"eTMF",
|
||||
"FAQ",
|
||||
"共享库",
|
||||
]
|
||||
|
||||
|
||||
def test_default_business_permissions_match_role_responsibilities():
|
||||
matrix = normalize_permission_matrix()
|
||||
|
||||
assert matrix["PM"]["project_members"] == {"read": True, "write": True}
|
||||
assert matrix["PM"]["sites"] == {"read": True, "write": True}
|
||||
assert matrix["CRA"]["project_members"] == {"read": False, "write": False}
|
||||
assert matrix["CRA"]["audit_export"] == {"read": False, "write": False}
|
||||
assert matrix["PV"]["sites"] == {"read": False, "write": False}
|
||||
assert matrix["IMP"]["audit_export"] == {"read": False, "write": False}
|
||||
assert matrix["QA"]["audit_export"] == {"read": False, "write": False}
|
||||
assert matrix["CRA"]["project_milestones"] == {"read": True, "write": True}
|
||||
assert matrix["CRA"]["subjects"] == {"read": True, "write": True}
|
||||
assert matrix["CRA"]["monitoring_audit"] == {"read": True, "write": True}
|
||||
assert matrix["PV"]["risk_issues"] == {"read": True, "write": True}
|
||||
assert matrix["MEDICAL_REVIEW"]["project_overview"] == {"read": True, "write": False}
|
||||
assert matrix["MEDICAL_REVIEW"]["subjects"] == {"read": True, "write": False}
|
||||
assert matrix["MEDICAL_REVIEW"]["risk_issues"] == {"read": True, "write": True}
|
||||
assert matrix["MEDICAL_REVIEW"]["monitoring_audit"] == {"read": True, "write": False}
|
||||
assert matrix["MEDICAL_REVIEW"]["faq"] == {"read": True, "write": True}
|
||||
assert matrix["MEDICAL_REVIEW"]["shared_library"] == {"read": True, "write": True}
|
||||
assert matrix["MEDICAL_REVIEW"]["fees"] == {"read": False, "write": False}
|
||||
assert matrix["MEDICAL_REVIEW"]["materials"] == {"read": False, "write": False}
|
||||
assert matrix["IMP"]["materials"] == {"read": True, "write": True}
|
||||
assert matrix["PM"]["faq"] == {"read": True, "write": True}
|
||||
assert matrix["CRA"]["faq"] == {"read": True, "write": True}
|
||||
assert matrix["PM"]["project_overview"] == {"read": True, "write": False}
|
||||
assert matrix["CRA"]["project_overview"] == {"read": True, "write": False}
|
||||
assert matrix["PM"]["shared_library"] == {"read": True, "write": True}
|
||||
assert matrix["CRA"]["shared_library"] == {"read": True, "write": True}
|
||||
|
||||
|
||||
def test_project_overview_is_read_only_even_when_write_is_submitted():
|
||||
matrix = normalize_permission_matrix({
|
||||
"CRA": {
|
||||
"project_overview": {"read": False, "write": True},
|
||||
}
|
||||
})
|
||||
|
||||
assert matrix["CRA"]["project_overview"] == {"read": True, "write": False}
|
||||
|
||||
|
||||
def test_pm_can_be_granted_management_backend_permissions():
|
||||
matrix = normalize_permission_matrix({
|
||||
"PM": {
|
||||
"project_members": {"read": True, "write": True},
|
||||
"sites": {"read": True, "write": True},
|
||||
"audit_export": {"read": True, "write": True},
|
||||
}
|
||||
})
|
||||
|
||||
assert matrix["PM"]["project_members"] == {"read": True, "write": True}
|
||||
assert matrix["PM"]["sites"] == {"read": True, "write": True}
|
||||
assert matrix["PM"]["audit_export"] == {"read": True, "write": True}
|
||||
|
||||
|
||||
def test_non_pm_roles_cannot_be_granted_management_backend_permissions():
|
||||
management_modules = ["project_members", "sites", "audit_export"]
|
||||
matrix = normalize_permission_matrix({
|
||||
role: {
|
||||
module: {"read": True, "write": True}
|
||||
for module in management_modules
|
||||
}
|
||||
for role in ["CRA", "PV", "MEDICAL_REVIEW", "IMP", "QA"]
|
||||
})
|
||||
|
||||
for role in ["CRA", "PV", "MEDICAL_REVIEW", "IMP", "QA"]:
|
||||
for module in management_modules:
|
||||
assert matrix[role][module] == {"read": False, "write": False}
|
||||
|
||||
|
||||
def test_each_non_admin_role_can_toggle_each_writable_module():
|
||||
roles = ["PM", "CRA", "PV", "MEDICAL_REVIEW", "IMP", "QA"]
|
||||
management_modules = {"project_members", "sites", "audit_export"}
|
||||
writable_modules = [
|
||||
module["key"]
|
||||
for module in PROJECT_PERMISSION_MODULES
|
||||
if module.get("writable") is not False and module["key"] not in management_modules
|
||||
]
|
||||
|
||||
for role in roles:
|
||||
enabled = normalize_permission_matrix({
|
||||
role: {
|
||||
module: {"read": False, "write": True}
|
||||
for module in writable_modules
|
||||
}
|
||||
})
|
||||
disabled = normalize_permission_matrix({
|
||||
role: {
|
||||
module["key"]: {"read": False, "write": False}
|
||||
for module in PROJECT_PERMISSION_MODULES
|
||||
}
|
||||
})
|
||||
|
||||
for module in writable_modules:
|
||||
assert enabled[role][module] == {"read": True, "write": True}
|
||||
assert disabled[role][module] == {"read": False, "write": False}
|
||||
|
||||
|
||||
def test_read_only_modules_never_accept_write_for_any_role():
|
||||
read_only_modules = [
|
||||
module["key"]
|
||||
for module in PROJECT_PERMISSION_MODULES
|
||||
if module.get("writable") is False
|
||||
]
|
||||
payload = {
|
||||
role: {
|
||||
module: {"read": False, "write": True}
|
||||
for module in read_only_modules
|
||||
}
|
||||
for role in ["ADMIN", "PM", "CRA", "PV", "MEDICAL_REVIEW", "IMP", "QA"]
|
||||
}
|
||||
|
||||
matrix = normalize_permission_matrix(payload)
|
||||
|
||||
for role in ["ADMIN", "PM", "CRA", "PV", "MEDICAL_REVIEW", "IMP", "QA"]:
|
||||
for module in read_only_modules:
|
||||
assert matrix[role][module] == {"read": True, "write": False}
|
||||
|
||||
|
||||
def test_medical_review_default_permissions_are_independent_from_pv():
|
||||
matrix = normalize_permission_matrix()
|
||||
|
||||
assert matrix["PV"]["monitoring_audit"] == {"read": True, "write": True}
|
||||
assert matrix["MEDICAL_REVIEW"]["monitoring_audit"] == {"read": True, "write": False}
|
||||
assert matrix["PV"]["fees"] == {"read": True, "write": False}
|
||||
assert matrix["MEDICAL_REVIEW"]["fees"] == {"read": False, "write": False}
|
||||
Reference in New Issue
Block a user