106 lines
3.1 KiB
Python
106 lines
3.1 KiB
Python
import asyncio
|
|
import os
|
|
from pathlib import Path
|
|
from typing import AsyncGenerator
|
|
|
|
import jwt
|
|
import pytest
|
|
from dotenv import load_dotenv
|
|
from httpx import ASGITransport, AsyncClient
|
|
from sqlalchemy import StaticPool
|
|
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker
|
|
|
|
from backend.session import get_session
|
|
from main import app
|
|
from models import BaseModel
|
|
from tests.fixture_loader import FixtureLoader
|
|
|
|
project_root = Path(__file__).parent.parent
|
|
env_path = project_root / ".env"
|
|
load_dotenv(env_path)
|
|
|
|
SECRET_KEY = os.getenv("SECRET_KEY")
|
|
PG_LOGIN = os.environ.get("PG_LOGIN")
|
|
PG_PASSWORD = os.environ.get("PG_PASSWORD")
|
|
TEST_DATABASE_URL = f"postgresql+asyncpg://{PG_LOGIN}:{PG_PASSWORD}@127.0.0.1/test"
|
|
TEST_DATABASE_URL_FOR_ASYNCPG = f"postgresql://{PG_LOGIN}:{PG_PASSWORD}@127.0.0.1/test"
|
|
|
|
|
|
# -------------------------------------------------------------------
|
|
# Create test database + connection pool
|
|
# -------------------------------------------------------------------
|
|
|
|
@pytest.fixture(scope="function")
|
|
def event_loop():
|
|
"""Force pytest to use a session-scoped event loop (needed for asyncpg)."""
|
|
loop = asyncio.get_event_loop_policy().new_event_loop()
|
|
yield loop
|
|
loop.close()
|
|
|
|
|
|
@pytest.fixture(scope="function")
|
|
async def db_session() -> AsyncGenerator[AsyncSession, None]:
|
|
# Create test engine
|
|
test_engine = create_async_engine(
|
|
TEST_DATABASE_URL,
|
|
poolclass=StaticPool, # Useful for tests
|
|
)
|
|
|
|
# Test session factory
|
|
TestAsyncSessionLocal = async_sessionmaker(
|
|
test_engine,
|
|
class_=AsyncSession,
|
|
expire_on_commit=False,
|
|
)
|
|
|
|
async with test_engine.begin() as conn:
|
|
await conn.run_sync(BaseModel.metadata.drop_all)
|
|
await conn.run_sync(BaseModel.metadata.create_all)
|
|
|
|
session = TestAsyncSessionLocal()
|
|
|
|
await FixtureLoader().load_fixtures(session)
|
|
|
|
try:
|
|
yield session
|
|
finally:
|
|
await session.close()
|
|
async with test_engine.begin() as conn:
|
|
await conn.run_sync(BaseModel.metadata.drop_all)
|
|
await test_engine.dispose()
|
|
|
|
|
|
# -------------------------------------------------------------------
|
|
# HTTPX AsyncClient for FastAPI tests
|
|
# -------------------------------------------------------------------
|
|
|
|
@pytest.fixture(scope="function")
|
|
async def client(db_session: AsyncSession):
|
|
def override_get_session():
|
|
try:
|
|
yield db_session
|
|
finally:
|
|
db_session.close()
|
|
|
|
app.dependency_overrides[get_session] = override_get_session
|
|
|
|
async with AsyncClient(
|
|
transport=ASGITransport(app=app), base_url="http://localhost:8000/api"
|
|
) as ac:
|
|
yield ac
|
|
|
|
|
|
# -------------------------------------------------------------------
|
|
# Auth token fixture
|
|
# -------------------------------------------------------------------
|
|
|
|
@pytest.fixture(scope="function")
|
|
async def admin_client(client: AsyncClient):
|
|
payload = {
|
|
"sub": "1",
|
|
"role": "admin",
|
|
}
|
|
auth_token = jwt.encode(payload, SECRET_KEY, algorithm="HS256")
|
|
client.headers.update({"Authorization": f"Bearer {auth_token}"})
|
|
return client
|