materia/tests/conftest.py

167 lines
4.7 KiB
Python
Raw Normal View History

2024-08-14 00:56:30 +05:00
import pytest_asyncio
from materia.models import (
User,
LoginType,
)
from materia.models.base import Base
from materia import security
2024-08-30 12:38:43 +05:00
from materia.app import Application
from materia.core import Config, Database, Cache, Cron
2024-08-14 00:56:30 +05:00
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
2024-08-30 12:38:43 +05:00
@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
2024-08-14 00:56:30 +05:00
@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(
2024-08-30 12:38:43 +05:00
api_config: Config, database: Database, cache: Cache, cron: Cron
2024-08-14 00:56:30 +05:00
) -> AsyncClient:
2024-08-30 12:38:43 +05:00
app = Application(api_config)
app.database = database
app.cache = cache
app.cron = cron
2024-08-30 23:28:58 +05:00
app.prepare_server()
2024-08-30 12:38:43 +05:00
async with LifespanManager(app.backend) as manager:
2024-08-14 00:56:30 +05:00
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