Создание отдельных сеансов SQLAlchemy между интеграционным тестом (Pytest) и приложением Flask - PullRequest
Новая
       15

Создание отдельных сеансов SQLAlchemy между интеграционным тестом (Pytest) и приложением Flask

2 голосов
/ 10 января 2020

У меня есть flask приложение (бэкэнд RESTful API), которое использует flask -sqlalchemy lib. Интеграционный тест использует Pytest с приборами, некоторые создают записи для целей тестирования. Проблема заключается в том, что при тестировании сценария, который, как ожидается, приведет к возникновению уникальной ошибки ограничения на уровне базы данных, все записи, созданные для теста с помощью осветителей, будут откатаны, что приведет к сбою других тестов с ошибкой sqlalchemy.exc.InvalidRequestError: This Session's transaction has been rolled back due to a previous exception during flush. To begin a new transaction with this Session, first issue Session.rollback(). Original exception was: (sqlite3.IntegrityError) UNIQUE constraint failed: my_table.field1.

Как Я go о создании отдельного сеанса SQLAlchemy для тестирования, чтобы тестовые записи могли фиксироваться в БД и не подвергаться влиянию ошибок, которые происходят внутри Flask жизненного цикла запроса? 

### globals.py ###

from flask_sqlalchemy import SQLAlchemy

db = SQLAlchemy()

### app.py ###

from sqlalchemy.exc import IntegrityError
from .globals import db

def handle_unique_constraint(error):
    return {"msg": "Duplicate"}, 409

def create_app(connection_string):
    app = Flask(__name__)
    app.register_error_handler(IntegrityError, handle_unique_constraint)
    app.config['SQLALCHEMY_DATABASE_URI'] = connection_string

    # register API blueprint ...
    # e.g. create new user record, done like this:
    #
    # db.session.add(User(**{'email': path_param}))
    # db.session.commit()

    db.init_app(app)
    return app

### conftest.py ###

from my_package.app import create_app
from my_package.globals import db as app_db
from my_package.models.user import User

@fixture(scope='session')
def app(request):
    app = create_app('https://localhost')
    app.debug = True

    with app.app_context():
        yield app

@fixture(scope='session')
def db(app):
    app_db.create_all()
    return app_db

@fixture(scope='session')
def client(app):

    with app.test_client() as client:
        yield client

@fixture(scope='function')
def test_user(db):
    user = User(**{'email': generate_random()})
    db.session.add(user)
    db.session.commit()
    db.session.refresh(user)

### test_user.py ###



# This test passses

def test_repeat_email(client, test_user):
    resp = client.post('/users/emails/{}'.format(test_user.email))
    assert resp.status_code == 409


# This test errors out during setting up test_user fixture
# with aforementioned error message

def test_good_email(client, test_user): # <- this 
    resp = client.post('/users/emails/{}'.format('unique@example.com'))
    assert resp.status_code == 201
...