Cheng Zhou
ab4f0d93ed
1、合并列表与详情读取权限，统一使用 subjects:read、project_members:read、monitoring_issues:read 等业务读取 key。 2、增加历史读取权限 key 的兼容映射，确保存量项目权限覆盖值不会丢失。 3、同步前端权限工具、路由权限与权限管理页面，避免继续引用已合并的旧权限 key。 4、补充项目角色启停保护与抽屉未保存变更守卫，防止停用 PM 或仍有关联成员的角色。
137 lines
4.2 KiB
Python
137 lines
4.2 KiB
Python
import uuid
|
|
from dataclasses import dataclass
|
|
|
|
import pytest
|
|
from fastapi import HTTPException
|
|
from sqlalchemy import text
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.api.v1.study_active_roles import _normalize_active_roles, update_active_roles
|
|
|
|
|
|
@dataclass
|
|
class UserStub:
|
|
id: uuid.UUID
|
|
is_admin: bool = False
|
|
|
|
|
|
def test_normalize_active_roles_allows_renamed_qa_project_role():
|
|
assert _normalize_active_roles(["PM", "QA", "CTA", "QA", "", None]) == ["PM", "QA", "CTA"]
|
|
|
|
|
|
def test_normalize_active_roles_still_rejects_system_admin_role():
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
_normalize_active_roles(["ADMIN"])
|
|
|
|
assert exc_info.value.status_code == 422
|
|
assert exc_info.value.detail == "ADMIN 不能作为项目角色"
|
|
|
|
|
|
def test_project_pm_update_keeps_pm_active_role():
|
|
assert _normalize_active_roles(["PV", "CRA"], preserve_pm=True) == ["PM", "PV", "CRA"]
|
|
|
|
|
|
def test_system_admin_update_can_remove_pm_active_role():
|
|
assert _normalize_active_roles(["PV", "CRA"], preserve_pm=False) == ["PV", "CRA"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_project_pm_cannot_remove_pm_from_active_roles(db_session: AsyncSession):
|
|
user_id = uuid.uuid4()
|
|
study_id = uuid.uuid4()
|
|
await db_session.execute(
|
|
text(
|
|
"""
|
|
INSERT INTO studies (id, code, name, status, is_locked, visit_schedule, active_roles)
|
|
VALUES (:id, :code, :name, :status, :is_locked, :visit_schedule, :active_roles)
|
|
"""
|
|
),
|
|
{
|
|
"id": str(study_id),
|
|
"code": f"PROJECT-PM-ACTIVE-ROLE-{study_id.hex[:8]}",
|
|
"name": "PROJECT-PM-ACTIVE-ROLE",
|
|
"status": "ACTIVE",
|
|
"is_locked": False,
|
|
"visit_schedule": "[]",
|
|
"active_roles": '["PM","PV","CRA"]',
|
|
},
|
|
)
|
|
await db_session.commit()
|
|
|
|
result = await update_active_roles(
|
|
study_id=study_id,
|
|
payload={"active_roles": ["PV", "CRA"]},
|
|
current_user=UserStub(id=user_id),
|
|
db=db_session,
|
|
)
|
|
|
|
assert result["active_roles"] == ["PM", "PV", "CRA"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cannot_remove_active_role_assigned_to_active_member(db_session: AsyncSession):
|
|
user_id = uuid.uuid4()
|
|
member_id = uuid.uuid4()
|
|
study_id = uuid.uuid4()
|
|
await db_session.execute(
|
|
text(
|
|
"""
|
|
INSERT INTO studies (id, code, name, status, is_locked, visit_schedule, active_roles)
|
|
VALUES (:id, :code, :name, :status, :is_locked, :visit_schedule, :active_roles)
|
|
"""
|
|
),
|
|
{
|
|
"id": str(study_id),
|
|
"code": f"PROJECT-ROLE-IN-USE-{study_id.hex[:8]}",
|
|
"name": "PROJECT-ROLE-IN-USE",
|
|
"status": "ACTIVE",
|
|
"is_locked": False,
|
|
"visit_schedule": "[]",
|
|
"active_roles": '["PM","PV","CRA"]',
|
|
},
|
|
)
|
|
await db_session.execute(
|
|
text(
|
|
"""
|
|
INSERT INTO users (id, email, password_hash, full_name, clinical_department, is_admin, status)
|
|
VALUES (:id, :email, :password_hash, :full_name, :clinical_department, :is_admin, :status)
|
|
"""
|
|
),
|
|
{
|
|
"id": str(user_id),
|
|
"email": f"{user_id.hex}@example.com",
|
|
"password_hash": "hash",
|
|
"full_name": "PV User",
|
|
"clinical_department": "Clinical",
|
|
"is_admin": False,
|
|
"status": "ACTIVE",
|
|
},
|
|
)
|
|
await db_session.execute(
|
|
text(
|
|
"""
|
|
INSERT INTO study_members (id, study_id, user_id, role_in_study, is_active)
|
|
VALUES (:id, :study_id, :user_id, :role, :active)
|
|
"""
|
|
),
|
|
{
|
|
"id": str(member_id),
|
|
"study_id": str(study_id),
|
|
"user_id": str(user_id),
|
|
"role": "PV",
|
|
"active": True,
|
|
},
|
|
)
|
|
await db_session.commit()
|
|
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
await update_active_roles(
|
|
study_id=study_id,
|
|
payload={"active_roles": ["CRA"]},
|
|
current_user=UserStub(id=uuid.uuid4(), is_admin=True),
|
|
db=db_session,
|
|
)
|
|
|
|
assert exc_info.value.status_code == 409
|
|
assert exc_info.value.detail == "角色 PV 仍有成员使用，不能停用"
|