Trying to get a 'global' session to work for SQLAlchemy database library tests

[Ken Lareau]

In the process of trying to find an efficient way to manage a test database

for a large set of tests for a database library I'm writing that uses SQLAlchemy,

I came across this page:

    http://alextechrants.blogspot.fi/2013/08/unit-testing-sqlalchemy-apps.html

This is definitely what I want to do, with one catch: I already have a session
management system in place for the library that seems to conflict with the

sample code given on the webpage, and I'm not having luck reconciling it.

Basically I have an 'init_session' I use that looks as follows:

def init_session(db_user, db_password, **kwargs):
    """Initialize database session"""

    if 'hostname' not in kwargs or 'db_name' not in kwargs:
        db_host, db_name = load_db_config()
        kwargs.setdefault('hostname', db_host)
        kwargs.setdefault('db_name', db_name)

    dbconn_string = create_dbconn_string(db_user, db_password, **kwargs)
    engine = create_engine(dbconn_string)

    # Ensure connection information is valid
    try:
        engine.execute('select 1').scalar()
    except sqlalchemy.exc.DBAPIError, e:
        raise PermissionsException(e)

    Session.configure(bind=engine)

Session is defined as follows (in another file at module-level):

Session = scoped_session(sessionmaker())

From all other modules, I simply do a:

from <db-meta-module> import Session

and I am able to use Session to manage all my access to the ORM.

My attempt to modify the code on the webpage currently looks like this:

import unittest2 as unittest

from tagopsdb.database import init_database, init_session
from tagopsdb.database.meta import Session


def setup_module():
    global transaction, connection, engine

    # Connect to the database and create the schema within a transaction
    engine = init_session('dbtest', 'dbtestpasswd', hostname='localhost',
                          db_name='TagOpsDB_Testing')
    init_database()

    # If you want to insert fixtures to the DB, do it here


def teardown_module():
    # Roll back the top level transaction and disconnect from the database
    Session.rollback()
    Session.close()
    engine.dispose()


class DatabaseTest(unittest.TestCase):
    """Base unittest class to manage nested transactions for testing"""

    def setup(self):
        self.__transaction = Session.begin_nested()


    def teardown(self):
        self.__transaction.rollback()

I modified my test classes to subclass DatabaseTest, but an attempt
to run the tests results in:

UnboundExecutionError: Could not locate a bind configured on mapper Mapper|Environments|environments or this Session

This does not completely surprise me as my understanding about sessions
in SQLAlchemy is still not solid enough to always understand why my

current code works (and I have a full application using this without issue
at the moment).

So I guess the question is: am I missing something really obvious here,

or will I need to rethink how I deal with sessions in my library code?

Thanks in advance.

--
- Ken Lareau

[Michael Bayer]

On Oct 11, 2013, at 7:14 PM, Ken Lareau <kla...@tagged.com> wrote:

In the process of trying to find an efficient way to manage a test database

for a large set of tests for a database library I'm writing that uses SQLAlchemy,

I came across this page:

    http://alextechrants.blogspot.fi/2013/08/unit-testing-sqlalchemy-apps.html

This is definitely what I want to do, with one catch: I already have a session
management system in place for the library that seems to conflict with the

sample code given on the webpage, and I'm not having luck reconciling it.

I find the approach on that page a little awkward - it's using new globals for no good reason and also the begin_nested() seems strange.   The test itself then has a "self.session", so the test itself is using a test-bound session, the choice of globals for "connection" and "engine" seems even more weird.

The way this works is:

1. test fixture gets at an engine, from configurational system, locally, whereever.

2. test fixture gets a connection, holds onto it locally.

3. test fixture gets a transaction from connection - this is a top level transaction, using connection.begin()

4. test fixture then does whatever the test needs to get at a session.  if the code being tested relies upon a global registry, it injects the connection.  below is using a traditional scoped session:

    def setUp(self):

        self.conn = engine.connect()

        self.trans = self.conn.begin()

        from application.model import the_scoped_session

        self.session = the_scoped_session(bind=self.conn)

