Cheng Zhou
134 lines
4.3 KiB
Python
134 lines
4.3 KiB
Python
import pytest
|
|
import pytest_asyncio
|
|
from httpx import AsyncClient
|
|
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
|
|
|
|
from app.main import create_app
|
|
from app.core.deps import get_db_session
|
|
from app.core.security import hash_password
|
|
from app.crud import user as user_crud
|
|
from app.models.user import User, UserRole, UserStatus
|
|
|
|
TEST_DATABASE_URL = "sqlite+aiosqlite:///:memory:"
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def client_and_db():
|
|
engine = create_async_engine(TEST_DATABASE_URL, future=True)
|
|
TestSessionLocal = async_sessionmaker(engine, expire_on_commit=False)
|
|
async with engine.begin() as conn:
|
|
await conn.run_sync(User.__table__.create)
|
|
|
|
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 with TestSessionLocal() as session:
|
|
admin = User(
|
|
email="admin@test.com",
|
|
password_hash=hash_password("admin123"),
|
|
full_name="Admin",
|
|
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
|
|
async def test_register_creates_pending_user(client_and_db):
|
|
client, SessionLocal = client_and_db
|
|
payload = {
|
|
"email": "newuser@test.com",
|
|
"password": "Password123",
|
|
"full_name": "New User",
|
|
"role": "CRA",
|
|
"department": "Clinical",
|
|
}
|
|
resp = await client.post("/api/v1/auth/register", json=payload)
|
|
assert resp.status_code == 201
|
|
async with SessionLocal() as session:
|
|
user = await user_crud.get_by_email(session, payload["email"])
|
|
assert user is not None
|
|
assert user.status == UserStatus.PENDING
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_root_returns_service_metadata(client_and_db):
|
|
client, _ = client_and_db
|
|
resp = await client.get("/")
|
|
assert resp.status_code == 200
|
|
assert resp.json() == {
|
|
"service": "ctms-backend",
|
|
"status": "ok",
|
|
"health": "/health",
|
|
"docs": "/docs",
|
|
"api": "/api/v1",
|
|
}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_login_blocked_before_approval(client_and_db):
|
|
client, SessionLocal = client_and_db
|
|
payload = {
|
|
"email": "pending@test.com",
|
|
"password": "Password123",
|
|
"full_name": "Pending User",
|
|
"role": "PV",
|
|
"department": "Safety",
|
|
}
|
|
await client.post("/api/v1/auth/register", json=payload)
|
|
resp = await client.post("/api/v1/auth/login", json={"email": payload["email"], "password": payload["password"]})
|
|
assert resp.status_code == 401
|
|
assert "账号未审核" in resp.json().get("detail", "")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_admin_can_approve_user(client_and_db):
|
|
client, SessionLocal = client_and_db
|
|
payload = {
|
|
"email": "approve@test.com",
|
|
"password": "Password123",
|
|
"full_name": "Approve Target",
|
|
"role": "IMP",
|
|
"department": "Supply",
|
|
}
|
|
await client.post("/api/v1/auth/register", json=payload)
|
|
async with SessionLocal() as session:
|
|
user = await user_crud.get_by_email(session, payload["email"])
|
|
user_id = user.id
|
|
|
|
admin_login = await client.post("/api/v1/auth/login", json={"email": "admin@test.com", "password": "admin123"})
|
|
token = admin_login.json()["access_token"]
|
|
headers = {"Authorization": f"Bearer {token}"}
|
|
|
|
resp = await client.post(f"/api/v1/admin/users/{user_id}/approve", json={"action": "approve"}, headers=headers)
|
|
assert resp.status_code == 200
|
|
async with SessionLocal() as session:
|
|
refreshed = await user_crud.get_by_email(session, payload["email"])
|
|
assert refreshed.status == UserStatus.ACTIVE
|
|
assert refreshed.approved_by is not None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_admin_role_cannot_register(client_and_db):
|
|
client, _ = client_and_db
|
|
payload = {
|
|
"email": "admin-register@test.com",
|
|
"password": "Password123",
|
|
"full_name": "Bad Admin",
|
|
"role": "ADMIN",
|
|
"department": "IT",
|
|
}
|
|
resp = await client.post("/api/v1/auth/register", json=payload)
|
|
assert resp.status_code in (400, 422)
|