111 lines
2.9 KiB
Python
111 lines
2.9 KiB
Python
"""Test fixtures — async database and HTTP client."""
|
|
|
|
import pytest
|
|
import pytest_asyncio
|
|
from sqlalchemy.ext.asyncio import (
|
|
create_async_engine,
|
|
async_sessionmaker,
|
|
)
|
|
from httpx import ASGITransport, AsyncClient
|
|
|
|
from app.database import Base, get_db
|
|
from app.main import app
|
|
from app.auth import hash_password
|
|
from app.models import User
|
|
|
|
import os
|
|
|
|
DATABASE_URL = os.getenv(
|
|
"DATABASE_URL",
|
|
"postgresql+asyncpg://pokedexter:pokedexter_secret@localhost/pokedexter",
|
|
)
|
|
|
|
|
|
@pytest_asyncio.fixture(scope="session")
|
|
async def engine():
|
|
"""Create a fresh database for testing."""
|
|
e = create_async_engine(DATABASE_URL, echo=False)
|
|
async with e.begin() as conn:
|
|
await conn.run_sync(Base.metadata.drop_all)
|
|
await conn.run_sync(Base.metadata.create_all)
|
|
yield e
|
|
async with e.begin() as conn:
|
|
await conn.run_sync(Base.metadata.drop_all)
|
|
await e.dispose()
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def session(engine):
|
|
"""Per-test database session.
|
|
|
|
Data commited her er synlig for appens egen session via HTTP client.
|
|
Renses automatisk af _clean_db efter hver test.
|
|
"""
|
|
maker = async_sessionmaker(bind=engine, expire_on_commit=False)
|
|
async with maker() as s:
|
|
yield s
|
|
|
|
|
|
@pytest_asyncio.fixture(autouse=True)
|
|
async def _clean_db(engine):
|
|
"""Slet alt data fra alle tabeller efter hver test."""
|
|
yield
|
|
async with engine.begin() as conn:
|
|
for table in reversed(Base.metadata.sorted_tables):
|
|
await conn.execute(table.delete())
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def client():
|
|
"""HTTP client — appen bruger sin egen DB session via get_db()."""
|
|
transport = ASGITransport(app=app)
|
|
async with AsyncClient(transport=transport, base_url="http://test") as ac:
|
|
yield ac
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def admin_user(session):
|
|
"""Insert admin user into test DB (ryddes af _clean_db)."""
|
|
u = User(
|
|
username="admin",
|
|
password_hash=hash_password("admin123"),
|
|
role="admin",
|
|
)
|
|
session.add(u)
|
|
await session.commit()
|
|
return u
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def normal_user(session):
|
|
"""Insert normal user into test DB (ryddes af _clean_db)."""
|
|
u = User(
|
|
username="testuser",
|
|
password_hash=hash_password("test123"),
|
|
role="user",
|
|
)
|
|
session.add(u)
|
|
await session.commit()
|
|
return u
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def admin_token(client, admin_user):
|
|
"""Log in as admin and return the JWT token."""
|
|
resp = await client.post("/api/login", json={
|
|
"username": "admin",
|
|
"password": "admin123",
|
|
})
|
|
assert resp.status_code == 200
|
|
return resp.cookies.get("token")
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def user_token(client, normal_user):
|
|
"""Log in as normal user and return the JWT token."""
|
|
resp = await client.post("/api/login", json={
|
|
"username": "testuser",
|
|
"password": "test123",
|
|
})
|
|
assert resp.status_code == 200
|
|
return resp.cookies.get("token") |