For every unit test I want to rollback already commited statements using pytest.fixture
and scoped_session
.
engine = create_engine(DATABASE_URL)
init_database
creates the database if it doesn't exist using sqlalchemy
import pytest
from sqlalchemy_utils import drop_database
from database_test_setup.manage import Session
@pytest.fixture(scope="session")
def db_engine():
from settings import settings
from database import engine, init_database, DATABASE_URL
init_database(user=settings.DB_USER, pwd=settings.DB_PWD, host=settings.DB_HOST, port=settings.DB_PORT,
db_name=settings.DB_NAME)
Session.configure(bind=engine)
yield engine
drop_database(DATABASE_URL)
engine.dispose()
@pytest.fixture(scope="function", autouse=True)
def db_session(db_engine):
session_ = Session()
session_.begin_nested()
yield session_
session_.rollback()
Order
is a simple SQLAlchemy modelfrom factory import Sequence
from factory.alchemy import SQLAlchemyModelFactory
from database_test_setup.manage import Session
from order.models import Order
class BaseFactory(SQLAlchemyModelFactory):
"""Base Factory."""
class Meta:
"""Factory configuration."""
abstract = True
sqlalchemy_session = Session
sqlalchemy_session_persistence = "flush"
class OrderFactory(BaseFactory):
id = Sequence(lambda n: n)
quality = "X"
start_date = "2021-09-15T17:53:00"
end_date = "2021-09-15T15:53:00"
class Meta:
model = Order
from sqlalchemy.orm import scoped_session, sessionmaker
Session = scoped_session(sessionmaker())
When I use the db_session
fixture in my test to commit a statement, the statement will not be affected by session_.rollback
.
The same is true if I set sqlalchemy_session_persistence = "commit"
in my SQLAlchemyModelFactory
.
I understand that this question has been answered before, but the proposed solution doesn't seem to work for me, even though I use the same setup except using a separate module for creating a scoped_session
instead of a fixture to use additionally in my SQLAlchemyModelFactory
.
I had the same issue. It seems, this does the trick. You have to add a transaction sublayer and roll that back. My fixture looks like so:
@pytest.fixture(scope="function")
def session(engine):
# See https://docs.sqlalchemy.org/en/20/orm/session_transaction.html#joining-a-session-into-an-external-transaction-such-as-for-test-suites
# join_transaction_mode="create_savepoint" gives error
connection = engine.connect()
transaction = connection.begin()
# Create a new session
session = Session(bind=connection)
yield session
# rollback - everything that happened with the
# Session above (including calls to commit())
# is rolled back.
session.close()
transaction.rollback()
connection.close()