Cheng Zhou
387 lines
13 KiB
Python
387 lines
13 KiB
Python
import uuid
|
|
import json
|
|
from dataclasses import dataclass
|
|
|
|
import pytest
|
|
from fastapi import HTTPException
|
|
from sqlalchemy import text
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.api.v1.api_permissions import get_my_study_api_permissions, update_study_api_permissions
|
|
from app.api.v1.members import add_member, update_member
|
|
from app.api.v1.permission_monitoring import resolve_monitoring_scope
|
|
from app.api.v1.system_permissions import list_system_permissions
|
|
from app.core.deps import require_admin_or_any_project_pm, require_system_permission
|
|
from app.schemas.member import StudyMemberCreate, StudyMemberUpdate
|
|
|
|
|
|
@dataclass
|
|
class UserStub:
|
|
id: uuid.UUID
|
|
is_admin: bool = False
|
|
|
|
|
|
async def _seed_user(db: AsyncSession, user_id: uuid.UUID, *, is_admin: bool = False) -> None:
|
|
await db.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": f"User {user_id.hex[:6]}",
|
|
"clinical_department": "Clinical",
|
|
"is_admin": is_admin,
|
|
"status": "ACTIVE",
|
|
},
|
|
)
|
|
|
|
|
|
async def _seed_named_user(db: AsyncSession, user_id: uuid.UUID, full_name: str, *, is_admin: bool = False) -> None:
|
|
await db.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": full_name,
|
|
"clinical_department": "Clinical",
|
|
"is_admin": is_admin,
|
|
"status": "ACTIVE",
|
|
},
|
|
)
|
|
|
|
|
|
async def _seed_study(db: AsyncSession, study_id: uuid.UUID, code: str) -> None:
|
|
await db.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": code,
|
|
"name": code,
|
|
"status": "ACTIVE",
|
|
"is_locked": False,
|
|
"visit_schedule": "[]",
|
|
"active_roles": "[]",
|
|
},
|
|
)
|
|
|
|
|
|
async def _seed_member(db: AsyncSession, study_id: uuid.UUID, user_id: uuid.UUID, role: str, active: bool = True) -> None:
|
|
await db.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(uuid.uuid4()),
|
|
"study_id": str(study_id),
|
|
"user_id": str(user_id),
|
|
"role": role,
|
|
"active": active,
|
|
},
|
|
)
|
|
|
|
|
|
async def _seed_member_return_id(
|
|
db: AsyncSession,
|
|
study_id: uuid.UUID,
|
|
user_id: uuid.UUID,
|
|
role: str,
|
|
active: bool = True,
|
|
) -> uuid.UUID:
|
|
member_id = uuid.uuid4()
|
|
await db.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": role,
|
|
"active": active,
|
|
},
|
|
)
|
|
return member_id
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_project_pm_can_view_system_permission_definitions(db_session: AsyncSession):
|
|
pm_id = uuid.uuid4()
|
|
study_id = uuid.uuid4()
|
|
await _seed_user(db_session, pm_id)
|
|
await _seed_study(db_session, study_id, "PM-SYSTEM-PERMS")
|
|
await _seed_member(db_session, study_id, pm_id, "PM")
|
|
await db_session.commit()
|
|
|
|
dependency = require_admin_or_any_project_pm()
|
|
await dependency(current_user=UserStub(id=pm_id), db=db_session)
|
|
data = await list_system_permissions()
|
|
|
|
assert data["permissions"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_non_pm_cannot_view_system_permission_definitions(db_session: AsyncSession):
|
|
cra_id = uuid.uuid4()
|
|
study_id = uuid.uuid4()
|
|
await _seed_user(db_session, cra_id)
|
|
await _seed_study(db_session, study_id, "CRA-SYSTEM-PERMS")
|
|
await _seed_member(db_session, study_id, cra_id, "CRA")
|
|
await db_session.commit()
|
|
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
dependency = require_admin_or_any_project_pm()
|
|
await dependency(current_user=UserStub(id=cra_id), db=db_session)
|
|
|
|
assert exc_info.value.status_code == 403
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_project_pm_can_open_project_permission_config_without_global_pm_role(db_session: AsyncSession):
|
|
user_id = uuid.uuid4()
|
|
study_id = uuid.uuid4()
|
|
await _seed_user(db_session, user_id)
|
|
await _seed_study(db_session, study_id, "PROJECT-PM-NO-GLOBAL-ROLE")
|
|
await _seed_member(db_session, study_id, user_id, "PM")
|
|
await db_session.commit()
|
|
|
|
dependency = require_system_permission("system:permissions:project_config")
|
|
result = await dependency(study_id=study_id, current_user=UserStub(id=user_id), db=db_session)
|
|
|
|
assert result.id == user_id
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_non_project_pm_cannot_open_project_permission_config_even_with_active_membership(db_session: AsyncSession):
|
|
user_id = uuid.uuid4()
|
|
study_id = uuid.uuid4()
|
|
await _seed_user(db_session, user_id)
|
|
await _seed_study(db_session, study_id, "PROJECT-NON-PM-DENIED")
|
|
await _seed_member(db_session, study_id, user_id, "CRA")
|
|
await db_session.commit()
|
|
|
|
dependency = require_system_permission("system:permissions:project_config")
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
await dependency(study_id=study_id, current_user=UserStub(id=user_id), db=db_session)
|
|
|
|
assert exc_info.value.status_code == 403
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_pm_permission_update_does_not_persist_admin_or_pm_overrides(db_session: AsyncSession):
|
|
study_id = uuid.uuid4()
|
|
admin_id = uuid.uuid4()
|
|
await _seed_study(db_session, study_id, "PM-PERM-SKIP")
|
|
await _seed_user(db_session, admin_id, is_admin=True)
|
|
await db_session.commit()
|
|
|
|
result = await update_study_api_permissions(
|
|
study_id=study_id,
|
|
payload={
|
|
"ADMIN": {"subjects:delete": False},
|
|
"PM": {"subjects:delete": False},
|
|
"CRA": {"subjects:delete": True},
|
|
},
|
|
current_user=UserStub(id=admin_id, is_admin=True),
|
|
db=db_session,
|
|
)
|
|
|
|
rows = (
|
|
await db_session.execute(
|
|
text(
|
|
"""
|
|
SELECT role, endpoint_key, allowed
|
|
FROM api_endpoint_permissions
|
|
WHERE study_id = :study_id
|
|
"""
|
|
),
|
|
{"study_id": str(study_id)},
|
|
)
|
|
).all()
|
|
assert ("CRA", "subjects:delete", True) in rows
|
|
assert all(row.role not in {"ADMIN", "PM"} for row in rows)
|
|
assert "ADMIN" not in result
|
|
assert result["PM"]["subjects:delete"]["allowed"] is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_project_member_can_read_own_effective_permissions(db_session: AsyncSession):
|
|
cra_id = uuid.uuid4()
|
|
study_id = uuid.uuid4()
|
|
await _seed_user(db_session, cra_id)
|
|
await _seed_study(db_session, study_id, "CRA-MY-PERMS")
|
|
await _seed_member(db_session, study_id, cra_id, "CRA")
|
|
await db_session.commit()
|
|
|
|
result = await get_my_study_api_permissions(
|
|
study_id=study_id,
|
|
current_user=UserStub(id=cra_id),
|
|
db=db_session,
|
|
)
|
|
|
|
assert list(result) == ["CRA"]
|
|
assert result["CRA"]["sites:read"]["allowed"] is True
|
|
assert result["CRA"]["sites:update"]["allowed"] is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_project_pm_monitoring_scope_is_limited_to_own_projects(db_session: AsyncSession):
|
|
pm_id = uuid.uuid4()
|
|
own_study_id = uuid.uuid4()
|
|
other_study_id = uuid.uuid4()
|
|
await _seed_user(db_session, pm_id)
|
|
await _seed_study(db_session, own_study_id, "PM-MONITOR-OWN")
|
|
await _seed_study(db_session, other_study_id, "PM-MONITOR-OTHER")
|
|
await _seed_member(db_session, own_study_id, pm_id, "PM")
|
|
await db_session.commit()
|
|
|
|
scope = await resolve_monitoring_scope(db_session, UserStub(id=pm_id))
|
|
|
|
assert scope.is_admin is False
|
|
assert scope.study_ids == {own_study_id}
|
|
assert scope.can_access_study(own_study_id)
|
|
assert not scope.can_access_study(other_study_id)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_project_pm_cannot_update_peer_pm_member(db_session: AsyncSession):
|
|
study_id = uuid.uuid4()
|
|
actor_id = uuid.uuid4()
|
|
peer_id = uuid.uuid4()
|
|
await _seed_study(db_session, study_id, "PM-PEER-MEMBER")
|
|
await _seed_user(db_session, actor_id)
|
|
await _seed_user(db_session, peer_id)
|
|
await _seed_member(db_session, study_id, actor_id, "PM")
|
|
peer_member_id = await _seed_member_return_id(db_session, study_id, peer_id, "PM")
|
|
await db_session.commit()
|
|
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
await update_member(
|
|
study_id=study_id,
|
|
member_id=peer_member_id,
|
|
member_in=StudyMemberUpdate(role_in_study="CRA"),
|
|
current_user=UserStub(id=actor_id),
|
|
db=db_session,
|
|
)
|
|
|
|
assert exc_info.value.status_code == 403
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_project_pm_cannot_grant_peer_pm_role(db_session: AsyncSession):
|
|
study_id = uuid.uuid4()
|
|
actor_id = uuid.uuid4()
|
|
cra_id = uuid.uuid4()
|
|
await _seed_study(db_session, study_id, "PM-GRANT-PM")
|
|
await _seed_user(db_session, actor_id)
|
|
await _seed_user(db_session, cra_id)
|
|
await _seed_member(db_session, study_id, actor_id, "PM")
|
|
cra_member_id = await _seed_member_return_id(db_session, study_id, cra_id, "CRA")
|
|
await db_session.commit()
|
|
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
await update_member(
|
|
study_id=study_id,
|
|
member_id=cra_member_id,
|
|
member_in=StudyMemberUpdate(role_in_study="PM"),
|
|
current_user=UserStub(id=actor_id),
|
|
db=db_session,
|
|
)
|
|
|
|
assert exc_info.value.status_code == 403
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_project_member_add_audit_uses_member_display_name(db_session: AsyncSession):
|
|
study_id = uuid.uuid4()
|
|
admin_id = uuid.uuid4()
|
|
target_user_id = uuid.uuid4()
|
|
await _seed_study(db_session, study_id, "MEMBER-AUDIT-NAME")
|
|
await _seed_user(db_session, admin_id, is_admin=True)
|
|
await _seed_named_user(db_session, target_user_id, "张三")
|
|
await db_session.commit()
|
|
|
|
await add_member(
|
|
study_id=study_id,
|
|
member_in=StudyMemberCreate(user_id=target_user_id, role_in_study="CRA"),
|
|
current_user=UserStub(id=admin_id, is_admin=True),
|
|
db=db_session,
|
|
)
|
|
|
|
row = (
|
|
await db_session.execute(
|
|
text(
|
|
"""
|
|
SELECT detail
|
|
FROM audit_logs
|
|
WHERE study_id = :study_id AND action = 'PROJECT_MEMBER_ADDED'
|
|
ORDER BY created_at DESC
|
|
LIMIT 1
|
|
"""
|
|
),
|
|
{"study_id": str(study_id)},
|
|
)
|
|
).one()
|
|
detail = json.loads(row.detail)
|
|
assert detail["targetName"] == "张三"
|
|
assert str(target_user_id) not in detail["targetName"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_project_member_update_audit_uses_member_display_name(db_session: AsyncSession):
|
|
study_id = uuid.uuid4()
|
|
admin_id = uuid.uuid4()
|
|
target_user_id = uuid.uuid4()
|
|
await _seed_study(db_session, study_id, "MEMBER-UPDATE-AUDIT-NAME")
|
|
await _seed_user(db_session, admin_id, is_admin=True)
|
|
await _seed_named_user(db_session, target_user_id, "李四")
|
|
member_id = await _seed_member_return_id(db_session, study_id, target_user_id, "CRA")
|
|
await db_session.commit()
|
|
|
|
await update_member(
|
|
study_id=study_id,
|
|
member_id=member_id,
|
|
member_in=StudyMemberUpdate(role_in_study="CTA"),
|
|
current_user=UserStub(id=admin_id, is_admin=True),
|
|
db=db_session,
|
|
)
|
|
|
|
row = (
|
|
await db_session.execute(
|
|
text(
|
|
"""
|
|
SELECT detail
|
|
FROM audit_logs
|
|
WHERE study_id = :study_id AND action = 'PROJECT_MEMBER_UPDATED'
|
|
ORDER BY created_at DESC
|
|
LIMIT 1
|
|
"""
|
|
),
|
|
{"study_id": str(study_id)},
|
|
)
|
|
).one()
|
|
detail = json.loads(row.detail)
|
|
assert detail["targetName"] == "李四"
|
|
assert str(target_user_id) not in detail["targetName"]
|