118 lines
3.2 KiB
Python
118 lines
3.2 KiB
Python
import asyncio
|
|
import uuid
|
|
|
|
import pytest
|
|
import pytest_asyncio
|
|
from httpx import ASGITransport, AsyncClient
|
|
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
|
|
|
|
from app.core.config import settings
|
|
from app.core.security import get_password_hash
|
|
from app.db.base import Base
|
|
from app.db.session import get_session
|
|
from app.main import app
|
|
from app.models.plant import NutrientDemand, PlantFamily, WaterDemand
|
|
from app.models.tenant import Tenant
|
|
from app.models.user import TenantRole, User, UserTenant
|
|
|
|
_engine = create_async_engine(settings.DATABASE_URL, echo=False)
|
|
_SessionLocal = async_sessionmaker(
|
|
bind=_engine,
|
|
class_=AsyncSession,
|
|
expire_on_commit=False,
|
|
)
|
|
|
|
|
|
@pytest.fixture(scope="session")
|
|
def event_loop():
|
|
loop = asyncio.new_event_loop()
|
|
yield loop
|
|
loop.close()
|
|
|
|
|
|
@pytest_asyncio.fixture(scope="session")
|
|
async def setup_database():
|
|
async with _engine.begin() as conn:
|
|
await conn.run_sync(Base.metadata.create_all)
|
|
yield
|
|
async with _engine.begin() as conn:
|
|
await conn.run_sync(Base.metadata.drop_all)
|
|
await _engine.dispose()
|
|
|
|
|
|
async def _override_get_session():
|
|
async with _SessionLocal() as session:
|
|
yield session
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def client(setup_database):
|
|
app.dependency_overrides[get_session] = _override_get_session
|
|
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as ac:
|
|
yield ac
|
|
app.dependency_overrides.clear()
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def db(setup_database):
|
|
async with _SessionLocal() as session:
|
|
yield session
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def test_tenant(db: AsyncSession):
|
|
tenant = Tenant(
|
|
id=uuid.uuid4(),
|
|
name="Test Tenant",
|
|
slug=f"test-{uuid.uuid4().hex[:8]}",
|
|
)
|
|
db.add(tenant)
|
|
await db.commit()
|
|
yield tenant
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def test_family(db: AsyncSession):
|
|
family = PlantFamily(
|
|
id=uuid.uuid4(),
|
|
name=f"Testfamilie-{uuid.uuid4().hex[:6]}",
|
|
latin_name=f"Familia testus {uuid.uuid4().hex[:6]}",
|
|
)
|
|
db.add(family)
|
|
await db.commit()
|
|
yield family
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def test_user(db: AsyncSession, test_tenant: Tenant):
|
|
email = f"user-{uuid.uuid4().hex[:6]}@test.local"
|
|
password = "testpass123"
|
|
user = User(
|
|
id=uuid.uuid4(),
|
|
email=email,
|
|
hashed_password=get_password_hash(password),
|
|
full_name="Test User",
|
|
is_superadmin=False,
|
|
)
|
|
db.add(user)
|
|
await db.flush()
|
|
db.add(UserTenant(user_id=user.id, tenant_id=test_tenant.id, role=TenantRole.READ_WRITE))
|
|
await db.commit()
|
|
# Attach plaintext password for use in auth fixtures
|
|
user._plain_password = password # type: ignore[attr-defined]
|
|
yield user
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def auth_headers(client: AsyncClient, test_user: User, test_tenant: Tenant):
|
|
resp = await client.post(
|
|
"/api/v1/auth/login",
|
|
json={"email": test_user.email, "password": test_user._plain_password}, # type: ignore[attr-defined]
|
|
)
|
|
assert resp.status_code == 200
|
|
token = resp.json()["access_token"]
|
|
return {
|
|
"Authorization": f"Bearer {token}",
|
|
"X-Tenant-ID": str(test_tenant.id),
|
|
}
|