Eror

FAILED tests/users/api/test_auth.py::test_create_user_failed_unique_phone_number - RuntimeError: Task <Task pending name='Task-22' coro=<test_create_user_failed_unique_phone_number() running at /app/tests/users/api/test_auth.py:36> cb=[_run_until_complete_cb() at /usr/local/lib/python3.12/asyncio/base_events...

Error happens all time when I try to call db.commit. For database connection I use for database url postgresql+asyncpg.

# conftest.py


pytest_plugins = [
    "tests.users.factories.users",
]

async def init_db(engine):
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)

async def close_db(engine):
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)

@pytest.fixture(scope="session", autouse=True)
async def db_engine():
    engine = create_async_engine(settings.DATABASE_URL)
    await init_db(engine)
    yield engine
    await close_db(engine)
    await engine.dispose()

@pytest.fixture
async def db() -> AsyncIterable[AsyncSession]:
    async with SessionLocal() as session:
        yield session

@pytest.fixture
async def db_ins(db: AsyncSession) -> AsyncIterable[DBInsert]:
    async def insert(factory: SQLAlchemyModelFactory, **kwargs: Any) -> Any:
        instance = factory.build(**kwargs)
        db.add(instance)
        await db.commit()
        return instance
    yield insert

@pytest.fixture(scope="session")
async def client() -> AsyncIterable[AsyncClient]:
    async with AsyncClient(app=app, base_url="http://testserver") as client:
        yield client

# tests.users.factories.users

class UserFactory(SQLAlchemyModelFactory):
    class Meta:
        model = Users
        abstract = True

    first_name = factory.Faker("first_name")
    last_name = factory.Faker("last_name")
    email = factory.Faker("email")
    phone = factory.Sequence(lambda n: fake_phone_number())
    is_active = True

    hash_password: str = factory.LazyAttribute(lambda _: PasswordHelper().hash("12345678"))

class AdminUserFactory(UserFactory):
    role = Role.ADMIN

@pytest.fixture
async def admin_user(db_ins: DBInsert) -> Users:
    user = await db_ins(AdminUserFactory)
    return user

When I use admin_user fixture, I receive RuntimeError: Event loop is closed exception. It happened when we call db.commit() in db_ins fixture.

I want to fix this problem with database

It's an issue of latest versions of pytest-asyncio (https://github.com/pytest-dev/pytest-asyncio/issues/706). This happens when you use async fixtures with different scopes. To make this working use pytest-asyncio version 0.21.1 or less