中文:收口权限与中心/立项配置改造

This commit is contained in:
Cheng Zhou
2026-05-12 10:16:52 +08:00
parent 6e90370a5f
commit 77e842637d
67 changed files with 706 additions and 669 deletions
@@ -0,0 +1,16 @@
from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
def test_project_management_write_routes_allow_project_admin_and_pm():
for relative_path in (
"app/api/v1/studies.py",
"app/api/v1/sites.py",
"app/api/v1/members.py",
):
source = (ROOT / relative_path).read_text(encoding="utf-8")
assert 'require_study_roles(["PM"])' not in source
assert 'require_study_roles(["ADMIN", "PM"])' in source
-107
View File
@@ -1,107 +0,0 @@
import uuid
import pytest
from fastapi import HTTPException
from app.api.v1 import users as users_api
from app.models.user import User, UserRole, UserStatus
from app.schemas.user import UserRead, UserUpdate
def _make_user(
*,
email: str,
role: UserRole = UserRole.ADMIN,
status: UserStatus = UserStatus.ACTIVE,
) -> User:
return User(
id=uuid.uuid4(),
email=email,
password_hash="hashed",
full_name="User",
department="SYSTEM",
role=role,
status=status,
)
@pytest.mark.asyncio
async def test_delete_user_blocks_protected_admin(monkeypatch):
current_user = _make_user(email="other-admin@huapont.cn")
protected_admin = _make_user(email="admin@huapont.cn")
async def fake_get_by_id(_db, _user_id):
return protected_admin
async def fake_count_active_admins(_db):
return 2
async def fake_user_has_memberships(_db, _user_id):
return False
async def fake_delete_user(_db, _user):
raise AssertionError("protected admin should not be deleted")
monkeypatch.setattr(users_api.user_crud, "get_by_id", fake_get_by_id)
monkeypatch.setattr(users_api.user_crud, "count_active_admins", fake_count_active_admins)
monkeypatch.setattr(users_api.member_crud, "user_has_memberships", fake_user_has_memberships)
monkeypatch.setattr(users_api.user_crud, "delete_user", fake_delete_user)
with pytest.raises(HTTPException) as exc_info:
await users_api.delete_user(protected_admin.id, db=object(), current_user=current_user)
assert exc_info.value.status_code == 400
@pytest.mark.asyncio
async def test_update_user_blocks_protected_admin_email_change(monkeypatch):
protected_admin = _make_user(email="admin@huapont.cn")
async def fake_get_by_id(_db, _user_id):
return protected_admin
async def fake_count_active_admins(_db):
return 2
async def fake_update_user(_db, _user, _payload):
raise AssertionError("protected admin email should not be changed")
monkeypatch.setattr(users_api.user_crud, "get_by_id", fake_get_by_id)
monkeypatch.setattr(users_api.user_crud, "count_active_admins", fake_count_active_admins)
monkeypatch.setattr(users_api.user_crud, "update_user", fake_update_user)
with pytest.raises(HTTPException) as exc_info:
await users_api.update_user(
protected_admin.id,
UserUpdate(email="renamed@huapont.cn"),
db=object(),
current_user=protected_admin,
)
assert exc_info.value.status_code == 400
@pytest.mark.asyncio
async def test_update_user_allows_protected_admin_password_change(monkeypatch):
protected_admin = _make_user(email="admin@huapont.cn")
captured_payload = {}
async def fake_get_by_id(_db, _user_id):
return protected_admin
async def fake_update_user(_db, _user, payload):
captured_payload["password"] = payload.password
return _user
monkeypatch.setattr(users_api.user_crud, "get_by_id", fake_get_by_id)
monkeypatch.setattr(users_api.user_crud, "update_user", fake_update_user)
result = await users_api.update_user(
protected_admin.id,
UserUpdate(password="Password123"),
db=object(),
current_user=protected_admin,
)
assert isinstance(result, UserRead | User)
assert captured_payload["password"] == "Password123"
+25
View File
@@ -0,0 +1,25 @@
from dataclasses import dataclass
from app.core import rbac
@dataclass
class UserStub:
role: str
@dataclass
class MemberStub:
role_in_study: str
def test_non_admin_global_role_does_not_grant_project_document_permission():
assert not rbac.is_allowed("delete_document", UserStub(role="PM"), MemberStub(role_in_study="CRA"))
def test_project_role_grants_project_document_permission():
assert rbac.is_allowed("delete_document", UserStub(role="CRA"), MemberStub(role_in_study="PM"))
def test_global_admin_keeps_system_document_permission_without_membership():
assert rbac.is_allowed("delete_document", UserStub(role="ADMIN"), None)
+5 -5
View File
@@ -88,7 +88,7 @@ async def client_and_db():
email="admin@test.com",
password_hash=hash_password("admin123"),
full_name="Admin",
department="Admin",
clinical_department="Admin",
role=UserRole.ADMIN,
status=UserStatus.ACTIVE,
)
@@ -109,7 +109,7 @@ async def test_register_creates_pending_user(client_and_db):
"password": "Password123",
"full_name": "New User",
"role": "CRA",
"department": "Clinical",
"clinical_department": "Clinical",
}
resp = await client.post("/api/v1/auth/register", json=payload)
assert resp.status_code == 201
@@ -141,7 +141,7 @@ async def test_login_blocked_before_approval(client_and_db):
"password": "Password123",
"full_name": "Pending User",
"role": "PV",
"department": "Safety",
"clinical_department": "Safety",
}
await client.post("/api/v1/auth/register", json=payload)
resp = await encrypted_login(client, payload["email"], payload["password"])
@@ -157,7 +157,7 @@ async def test_admin_can_approve_user(client_and_db):
"password": "Password123",
"full_name": "Approve Target",
"role": "IMP",
"department": "Supply",
"clinical_department": "Supply",
}
await client.post("/api/v1/auth/register", json=payload)
async with SessionLocal() as session:
@@ -184,7 +184,7 @@ async def test_admin_role_cannot_register(client_and_db):
"password": "Password123",
"full_name": "Bad Admin",
"role": "ADMIN",
"department": "IT",
"clinical_department": "IT",
}
resp = await client.post("/api/v1/auth/register", json=payload)
assert resp.status_code in (400, 422)
+72
View File
@@ -0,0 +1,72 @@
import uuid
import pytest
from app.api.v1 import studies as studies_api
from app.models.study import Study
from app.models.study_member import StudyMember
from app.models.user import User, UserRole, UserStatus
def _make_user(role: UserRole = UserRole.CRA) -> User:
return User(
id=uuid.uuid4(),
email="cra-pm@test.com",
password_hash="hashed",
full_name="CRA PM",
clinical_department="Clinical",
role=role,
status=UserStatus.ACTIVE,
)
def _make_study() -> Study:
return Study(
id=uuid.uuid4(),
code="STUDY-001",
name="测试项目",
status="ACTIVE",
is_locked=False,
visit_schedule=[],
)
@pytest.mark.asyncio
async def test_list_studies_returns_project_role_for_non_admin(monkeypatch):
current_user = _make_user(UserRole.CRA)
study = _make_study()
member = StudyMember(
id=uuid.uuid4(),
study_id=study.id,
user_id=current_user.id,
role_in_study="PM",
is_active=True,
)
async def fake_list_studies_for_user(_db, _user_id, skip=0, limit=100):
return [study]
async def fake_get_member(_db, _study_id, _user_id):
return member
monkeypatch.setattr(studies_api.study_crud, "list_studies_for_user", fake_list_studies_for_user)
monkeypatch.setattr(studies_api.member_crud, "get_member", fake_get_member)
result = await studies_api.list_studies(db=object(), current_user=current_user)
assert result["items"][0].role_in_study == "PM"
@pytest.mark.asyncio
async def test_list_studies_does_not_map_system_admin_to_project_admin(monkeypatch):
current_user = _make_user(UserRole.ADMIN)
study = _make_study()
async def fake_list_studies(_db, skip=0, limit=100):
return [study]
monkeypatch.setattr(studies_api.study_crud, "list_studies", fake_list_studies)
result = await studies_api.list_studies(db=object(), current_user=current_user)
assert result["items"][0].role_in_study is None