#Generate DB fixtures data before all Tests runned in sync definition

1 messages · Page 1 of 1 (latest)

crisp gorge
#

Hello !

Just for reference it, I have make a .conftest file that generate db data from fixtures files (json) before all tests are runned (app writted async and tests writted sync)

import asyncio
from os import environ
from pathlib import Path

from advanced_alchemy.utils.fixtures import open_fixture_async
from dotenv import load_dotenv
from litestar.contrib.sqlalchemy.base import UUIDBase
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from modules.authentication.services import RoleService, UserService

def run_async(coro):
    """used because tests are sync and app/db async"""
    return asyncio.run(coro)


def pytest_sessionstart(session):
    """pytest function who are runned before all tests on pytest call"""
    run_async(fill_db_with_fixtures())
jolly estuaryBOT
#
Notes for Generate DB fixtures data before all Tests runned in sync definition
At your assistance

@crisp gorge

No Response?

If no response in a reasonable time, ping @Member.

Closing

To close, type !solve or byte solve.

MCVE

Please include an MCVE so that we can reproduce your issue locally.

crisp gorge
# 
load_dotenv()

async def fill_db_with_fixtures():
    """
    reset test DB and fill it with fixtures data
    """
    fixtures_path = Path("tests/modules/authentication/fixtures/")
    engine = create_async_engine(environ.get("DATABASE_TEST_URL"))
    async_session_factory = async_sessionmaker(
        engine, expire_on_commit=False, class_=AsyncSession
    )
    async with engine.begin() as conn:
        await conn.run_sync(UUIDBase.metadata.drop_all)  # reset DB
        await conn.run_sync(UUIDBase.metadata.create_all)

    async with async_session_factory() as session:
        # ROLES
        async with RoleService.new(session) as service:
            fixture = await open_fixture_async(
                fixtures_path, RoleService.repository_type.model_type.__tablename__
            )
            for obj in fixture:
                _obj = await service.create(data=obj)
                if _obj.name == "admin":
                    superuser_role = _obj
            # await service.repository.session.commit() #used if not autocommit or repo usage
        # USER
        async with UserService.new(session) as service:
            fixture = await open_fixture_async(
                fixtures_path, UserService.repository_type.model_type.__tablename__
            )
            for obj in fixture:
                _obj = await service.create(data=obj)
                if _obj.email == "[email protected]":
                    _obj.roles.append(superuser_role)
            # await service.repository.session.commit() #used if not autocommit or repo usage
magic onyx
#

Thanks for this

magic onyx
#

@cursive dirge I feel this is something pytest-databases does, can you confirm if there is some reinventing the wheel going on here or not?