Cheng Zhou
443 lines
14 KiB
Python
443 lines
14 KiB
Python
"""单元测试：API权限检查函数"""
|
|
|
|
import pytest
|
|
import uuid
|
|
from pathlib import Path
|
|
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.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
|
|
from app.core.permission_monitor import PermissionMonitor, set_permission_monitor
|
|
from app.models.api_endpoint_permission import ApiEndpointPermission
|
|
from app.models.study import Study
|
|
from app.models.user import UserRole
|
|
from app.schemas.member import StudyMemberCreate
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_api_permission_check_allowed(db_session: AsyncSession):
|
|
"""测试接口级权限检查 - 允许"""
|
|
study_id = uuid.uuid4()
|
|
|
|
# 创建权限记录
|
|
perm = ApiEndpointPermission(
|
|
study_id=study_id,
|
|
role="CRA",
|
|
endpoint_key="subjects:create",
|
|
allowed=True,
|
|
)
|
|
db_session.add(perm)
|
|
await db_session.commit()
|
|
|
|
# 验证权限（禁用前置权限检查）
|
|
result = await role_has_api_permission(
|
|
db_session, study_id, "CRA", "subjects:create", check_prerequisites=False
|
|
)
|
|
assert result is True
|
|
|
|
|
|
def test_all_backend_permission_guards_are_configurable():
|
|
"""确保后端实际鉴权使用的 operation key 都能在权限管理中配置。"""
|
|
api_dir = Path(__file__).resolve().parents[1] / "app" / "api" / "v1"
|
|
used_keys: set[str] = set()
|
|
for path in api_dir.rglob("*.py"):
|
|
source = path.read_text()
|
|
tree = ast.parse(source)
|
|
for node in ast.walk(tree):
|
|
if not isinstance(node, ast.Call):
|
|
continue
|
|
if isinstance(node.func, ast.Name) and node.func.id == "require_api_permission":
|
|
if node.args and isinstance(node.args[0], ast.Constant) and isinstance(node.args[0].value, str):
|
|
used_keys.add(node.args[0].value)
|
|
if isinstance(node.func, ast.Name) and node.func.id == "_ensure_project_access":
|
|
if len(node.args) >= 4 and isinstance(node.args[3], ast.Constant) and isinstance(node.args[3].value, str):
|
|
used_keys.add(node.args[3].value)
|
|
|
|
assert used_keys
|
|
assert used_keys <= set(API_ENDPOINT_PERMISSIONS)
|
|
|
|
|
|
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"
|
|
source = service_path.read_text()
|
|
|
|
assert '"create_document": "documents:create"' in source
|
|
assert '"create_version": "documents:update"' in source
|
|
assert '"delete_document": "documents:delete"' in source
|
|
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}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_default_matrix_covers_every_role_and_permission(db_session: AsyncSession):
|
|
"""逐一验证预设项目权限角色在每个权限上的默认矩阵。"""
|
|
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 set(matrix) == set(PROJECT_PERMISSION_ROLES)
|
|
for role in PROJECT_PERMISSION_ROLES:
|
|
assert set(matrix[role]) == set(API_ENDPOINT_PERMISSIONS)
|
|
for endpoint_key, config in API_ENDPOINT_PERMISSIONS.items():
|
|
assert matrix[role][endpoint_key]["allowed"] is (role in config["default_roles"])
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_default_matrix_round_trips_to_backend_checks(db_session: AsyncSession):
|
|
"""默认权限矩阵应与后端实际鉴权结果一致。"""
|
|
study_id = uuid.uuid4()
|
|
|
|
for role in PROJECT_PERMISSION_ROLES:
|
|
for endpoint_key, config in API_ENDPOINT_PERMISSIONS.items():
|
|
allowed = await role_has_api_permission(
|
|
db_session,
|
|
study_id,
|
|
role,
|
|
endpoint_key,
|
|
check_prerequisites=False,
|
|
)
|
|
assert allowed is (role in config["default_roles"])
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_full_permission_matrix_round_trips_to_backend_checks(db_session: AsyncSession):
|
|
"""逐一验证前端提交格式会落库，并被后端鉴权函数按相同结果读取。"""
|
|
cache = PermissionCache()
|
|
monitor = PermissionMonitor()
|
|
set_permission_cache(cache)
|
|
set_permission_monitor(monitor)
|
|
|
|
study_id = uuid.uuid4()
|
|
configurable_roles = [role for role in PROJECT_PERMISSION_ROLES if role != "PM"]
|
|
payload = {
|
|
role: {
|
|
endpoint_key: index % 2 == role_index % 2
|
|
for index, endpoint_key in enumerate(API_ENDPOINT_PERMISSIONS)
|
|
}
|
|
for role_index, role in enumerate(configurable_roles)
|
|
}
|
|
|
|
matrix = await replace_api_endpoint_permissions(db_session, study_id, payload)
|
|
|
|
for role in configurable_roles:
|
|
for endpoint_key, expected in payload[role].items():
|
|
assert matrix[role][endpoint_key]["allowed"] is expected
|
|
allowed = await role_has_api_permission(
|
|
db_session,
|
|
study_id,
|
|
role,
|
|
endpoint_key,
|
|
check_prerequisites=False,
|
|
)
|
|
assert allowed is expected
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_custom_active_role_permissions_take_effect(db_session: AsyncSession):
|
|
"""验证项目自定义角色配置权限后能被后端鉴权读取。"""
|
|
study_id = uuid.uuid4()
|
|
db_session.add(
|
|
Study(
|
|
id=study_id,
|
|
code=f"CUSTOM-{study_id.hex[:8]}",
|
|
name="自定义角色权限测试",
|
|
status="ACTIVE",
|
|
is_locked=False,
|
|
visit_schedule=[],
|
|
active_roles=["DATA_MANAGER"],
|
|
)
|
|
)
|
|
await db_session.commit()
|
|
|
|
matrix = await get_api_endpoint_permissions(db_session, study_id)
|
|
assert "DATA_MANAGER" in matrix
|
|
assert matrix["DATA_MANAGER"]["subjects:read"]["allowed"] is False
|
|
|
|
await replace_api_endpoint_permissions(
|
|
db_session,
|
|
study_id,
|
|
{"DATA_MANAGER": {"subjects:read": True, "subjects:update": False}},
|
|
)
|
|
|
|
assert await role_has_api_permission(
|
|
db_session,
|
|
study_id,
|
|
"DATA_MANAGER",
|
|
"subjects:read",
|
|
check_prerequisites=False,
|
|
) is True
|
|
assert await role_has_api_permission(
|
|
db_session,
|
|
study_id,
|
|
"DATA_MANAGER",
|
|
"subjects:update",
|
|
check_prerequisites=False,
|
|
) is False
|
|
|
|
|
|
def test_admin_cannot_be_used_as_project_custom_role():
|
|
"""避免项目角色 ADMIN 触发后端系统管理员绕过逻辑。"""
|
|
with pytest.raises(ValueError):
|
|
StudyMemberCreate(user_id=uuid.uuid4(), role_in_study="ADMIN")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_api_permission_check_denied(db_session: AsyncSession):
|
|
"""测试接口级权限检查 - 拒绝"""
|
|
study_id = uuid.uuid4()
|
|
|
|
# 创建权限记录
|
|
perm = ApiEndpointPermission(
|
|
study_id=study_id,
|
|
role="PV",
|
|
endpoint_key="subjects:create",
|
|
allowed=False,
|
|
)
|
|
db_session.add(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角色总是被允许"""
|
|
study_id = uuid.uuid4()
|
|
|
|
result = await role_has_api_permission(
|
|
db_session, study_id, "ADMIN", "POST:/subjects"
|
|
)
|
|
assert result is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_api_permission_read_endpoint(db_session: AsyncSession):
|
|
"""测试读取端点权限"""
|
|
study_id = uuid.uuid4()
|
|
|
|
# 创建读取权限
|
|
perm = ApiEndpointPermission(
|
|
study_id=study_id,
|
|
role="PV",
|
|
endpoint_key="subjects:read",
|
|
allowed=True,
|
|
)
|
|
db_session.add(perm)
|
|
await db_session.commit()
|
|
|
|
# 验证权限（禁用前置权限检查）
|
|
result = await role_has_api_permission(
|
|
db_session, study_id, "PV", "subjects:read", check_prerequisites=False
|
|
)
|
|
assert result is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_api_permission_different_endpoints(db_session: AsyncSession):
|
|
"""测试不同端点的权限独立"""
|
|
study_id = uuid.uuid4()
|
|
|
|
# 创建权限：允许GET，拒绝POST
|
|
get_perm = ApiEndpointPermission(
|
|
study_id=study_id,
|
|
role="CRA",
|
|
endpoint_key="subjects:read",
|
|
allowed=True,
|
|
)
|
|
post_perm = ApiEndpointPermission(
|
|
study_id=study_id,
|
|
role="CRA",
|
|
endpoint_key="subjects:create",
|
|
allowed=False,
|
|
)
|
|
db_session.add(get_perm)
|
|
db_session.add(post_perm)
|
|
await db_session.commit()
|
|
|
|
# 验证权限（禁用前置权限检查）
|
|
get_result = await role_has_api_permission(
|
|
db_session, study_id, "CRA", "subjects:read", check_prerequisites=False
|
|
)
|
|
post_result = await role_has_api_permission(
|
|
db_session, study_id, "CRA", "subjects:create", check_prerequisites=False
|
|
)
|
|
|
|
assert get_result is True
|
|
assert post_result is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_api_permission_different_roles(db_session: AsyncSession):
|
|
"""测试不同角色的权限独立"""
|
|
study_id = uuid.uuid4()
|
|
|
|
# 创建权限：CRA允许，PV拒绝
|
|
cra_perm = ApiEndpointPermission(
|
|
study_id=study_id,
|
|
role="CRA",
|
|
endpoint_key="subjects:create",
|
|
allowed=True,
|
|
)
|
|
pv_perm = ApiEndpointPermission(
|
|
study_id=study_id,
|
|
role="PV",
|
|
endpoint_key="subjects:create",
|
|
allowed=False,
|
|
)
|
|
db_session.add(cra_perm)
|
|
db_session.add(pv_perm)
|
|
await db_session.commit()
|
|
|
|
# 验证权限（禁用前置权限检查）
|
|
cra_result = await role_has_api_permission(
|
|
db_session, study_id, "CRA", "subjects:create", check_prerequisites=False
|
|
)
|
|
pv_result = await role_has_api_permission(
|
|
db_session, study_id, "PV", "subjects:create", check_prerequisites=False
|
|
)
|
|
|
|
assert cra_result is True
|
|
assert pv_result is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_api_permission_different_studies(db_session: AsyncSession):
|
|
"""测试不同项目的权限独立"""
|
|
study_id_1 = uuid.uuid4()
|
|
study_id_2 = uuid.uuid4()
|
|
|
|
# 创建权限：项目1允许，项目2拒绝
|
|
perm_1 = ApiEndpointPermission(
|
|
study_id=study_id_1,
|
|
role="CRA",
|
|
endpoint_key="subjects:create",
|
|
allowed=True,
|
|
)
|
|
perm_2 = ApiEndpointPermission(
|
|
study_id=study_id_2,
|
|
role="CRA",
|
|
endpoint_key="subjects:create",
|
|
allowed=False,
|
|
)
|
|
db_session.add(perm_1)
|
|
db_session.add(perm_2)
|
|
await db_session.commit()
|
|
|
|
# 验证权限（禁用前置权限检查）
|
|
result_1 = await role_has_api_permission(
|
|
db_session, study_id_1, "CRA", "subjects:create", check_prerequisites=False
|
|
)
|
|
result_2 = await role_has_api_permission(
|
|
db_session, study_id_2, "CRA", "subjects:create", check_prerequisites=False
|
|
)
|
|
|
|
assert result_1 is True
|
|
assert result_2 is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_api_permission_none_role(db_session: AsyncSession):
|
|
"""测试None角色的权限检查"""
|
|
study_id = uuid.uuid4()
|
|
|
|
result = await role_has_api_permission(
|
|
db_session, study_id, None, "subjects:create", check_prerequisites=False
|
|
)
|
|
assert result is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_api_permission_unknown_endpoint(db_session: AsyncSession):
|
|
"""测试未知端点的权限检查"""
|
|
study_id = uuid.uuid4()
|
|
|
|
result = await role_has_api_permission(
|
|
db_session, study_id, "CRA", "unknown:endpoint", check_prerequisites=False
|
|
)
|
|
assert result is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_api_permission_check_uses_project_permission_cache(db_session: AsyncSession):
|
|
"""测试接口权限检查会复用项目权限缓存并记录命中指标"""
|
|
cache = PermissionCache()
|
|
monitor = PermissionMonitor()
|
|
set_permission_cache(cache)
|
|
set_permission_monitor(monitor)
|
|
study_id = uuid.uuid4()
|
|
|
|
db_session.add(
|
|
ApiEndpointPermission(
|
|
study_id=study_id,
|
|
role="CRA",
|
|
endpoint_key="subjects:create",
|
|
allowed=True,
|
|
)
|
|
)
|
|
await db_session.commit()
|
|
|
|
first_result = await role_has_api_permission(
|
|
db_session, study_id, "CRA", "subjects:create", check_prerequisites=False
|
|
)
|
|
await db_session.execute(
|
|
delete(ApiEndpointPermission).where(ApiEndpointPermission.study_id == study_id)
|
|
)
|
|
await db_session.commit()
|
|
second_result = await role_has_api_permission(
|
|
db_session, study_id, "CRA", "subjects:create", check_prerequisites=False
|
|
)
|
|
|
|
cache_metrics = monitor.get_metrics()["cache_metrics"]
|
|
assert first_result is True
|
|
assert second_result is True
|
|
assert cache.get_cache_stats()["project_permissions_count"] == 1
|
|
assert cache_metrics["cache_misses"] == 1
|
|
assert cache_metrics["cache_hits"] == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_replace_api_endpoint_permissions_invalidates_project_permission_cache(
|
|
db_session: AsyncSession,
|
|
):
|
|
"""测试替换权限后会失效旧缓存并记录失效指标"""
|
|
from app.core.project_permissions import replace_api_endpoint_permissions
|
|
|
|
cache = PermissionCache()
|
|
monitor = PermissionMonitor()
|
|
set_permission_cache(cache)
|
|
set_permission_monitor(monitor)
|
|
study_id = uuid.uuid4()
|
|
|
|
await replace_api_endpoint_permissions(
|
|
db_session,
|
|
study_id,
|
|
{"CRA": {"subjects:create": True}},
|
|
)
|
|
assert cache.get_cache_stats()["project_permissions_count"] == 1
|
|
|
|
await replace_api_endpoint_permissions(
|
|
db_session,
|
|
study_id,
|
|
{"CRA": {"subjects:create": False}},
|
|
)
|
|
|
|
result = await role_has_api_permission(
|
|
db_session, study_id, "CRA", "subjects:create", check_prerequisites=False
|
|
)
|
|
cache_metrics = monitor.get_metrics()["cache_metrics"]
|
|
assert result is False
|
|
assert cache_metrics["cache_invalidations"] == 2
|