jeancochrane / jeancochrane/pytest-flask-sqlalchemy
App session does not rollback, so tests are not isolated
- Dominant language
- Python
- Stars
- 253
- Forks
- 45
- PR merge metrics
- No merged PRs in 30d
Description
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
```
Contributor guide
No contributing guide indexed for this repository
Assessment
This issue has not been assessed yet.