Files
securelens-backend/tests/conftest.py

72 lines
1.9 KiB
Python

import pytest
import pytest_asyncio
from httpx import ASGITransport, AsyncClient
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from app.database import Base, get_db
from app.main import app
from app.models.user import User
from app.utils.auth import create_access_token, hash_password
TEST_DB_URL = "sqlite+aiosqlite://"
test_engine = create_async_engine(TEST_DB_URL, echo=False)
TestSessionLocal = async_sessionmaker(
bind=test_engine, class_=AsyncSession, expire_on_commit=False
)
async def override_get_db():
async with TestSessionLocal() as session:
try:
yield session
await session.commit()
except Exception:
await session.rollback()
raise
app.dependency_overrides[get_db] = override_get_db
@pytest_asyncio.fixture(autouse=True)
async def setup_db():
async with test_engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
yield
async with test_engine.begin() as conn:
await conn.run_sync(Base.metadata.drop_all)
@pytest.fixture
def client():
from fastapi.testclient import TestClient
return TestClient(app)
@pytest_asyncio.fixture
async def async_client():
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://test") as ac:
yield ac
@pytest_asyncio.fixture
async def test_user():
async with TestSessionLocal() as session:
user = User(
email="test@example.com",
username="testuser",
hashed_password=hash_password("testpassword123"),
)
session.add(user)
await session.commit()
await session.refresh(user)
return user
@pytest_asyncio.fixture
async def auth_headers(test_user):
token = create_access_token(test_user.id)
return {"Authorization": f"Bearer {token}"}