Trying to use py.test fixtures with Pecan + SQLAlchemy

gistfile1.py

# app name: ayni
# this works, inheriting from it, committing and re-using the database
#
# ayni model __init__.py file doesn't have anything odd and can be found:
# https://github.com/alfredodeza/ayni/blob/master/ayni/models/__init__.py
#
import os
from unittest import TestCase
from pecan import set_config
from pecan.testing import load_test_app

from copy import deepcopy
import os
import subprocess

from pecan import conf
from pecan import configuration
from sqlalchemy import create_engine
from sqlalchemy.pool import NullPool

from ayni import models as amodel


class TestModel(object):

    config = configuration.conf_from_file(config_file()).to_dict()

    __db__ = None

    @classmethod
    def setup_class(cls):
        if TestModel.__db__ is None:
            TestModel.__db__ = 'aynitest'
            # Connect and create the temporary database
            print "=" * 80
            print "CREATING TEMPORARY DATABASE FOR TESTS"
            print "=" * 80
            #subprocess.call(['dropdb', TestModel.__db__])
            subprocess.call(['createdb', TestModel.__db__])

            # Bind and create the database tables
            amodel.clear()
            engine_url = '%s/%s' % (__bind__, TestModel.__db__)

            db_engine = create_engine(
                    engine_url,
                    encoding='utf-8',
                    poolclass=NullPool)

            # AKA models.start()
            amodel.Session.bind = db_engine
            amodel.metadata.bind = amodel.Session.bind

            amodel.Base.metadata.create_all(db_engine)
            amodel.commit()
            amodel.clear()

    def setup(self):
        config = deepcopy(self.config)

        # Add the appropriate connection string to the app config.
        config['sqlalchemy'] = {
            'url': '%s/%s' % (__bind__, TestModel.__db__),
            'encoding': 'utf-8',
            'poolclass': NullPool
        }

        # Set up a fake app
        self.app = self.load_test_app(config)
        amodel.start()

    def load_test_app(self, config):
        return load_test_app(config)

    def teardown(self):
        from sqlalchemy.engine import reflection

        # Tear down and dispose the DB binding
        amodel.clear()

        # start a transaction
        engine = conf.sqlalchemy.engine
        conn = engine.connect()
        trans = conn.begin()

        inspector = reflection.Inspector.from_engine(engine)

        # gather all data first before dropping anything.
        # some DBs lock after things have been dropped in
        # a transaction.
        conn.execute("TRUNCATE TABLE %s RESTART IDENTITY CASCADE" % (
            ', '.join(inspector.get_table_names())
        ))

        trans.commit()
        conn.close()


# but these fixtures in a conftest.py file, although working, it will not rollback
# the transaction, causing assertions errors in tests that re-use the fixture
# Adapted from the nice examples in http://alexmic.net/flask-sqlalchemy-pytest/
# and trying to make Pecan do something like described in the docs, here:
# http://docs.sqlalchemy.org/en/rel_0_9/orm/session.html#joining-a-session-into-an-external-transaction-such-as-for-test-suites


DBNAME = 'aynitest'
BIND = 'postgresql+psycopg2://localhost'


def config_file():
    here = os.path.abspath(os.path.dirname(__file__))
    return os.path.join(here, 'config.py')


@pytest.fixture(scope='session')
def app(request):
    config = configuration.conf_from_file(config_file()).to_dict()

    # Add the appropriate connection string to the app config.
    config['sqlalchemy'] = {
        'url': '%s/%s' % (BIND, DBNAME),
        'encoding': 'utf-8',
        'poolclass': NullPool
    }

    # Set up a fake app
    app = load_test_app(config)
    return app


@pytest.fixture(scope='session')
def db(request):
    """Session-wide test database."""
    # Connect and create the temporary database
    print "=" * 80
    print "CREATING TEMPORARY DATABASE FOR TESTS"
    print "=" * 80
    subprocess.call(['dropdb', DBNAME])
    subprocess.call(['createdb', DBNAME])

    # Bind and create the database tables
    _db.clear()
    engine_url = '%s/%s' % (BIND, DBNAME)

    db_engine = create_engine(
        engine_url,
        encoding='utf-8',
        poolclass=NullPool)

    # AKA models.start()
    _db.Session.bind = db_engine
    _db.metadata.bind = _db.Session.bind

    _db.Base.metadata.create_all(db_engine)
    _db.commit()
    _db.clear()

    def teardown():
        _db.Base.metadata.drop_all(db_engine)

    request.addfinalizer(teardown)
    return _db


@pytest.fixture(scope='function')
def session(db, app, request):
    """Creates a new database session for a test.
    Trying to adapt SQLA's way of joining a transaction described
    here: http://docs.sqlalchemy.org/en/rel_0_9/orm/session.html#joining-a-session-into-an-external-transaction-such-as-for-test-suites
    """
    # start a transaction
    engine = conf.sqlalchemy.engine
    connection = engine.connect()
    transaction = connection.begin()

    Session = scoped_session(sessionmaker())
    Session(bind=connection, binds={})
    db.Session = Session

    def teardown():
        transaction.rollback()
        Session.close()
        connection.close()

    request.addfinalizer(teardown)
    return db