53 lines
1.5 KiB
Python
53 lines
1.5 KiB
Python
import uuid
|
|
from collections.abc import AsyncIterator
|
|
|
|
import pytest
|
|
from httpx import ASGITransport, AsyncClient
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.core.security import hash_password
|
|
from app.db.session import async_session_maker, engine
|
|
from app.main import app
|
|
from app.models.user import User
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
async def _dispose_engine_after_test() -> AsyncIterator[None]:
|
|
# Each test runs in its own event loop; pooled asyncpg connections from a
|
|
# previous test's loop are unusable in the next one, so drop the pool.
|
|
yield
|
|
await engine.dispose()
|
|
|
|
|
|
@pytest.fixture
|
|
async def db() -> AsyncIterator[AsyncSession]:
|
|
async with async_session_maker() as session:
|
|
yield session
|
|
|
|
|
|
@pytest.fixture
|
|
async def client() -> AsyncIterator[AsyncClient]:
|
|
transport = ASGITransport(app=app)
|
|
async with AsyncClient(transport=transport, base_url="http://test") as ac:
|
|
yield ac
|
|
|
|
|
|
@pytest.fixture
|
|
async def admin_user(db: AsyncSession) -> AsyncIterator[User]:
|
|
user = User(username=f"admin-{uuid.uuid4().hex[:8]}", hashed_password=hash_password("adminpass123"))
|
|
db.add(user)
|
|
await db.commit()
|
|
await db.refresh(user)
|
|
yield user
|
|
await db.delete(user)
|
|
await db.commit()
|
|
|
|
|
|
@pytest.fixture
|
|
async def auth_headers(client: AsyncClient, admin_user: User) -> dict[str, str]:
|
|
res = await client.post(
|
|
"/api/v1/auth/token", data={"username": admin_user.username, "password": "adminpass123"}
|
|
)
|
|
token = res.json()["access_token"]
|
|
return {"Authorization": f"Bearer {token}"}
|