App session does not rollback, so tests are not isolated

[yobuntu]

Hello, i'm struggling with sqlalchemy flask and pytest for years now, and i'm very happy to find your extension as i will maybe at least be able to run proper tests with this setting, but there must be something don't understand somewhere:

The rollback works, but only for the object inserted by the tests (in my case, the factoryboy factories). when the object are inserted by the pytest-flask.client fixture are not rolled back, despite the db.session mocking (see the test after the post test).

As i said before i spend lots of time and efforts in it but the sqlalchemy session mechanism is still a bit of voodoo art to me, so i may be doing something plain wrong.

conftest.py

import os
from datetime import date

import factory
import pytest
from factory.alchemy import SQLAlchemyModelFactory
from flask import current_app
from pytest_factoryboy import register
from sqlalchemy.orm import scoped_session

from alembic import command
from alembic.config import Config as AlembicConfig
from chores import create_app
from chores.configmodule import Config
from chores.model import Chore, Task

Session = scoped_session(
    lambda: current_app.extensions["sqlalchemy"].db.session,
    scopefunc=lambda: current_app.extensions["sqlalchemy"].db.session,
)


class BaseFactory(SQLAlchemyModelFactory):
    class Meta:
        abstract = True
        sqlalchemy_session = Session
        sqlalchemy_session_persistence = "flush"


@pytest.fixture(scope="session")
def database():
    ini_location = os.path.join(os.path.dirname(__file__), "..", "alembic.ini")
    sqlalchemy_url = Config.get_config("test").SQLALCHEMY_DATABASE_URI
    alembic_config = AlembicConfig(ini_location)
    alembic_config.set_main_option("sqlalchemy.url", sqlalchemy_url)
    command.upgrade(alembic_config, "head")
    yield sqlalchemy_url
    command.downgrade(alembic_config, "base")


@pytest.fixture(scope="session")
def app(database):
    app = create_app("test")
    yield app


@pytest.fixture(scope="session")
def _db(app):
    yield app.extensions["sqlalchemy"].db


@register
class ChoreFactory(BaseFactory):
    class Meta:
        model = Chore

    name = factory.Sequence(lambda n: f"chore_{n}")
    description = factory.Sequence(lambda n: f"description of chore_{n}")
    period = 5


@register
class TaskFactory(BaseFactory):
    class Meta:
        model = Task

    execution_date = date(year=2020, month=5, day=18)
    chore = factory.SubFactory(ChoreFactory)

setup.cfg

[tool:pytest]
mocked-sessions = chores.model.db.session

chores/model.py

from flask_sqlalchemy import SQLAlchemy

db = SQLAlchemy()


class Chore(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(), unique=True, nullable=False)
    description = db.Column(db.String())
    period = db.Column(db.Integer())


class Task(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    chore_id = db.Column(db.Integer, db.ForeignKey("chore.id"), nullable=False)
    chore = db.relationship("Chore", backref=db.backref("tasks", lazy=True))
    execution_date = db.Column(db.Date(), nullable=False)

test_chore.py

  def test_get_chore_should_return_list_of_chores(
        self, client, chore_factory, db_session
    ):
        chore1 = chore_factory()
        chore2 = chore_factory()

        r = client.get(url_for("chores.chores"))

        assert r.status_code == 200
        chores = json.loads(r.data)
        assert "chores" in chores
        assert chores["chores"][0]["name"] == chore1.name
        assert chores["chores"][1]["name"] == chore2.name

   def test_chore_is_empty_before_post(self, db_session):
        chores = db_session.query(Chore).all()
        assert len(chores) == 0

    def test_post_chore_should_add_a_chore(self, client):
        mimetype = "application/json"
        headers = {
            "Content-Type": mimetype,
            "Accept": mimetype,
        }
        chore_to_post = {
            "name": "a new chore",
            "description": "description of the new chore",
            "period": 9,
        }
        r = client.post(
            url_for("chores.chores"), data=json.dumps(chore_to_post), headers=headers
        )

        assert r.status_code == 200
        chore = Chore.query.filter_by(name=chore_to_post["name"]).one()
        assert chore.name == chore_to_post["name"]
        assert chore.period == chore_to_post["period"]
        assert chore.description == chore_to_post["description"]

    def test_chore_should_be_empty(self, db_session):
        chores = db_session.query(Chore).all()
        assert len(chores) == 0

[imanebosch]

Did you manage to find a solution? I'm dealing with the same problem

[yobuntu]

unfortunatly i did not find a solution, i tryed for a week, then posted this message as a bottle in the sea…

The workaround i use for now is to set the database fixture with a session scope, so the whole database is dropped and restored between each test, which is good for test isolation but very bad for time if your db contains many data and you have lots of test, so i hope to find a solution before my tests become more complex

[jayaddison]

@yobuntu Have you tried adding the db_session fixture to the test_post_chore_should_add_a_chore test case?

Without that fixture, I don't know whether the plugin will wrap your test case in a transaction.

[yobuntu]

i finaly managed to make it work !

so there was 2 main caveats:

• first : it is really important to configure mocked-sessions to the sessions your code uses !
• second: pytest has some kind of «cache» but making changes in conftest and setup.cfg without touching the test file somtimes makes weird things

so now that i compare what works and what doesn't , it seams the only diff is in the _db fixture which is now:

@pytest.fixture(scope="session")
def _db(app):
    yield db

where db is directly imported from the model (see the one which is mocked in setup.cfg

more over i changed the BaseFactory class to

class BaseFactory(SQLAlchemyModelFactory):
    class Meta:
        abstract = True
        sqlalchemy_session = Session
        sqlalchemy_session_persistence = "commit"

so i'm sure there are no side effect depending of a session.commit() is or is not in the tested code

I hope this will be usefull to you

also i autouse the db_session fixture with :

@pytest.fixture(autouse=True)
def enable_transactional_test(request, db_session):
    pass

so i don't need to pass the db_session fixture in every test

(but still i wonder if i have a full understanding of the problem)