Откат транзакции Flask-SQLAlchemy в нескольких базах данных с использованием привязок - PullRequest
Новая
       8

Откат транзакции Flask-SQLAlchemy в нескольких базах данных с использованием привязок

0 голосов
/ 22 февраля 2019

Я создаю приложение Flask, которое использует несколько баз данных, поэтому я использую привязки SQLAlchemy.

Поскольку я хочу изолировать все свои тесты, я создал прибор session, который начинает транзакцию и создаетscoped_session для каждого теста.После завершения теста я применяю откат к транзакции, чтобы снова получить исходное состояние.

Это работало нормально, когда у меня была только одна база данных.Тем не менее, это не работает с новым связыванием.В приведенном ниже примере у меня есть три теста:

  1. Тест один создает 2 User объектов (связывание по умолчанию).
  2. Тест два создает 2 Message объектов (messagesbind).
  3. Тест три создает 1 User и 1 Message.

После отката я печатаю количество строк в каждой таблице.Таблица User (привязка по умолчанию) всегда очищается, но Message (привязка messages) никогда не удаляет предыдущие строки.

Как удалить строки всех привязок? 

import pytest

from flask import Flask
from flask_sqlalchemy import SQLAlchemy

db = SQLAlchemy()


class User(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(80))

    def __repr__(self):
        return '<User %r>' % self.username


class Message(db.Model):
    __bind_key__ = 'messages'
    id = db.Column(db.Integer, primary_key=True)
    message = db.Column(db.String(80))

    def __repr__(self):
        return '<MyBindTable %r>' % self.message


def create_app():
    app = Flask(__name__)
    app.config[
        'SQLALCHEMY_DATABASE_URI'] = 'sqlite:///' + '/tmp/users.db'
    app.config['SQLALCHEMY_BINDS'] = {
        'messages': 'sqlite:///' + '/tmp/messages.db'
    }

    return app


@pytest.fixture(scope='session')
def test_app(request):
    app = create_app()

    ctx = app.app_context()
    ctx.push()

    def teardown():
        ctx.pop()

    request.addfinalizer(teardown)
    return app


@pytest.fixture(scope='session')
def db_fixture(test_app, request):
    db.init_app(test_app)
    db.create_all()

    def teardown():
        db.drop_all()

    request.addfinalizer(teardown)
    return db


@pytest.fixture(scope='function')
def session(db_fixture, request):
    connection = db_fixture.engine.connect()
    transaction = connection.begin()

    options = dict(bind=connection, binds={})
    session = db_fixture.create_scoped_session(options=options)

    db_fixture.session = session

    def teardown():
        transaction.rollback()
        print()
        print('Users:', len(User.query.all()))
        print('Messages:', len(Message.query.all()))
        connection.close()
        session.remove()

    request.addfinalizer(teardown)
    return session


def test_two_users(session):
    admin = User(username='user1')
    guest = User(username='user2')

    session.add(admin)
    session.add(guest)
    session.commit()

    users = User.query.all()

    assert len(users) == 2


def test_two_my_bind_table(session):
    message1 = Message(message='message1')
    message2 = Message(message='message2')

    session.add(message1)
    session.add(message2)
    session.commit()

    messages = Message.query.all()

    assert len(messages) == 2


def test_both_tables(session):
    admin = User(username='user3')
    row_bind = Message(message='message2')

    session.add(admin)
    session.add(row_bind)
    session.commit()

    messages = Message.query.all()
    users = User.query.all()

    assert len(users) == 1
    assert len(messages) == 1  # <-- It fails: len(messages) is 3

# OUTPUT:
# test.py::test_two_users PASSED
# Users: 0
# Messages: 0
#
# test.py::test_two_my_bind_table PASSED
# Users: 0
# Messages: 2  # <-- This should be 0
#
# test.py::test_both_tables FAILED
# Users: 0
# Messages: 3  # <-- This should be 0
...