gistfile1.py

@pytest.fixture(scope='session')
def dbconnection(request):
    """Returns an SQLALchemy Connection to the triancore_unittest database.
    Starts a transation and creates the tables in it.
    The transaction is rolled back in teardown.
    """

    def finish():
        transaction.rollback()
        connection.close()
        engine.dispose()

    # Connect to the database and create the schema within a transaction
    engine = create_engine('postgresql:///triancore_unittest')
    connection = engine.connect()
    transaction = connection.begin()
    Base.metadata.create_all(connection)

    request.addfinalizer(finish)
    return connection


@pytest.fixture()
def dbsession(request, dbconnection):
    """Returns an SQLAlchemy Session, wrapped in a nested transaction which is rolled back on teardown."""

    def finish():
        session.close()
        transaction.rollback()

    transaction = dbconnection.begin_nested()
    session = Session(dbconnection)
    request.addfinalizer(finish)
    return session


@pytest.fixture()
def datafixture(dbsession):
    """Returns a function that adds the given objects(s) to the database."""
    
    def add(objects):
        objects = [objects] if not isinstance(objects, (tuple, list)) else objects
        dbsession.add_all(objects)
        dbsession.flush()
    
    return add