test(register): 校验注册请求不再接收角色字段

- test_registration 改用全模型 metadata 创建表并注入自定义 GUID 类型,
  以便 StudyMember 等关联表在 SQLite 测试库中正常工作。
- 注册请求负载移除 role 字段,新增断言验证用户默认角色为 PV 且不再
  生成 study_members 关联;新增 schema 级用例确认 role 不在
  UserRegisterRequest 模型字段中。

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Cheng Zhou
2026-05-25 12:37:26 +08:00
parent a2add43fa5
commit dd2973c429
+48 -45
View File
@@ -4,9 +4,9 @@ from cryptography.hazmat.primitives.ciphers.aead import AESGCM
from cryptography.hazmat.primitives import hashes, serialization from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import padding from cryptography.hazmat.primitives.asymmetric import padding
from httpx import AsyncClient from httpx import AsyncClient
from sqlalchemy.dialects.postgresql import UUID from sqlalchemy import UUID as SA_UUID, select
from sqlalchemy.dialects.postgresql import UUID as PG_UUID
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
from sqlalchemy.ext.compiler import compiles
import base64 import base64
import json import json
import os import os
@@ -15,16 +15,15 @@ from app.main import create_app
from app.core.deps import get_db_session from app.core.deps import get_db_session
from app.core.security import hash_password from app.core.security import hash_password
from app.crud import user as user_crud from app.crud import user as user_crud
from app.db.base_class import Base
from tests.conftest import GUID
from app.models.study_member import StudyMember
from app.models.user import User, UserRole, UserStatus from app.models.user import User, UserRole, UserStatus
from app.schemas.user import UserRegisterRequest
TEST_DATABASE_URL = "sqlite+aiosqlite:///:memory:" TEST_DATABASE_URL = "sqlite+aiosqlite:///:memory:"
@compiles(UUID, "sqlite")
def compile_uuid_for_sqlite(_type, _compiler, **_kw):
return "CHAR(32)"
async def encrypted_auth_payload(client: AsyncClient, email: str, password: str) -> dict: async def encrypted_auth_payload(client: AsyncClient, email: str, password: str) -> dict:
key_resp = await client.get("/api/v1/auth/login-key") key_resp = await client.get("/api/v1/auth/login-key")
assert key_resp.status_code == 200 assert key_resp.status_code == 200
@@ -73,32 +72,44 @@ async def encrypted_login(client: AsyncClient, email: str, password: str):
async def client_and_db(): async def client_and_db():
engine = create_async_engine(TEST_DATABASE_URL, future=True) engine = create_async_engine(TEST_DATABASE_URL, future=True)
TestSessionLocal = async_sessionmaker(engine, expire_on_commit=False) TestSessionLocal = async_sessionmaker(engine, expire_on_commit=False)
async with engine.begin() as conn: original_column_types = {
await conn.run_sync(User.__table__.create) column: column.type
for table in Base.metadata.tables.values()
for column in table.columns
}
for table in Base.metadata.tables.values():
for column in table.columns:
if isinstance(column.type, (PG_UUID, SA_UUID)):
column.type = GUID()
try:
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
async def override_get_db():
async with TestSessionLocal() as session:
yield session
app = create_app()
app.dependency_overrides[get_db_session] = override_get_db
async def override_get_db():
async with TestSessionLocal() as session: async with TestSessionLocal() as session:
yield session admin = User(
email="admin@test.com",
password_hash=hash_password("admin123"),
full_name="Admin",
clinical_department="Admin",
role=UserRole.ADMIN,
status=UserStatus.ACTIVE,
)
session.add(admin)
await session.commit()
app = create_app() async with AsyncClient(app=app, base_url="http://test") as client:
app.dependency_overrides[get_db_session] = override_get_db yield client, TestSessionLocal
finally:
async with TestSessionLocal() as session: for column, original_type in original_column_types.items():
admin = User( column.type = original_type
email="admin@test.com", await engine.dispose()
password_hash=hash_password("admin123"),
full_name="Admin",
clinical_department="Admin",
role=UserRole.ADMIN,
status=UserStatus.ACTIVE,
)
session.add(admin)
await session.commit()
async with AsyncClient(app=app, base_url="http://test") as client:
yield client, TestSessionLocal
await engine.dispose()
@pytest.mark.asyncio @pytest.mark.asyncio
@@ -108,7 +119,6 @@ async def test_register_creates_pending_user(client_and_db):
"email": "newuser@test.com", "email": "newuser@test.com",
"password": "Password123", "password": "Password123",
"full_name": "New User", "full_name": "New User",
"role": "CRA",
"clinical_department": "Clinical", "clinical_department": "Clinical",
} }
resp = await client.post("/api/v1/auth/register", json=payload) resp = await client.post("/api/v1/auth/register", json=payload)
@@ -117,6 +127,11 @@ async def test_register_creates_pending_user(client_and_db):
user = await user_crud.get_by_email(session, payload["email"]) user = await user_crud.get_by_email(session, payload["email"])
assert user is not None assert user is not None
assert user.status == UserStatus.PENDING assert user.status == UserStatus.PENDING
assert user.role == UserRole.PV
member_rows = (
await session.execute(select(StudyMember).where(StudyMember.user_id == user.id))
).scalars().all()
assert member_rows == []
@pytest.mark.asyncio @pytest.mark.asyncio
@@ -140,7 +155,6 @@ async def test_login_blocked_before_approval(client_and_db):
"email": "pending@test.com", "email": "pending@test.com",
"password": "Password123", "password": "Password123",
"full_name": "Pending User", "full_name": "Pending User",
"role": "PV",
"clinical_department": "Safety", "clinical_department": "Safety",
} }
await client.post("/api/v1/auth/register", json=payload) await client.post("/api/v1/auth/register", json=payload)
@@ -156,7 +170,6 @@ async def test_admin_can_approve_user(client_and_db):
"email": "approve@test.com", "email": "approve@test.com",
"password": "Password123", "password": "Password123",
"full_name": "Approve Target", "full_name": "Approve Target",
"role": "IMP",
"clinical_department": "Supply", "clinical_department": "Supply",
} }
await client.post("/api/v1/auth/register", json=payload) await client.post("/api/v1/auth/register", json=payload)
@@ -176,18 +189,8 @@ async def test_admin_can_approve_user(client_and_db):
assert refreshed.approved_by is not None assert refreshed.approved_by is not None
@pytest.mark.asyncio def test_register_request_does_not_expose_role_input():
async def test_admin_role_cannot_register(client_and_db): assert "role" not in UserRegisterRequest.model_fields
client, _ = client_and_db
payload = {
"email": "admin-register@test.com",
"password": "Password123",
"full_name": "Bad Admin",
"role": "ADMIN",
"clinical_department": "IT",
}
resp = await client.post("/api/v1/auth/register", json=payload)
assert resp.status_code in (400, 422)
@pytest.mark.asyncio @pytest.mark.asyncio