now above, the test fixture has a hold on "self.session".  but - this is the *same* session that's in the registry (the registry here being "application.model.the_scoped_session").   if some other code somewhere calls upon the_scoped_session(), they get the *same* session.   it's a registry, that's the point of it.

if you have some other kind of registration thing in place, you'd need to figure out how to load it up with a new Session bound to that local connection.

5. test fixture releases the session:

    def tearDown(self):

        the_scoped_session.remove()

        self.trans.rollback()

        self.conn.close()

so note, we don't have to bother doing anything special with the Session at teardown time, we just dump it.   we roll back the transaction that we've created externally to it.

an example of running through this is in the docs at:

http://docs.sqlalchemy.org/en/rel_0_8/orm/session.html#joining-a-session-into-an-external-transaction

and I am able to use Session to manage all my access to the ORM.

My attempt to modify the code on the webpage currently looks like this:

import unittest2 as unittest

from tagopsdb.database import init_database, init_session
from tagopsdb.database.meta import Session


def setup_module():
    global transaction, connection, engine

    # Connect to the database and create the schema within a transaction
    engine = init_session('dbtest', 'dbtestpasswd', hostname='localhost',
                          db_name='TagOpsDB_Testing')
    init_database()

    # If you want to insert fixtures to the DB, do it here


def teardown_module():
    # Roll back the top level transaction and disconnect from the database
    Session.rollback()
    Session.close()
    engine.dispose()

I tend to organize things such that the scope of this transaction is *per test*, not per module as you're doing.  the engine, that can be per module, or preferably per-application.   But i'd be linking the lifespan of the Session to that of the transaction (which again begin_nested() should be a begin()).

I modified my test classes to subclass DatabaseTest, but an attempt
to run the tests results in:

UnboundExecutionError: Could not locate a bind configured on mapper Mapper|Environments|environments or this Session

make sure the Session is created with an explicit bind to the connection.

[Ken Lareau]

On Fri, Oct 11, 2013 at 5:53 PM, Michael Bayer <mik...@zzzcomputing.com> wrote:

On Oct 11, 2013, at 7:14 PM, Ken Lareau <kla...@tagged.com> wrote:

In the process of trying to find an efficient way to manage a test database

for a large set of tests for a database library I'm writing that uses SQLAlchemy,

I came across this page:

    http://alextechrants.blogspot.fi/2013/08/unit-testing-sqlalchemy-apps.html

This is definitely what I want to do, with one catch: I already have a session
management system in place for the library that seems to conflict with the

sample code given on the webpage, and I'm not having luck reconciling it.

I find the approach on that page a little awkward - it's using new globals for no good reason and also the begin_nested() seems strange.   The test itself then has a "self.session", so the test itself is using a test-bound session, the choice of globals for "connection" and "engine" seems even more weird.

Yeah, after looking at it a bit, it also seemed awkward to me, but the reason

I was looking is that it takes a good full 7 seconds to recreate an empty database

and doing that per test or even per module might get a bit painful (eventually

I suspect this library will have quite a few test modules to match all the tables
and uses of the library), but maybe I'll just pursue that for now (I mostly had

that working, at least).
 

My only concern with that is the enormous time to recreate/reset the
database for each test as mentioned above, but once again, I might
not be fully understanding if this is actually a concern or not.

I modified my test classes to subclass DatabaseTest, but an attempt
to run the tests results in:

UnboundExecutionError: Could not locate a bind configured on mapper Mapper|Environments|environments or this Session

make sure the Session is created with an explicit bind to the connection.

I have this at the end of my init_session():
 

Session.configure(bind=engine)

but I'm assuming that's not enough?

--
- Ken Lareau

[Michael Bayer]

On Oct 11, 2013, at 9:32 PM, Ken Lareau <kla...@tagged.com> wrote:

>
> I have this at the end of my init_session():
>
> Session.configure(bind=engine)
>
> but I'm assuming that's not enough?

I don't have a full code example in front of me but I'd imagine that the recipe you're using is injecting a non-bound session into your registry, so that the bind being configured above isn't getting used.