Tukukkk

I'm working on a Python application that uses the following SQLAlchemy pattern for separating unit tests into isolated transactions, and using pytest fixtures:

@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()

I'm defining the get_session() utility function such that I can also use it in my application code:

def get_session(bind=None):

    if bind is None:

        bind = create_engine(DATABASE_URL)

    return scoped_session(sessionmaker(bind=bind))

This pattern works flawlessly in unit test code that uses the session fixture directly:

def test_foo(session):

    assert session.query(Foo).count() == 0

However, whenever I want to test application code that should access the session by itself, without it being passed explicitly from the unit test, I get a failure:

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

Since I'm using a scoped_session I would expect that any application code that calls get_session() will get the same session that is bound from the test fixture, but that doesn't happen.

I know I can workaround this by passing the session directly to the function, but that seems just wrong.

Any better way to accomplish this?