ormar transactions in pytest

[cmflynn]

Hey im not sure if this truly a bug or not, so im opening the discussion here. perhaps related to: encode/databases#456

so Im trying to configure pytest fixtures to run each test in a transaction that will rollback after the test finishes. problem is the rollback is not actually preventing the data from being comitted in the first test, thus throwing primary key errors in the second test.

Ive verified the code in the testing docs here: https://collerek.github.io/ormar/transactions/ works in isolation, but cant seem to get it working in pytest fixtures. ive used a very similar pattern in sqlalchemy and it works fine so not sure what im missing here.

import databases
import ormar
import pytest
import sqlalchemy
from databases import Database

database = Database(url="postgresql://postgres@0.0.0.0:5432/postgres")
engine = sqlalchemy.create_engine("postgresql://postgres@0.0.0.0:5432/postgres")
metadata = sqlalchemy.MetaData()


class BaseMeta:
    metadata = metadata
    database = database


class Jimmy(ormar.Model):
    class Meta(BaseMeta):
        tablename = "jimmy_rus"

    name = ormar.String(primary_key=True, max_length=42)


@pytest.fixture(scope="session")
def anyio_backend():
    return "asyncio"


@pytest.mark.anyio
@pytest.fixture(scope="session", autouse=True)
async def database_init():

    await database.connect()

    BaseMeta.metadata.drop_all(bind=engine)
    BaseMeta.metadata.create_all(bind=engine)

    yield database

    BaseMeta.metadata.drop_all(bind=engine)
    await database.disconnect()


@pytest.mark.anyio
@pytest.fixture(scope="function", autouse=True)
async def db_session(
    database_init,
):
    async with database_init:
        async with database_init.transaction(force_rollback=True):
            yield


@pytest.mark.anyio
async def test_jimmy_can():
    await Jimmy.objects.create(
        name="jimmy",
    )
    await Jimmy.objects.create(
        name="jimmyRus",
    )
    jimmies = await Jimmy.objects.all()


@pytest.mark.anyio
async def test_jimmy_cant():
    jimmies = await Jimmy.objects.all()
    assert not jimmies  # fails here: E AssertionError: assert not [Jimmy({'name': 'jimmy'}), Jimmy({'name': 'jimmyRus'})]


    await Jimmy.objects.create(
        name="jimmy",
    )
    await Jimmy.objects.create(
        name="jimmyRus",
    )
    jimmies = await Jimmy.objects.all()

[cmflynn]

Update:
setting force_rollback=True on the Database object itself, has fixed this. However, I dont think that should be required for this pattern to work?
database = Database(url="postgresql://postgres@0.0.0.0:5432/postgres", force_rollback=True)

[SepehrBazyar]

Hi @cmflynn
First you will need the following fixture:

@fixture(autouse=True, scope="session")
async def create_test_database():
    """SetUp & TearDown Fixture to Create & Delete Database by Metadata"""

    async with engine.begin() as conn:
        await conn.run_sync(metadata.create_all)
        yield
        await conn.run_sync(metadata.drop_all)

Then you will empty the tables with below fixture:

@fixture(autouse=True, scope="function")
async def clean_tables_database():
    """Clearing Database Tables to Delete All Rows for Each Test Case"""

    try:
        yield
    finally:
        async with engine.begin() as conn:
            for table in reversed(metadata.sorted_tables):
                await conn.execute(table.delete())

Or you write this at the beginning of each test:

@pytest.mark.asyncio
async def test_something():
    async with database:
        async with database.transaction(force_rollback=True):
            # write your tests

[cmflynn]

Hi @SepehrBazyar, the issue is the last snippet you shared does not work as a pytest fixture. Unfortunately wrapping every test in that context manager is not an option so I have to use a fixture. Dropping the tables clearly works but is not my ideal solution. Ideally each test is ran in a transaction that is rolled back. If you see my update, forcing roll back on the Database object itself works but has other side effects that are problematic.

[SepehrBazyar]

Hi @SepehrBazyar, the issue is the last snippet you shared does not work as a pytest fixture. Unfortunately wrapping every test in that context manager is not an option so I have to use a fixture. Dropping the tables clearly works but is not my ideal solution. Ideally each test is ran in a transaction that is rolled back. If you see my update, forcing roll back on the Database object itself works but has other side effects that are problematic.

Yes, the last code is not an fixture and it must be written at the beginning of each test. You can read the documentation for more explanations.

If you want a fixture code, the second example I wrote will do it for you, and the way it works is to delete the rows of the tables, not drop them.

[cmflynn]

Yep that all makes sense, but the last example from the docs should be able to be used as a fixture. The fact it doesn't seems like a bug to me.

[collerek]

It works as expected in 0.22.0

import ormar
import pytest

from tests.lifespan import init_tests
from tests.settings import create_config

base_ormar_config = create_config()


class Jimmy(ormar.Model):
    ormar_config = base_ormar_config.copy(tablename="jimmies")
    name = ormar.String(primary_key=True, max_length=42)


create_test_database = init_tests(base_ormar_config)


@pytest.fixture(scope="session")
def anyio_backend():
    return "asyncio"


@pytest.mark.anyio
@pytest.fixture(scope="function", autouse=True)
async def db_session():
    async with base_ormar_config.database:
        async with base_ormar_config.database.transaction(force_rollback=True):
            yield


@pytest.mark.anyio
async def test_jimmy_can():
    await Jimmy.objects.create(
        name="jimmy",
    )
    await Jimmy.objects.create(
        name="jimmyRus",
    )
    jimmies = await Jimmy.objects.all()
    assert len(jimmies) == 2


@pytest.mark.anyio
async def test_jimmy_cant():
    jimmies = await Jimmy.objects.all()
    assert not jimmies

    await Jimmy.objects.create(
        name="jimmy",
    )
    await Jimmy.objects.create(
        name="jimmyRus",
    )
    jimmies = await Jimmy.objects.all()
    assert len(jimmies) == 2