167 lines
4.7 KiB
Python
167 lines
4.7 KiB
Python
import pytest_asyncio
|
|
from materia.models import (
|
|
User,
|
|
LoginType,
|
|
)
|
|
from materia.models.base import Base
|
|
from materia import security
|
|
from materia.app import Application
|
|
from materia.core import Config, Database, Cache, Cron
|
|
import sqlalchemy as sa
|
|
from sqlalchemy.pool import NullPool
|
|
from httpx import AsyncClient, ASGITransport, Cookies
|
|
from asgi_lifespan import LifespanManager
|
|
from pathlib import Path
|
|
|
|
|
|
@pytest_asyncio.fixture(scope="session")
|
|
async def config() -> Config:
|
|
conf = Config()
|
|
conf.database.port = 54320
|
|
conf.cache.port = 63790
|
|
|
|
return conf
|
|
|
|
|
|
@pytest_asyncio.fixture(scope="session")
|
|
async def database(config: Config) -> Database:
|
|
config_postgres = config
|
|
config_postgres.database.user = "postgres"
|
|
config_postgres.database.name = "postgres"
|
|
database_postgres = await Database.new(
|
|
config_postgres.database.url(), poolclass=NullPool
|
|
)
|
|
|
|
async with database_postgres.connection() as connection:
|
|
await connection.execution_options(isolation_level="AUTOCOMMIT")
|
|
await connection.execute(sa.text("create role pytest login"))
|
|
await connection.execute(sa.text("create database pytest owner pytest"))
|
|
await connection.commit()
|
|
|
|
await database_postgres.dispose()
|
|
|
|
config.database.user = "pytest"
|
|
config.database.name = "pytest"
|
|
database_pytest = await Database.new(config.database.url(), poolclass=NullPool)
|
|
|
|
yield database_pytest
|
|
|
|
await database_pytest.dispose()
|
|
|
|
async with database_postgres.connection() as connection:
|
|
await connection.execution_options(isolation_level="AUTOCOMMIT")
|
|
await connection.execute(sa.text("drop database pytest")),
|
|
await connection.execute(sa.text("drop role pytest"))
|
|
await connection.commit()
|
|
await database_postgres.dispose()
|
|
|
|
|
|
@pytest_asyncio.fixture(scope="session")
|
|
async def cache(config: Config) -> Cache:
|
|
config_pytest = config
|
|
config_pytest.cache.user = "pytest"
|
|
cache_pytest = await Cache.new(config_pytest.cache.url())
|
|
|
|
yield cache_pytest
|
|
|
|
|
|
@pytest_asyncio.fixture(scope="session")
|
|
async def cron(config: Config) -> Cache:
|
|
cron_pytest = Cron.new(
|
|
config.cron.workers_count,
|
|
backend_url=config.cache.url(),
|
|
broker_url=config.cache.url(),
|
|
)
|
|
|
|
yield cron_pytest
|
|
|
|
|
|
@pytest_asyncio.fixture(scope="function", autouse=True)
|
|
async def setup_database(database: Database):
|
|
async with database.connection() as connection:
|
|
await connection.run_sync(Base.metadata.create_all)
|
|
await connection.commit()
|
|
yield
|
|
async with database.connection() as connection:
|
|
await connection.run_sync(Base.metadata.drop_all)
|
|
await connection.commit()
|
|
|
|
|
|
@pytest_asyncio.fixture()
|
|
async def session(database: Database, request):
|
|
session = database.sessionmaker()
|
|
yield session
|
|
await session.rollback()
|
|
await session.close()
|
|
|
|
|
|
@pytest_asyncio.fixture(scope="function")
|
|
async def data(config: Config):
|
|
class TestData:
|
|
user = User(
|
|
name="PyTest",
|
|
lower_name="pytest",
|
|
email="pytest@example.com",
|
|
hashed_password=security.hash_password(
|
|
"iampytest", algo=config.security.password_hash_algo
|
|
),
|
|
login_type=LoginType.Plain,
|
|
is_admin=True,
|
|
)
|
|
|
|
return TestData()
|
|
|
|
|
|
@pytest_asyncio.fixture(scope="function")
|
|
async def api_config(config: Config, tmpdir) -> Config:
|
|
config.application.working_directory = Path(tmpdir)
|
|
config.oauth2.jwt_secret = "pytest_secret_key"
|
|
|
|
yield config
|
|
|
|
|
|
@pytest_asyncio.fixture(scope="function")
|
|
async def api_client(
|
|
api_config: Config, database: Database, cache: Cache, cron: Cron
|
|
) -> AsyncClient:
|
|
|
|
app = Application(api_config)
|
|
app.database = database
|
|
app.cache = cache
|
|
app.cron = cron
|
|
app.prepare_server()
|
|
|
|
async with LifespanManager(app.backend) as manager:
|
|
async with AsyncClient(
|
|
transport=ASGITransport(app=manager.app), base_url=api_config.server.url()
|
|
) as client:
|
|
yield client
|
|
|
|
|
|
@pytest_asyncio.fixture(scope="function")
|
|
async def auth_client(api_client: AsyncClient, api_config: Config) -> AsyncClient:
|
|
data = {"name": "PyTest", "password": "iampytest", "email": "pytest@example.com"}
|
|
|
|
await api_client.post(
|
|
"/api/auth/signup",
|
|
json=data,
|
|
)
|
|
auth = await api_client.post(
|
|
"/api/auth/signin",
|
|
json=data,
|
|
)
|
|
|
|
cookies = Cookies()
|
|
cookies.set(
|
|
"materia_at",
|
|
auth.cookies[api_config.security.cookie_access_token_name],
|
|
)
|
|
cookies.set(
|
|
"materia_rt",
|
|
auth.cookies[api_config.security.cookie_refresh_token_name],
|
|
)
|
|
|
|
api_client.cookies = cookies
|
|
|
|
yield api_client
|