Управление сеансами SQLAlchemy в функциях, независимых от тестов - PullRequest
Новая
       8

Управление сеансами SQLAlchemy в функциях, независимых от тестов

0 голосов
/ 19 ноября 2018

Я работаю над приложением Python, которое использует следующий шаблон SQLAlchemy для разделения модульных тестов на изолированные транзакции и использования приспособлений pytest: 

@fixture(scope='function')
def session(engine):
    connection = engine.connect()
    transaction = connection.begin()
    Session = get_session(bind=connection)

    session = Session()
    session.begin_nested()

    @listens_for(session, 'after_transaction_end')
    def resetart_savepoint(sess, trans):
        if trans.nested and not trans._parent.nested:
            session.expire_all()
            session.begin_nested()

    yield session

    session.close()
    transaction.rollback()
    connection.close()

Я определяю служебную функцию get_session() так, чтобы я также мог использовать ее в своем коде приложения: 

def get_session(bind=None):
    if bind is None:
        bind = create_engine(DATABASE_URL)
    return scoped_session(sessionmaker(bind=bind))

Этот шаблон работает безупречно в коде модульного тестирования, который напрямую использует прибор session: 

def test_foo(session):
    assert session.query(Foo).count() == 0

Однако , всякий раз, когда я хочу протестировать код приложения, который должен сам обращаться к сеансу, без явной передачи из модульного теста, я получаю ошибку: 

def create_foo_obj():
    Session = get_session()
    session = Session()
    session.add(Foo())
    session.commit()

def test_foo(session):
    create_foo_obj()
    assert session.query(Foo).count() == 1  # FAILS

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

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

Есть ли лучший способ сделать это?

...