How to accomplish setup/run-app/teardown with mapped classes and sessions?
Kumar McMillan
I would like to setup and teardown test data using mapped classes.
The problem is that those same mapped classes need to be used by the
application under test and in case there is an error, the teardown
still needs to run so that subsequent tests can setup more data. It
seemed like the setup/teardown could be accomplished with a privately
scoped session but what I see is that this "private" session collides
with that of the application. Here is a failing test case (also
attached) that illustrates exactly what I need to do (sorry it's a
little long). The reason it catches the IntegrityError is because
during testing any kind of error can happen and I need to teardown
data regardless. Should I give up and use insert statements and
engine objects for the setup/teardown? Or is there a way to make this
test case pass? I am using sqlalchemy 0.4.2p3
from sqlalchemy import *
from sqlalchemy.orm import scoped_session, sessionmaker, mapper
from sqlalchemy.exceptions import IntegrityError
PrivateSession = scoped_session(
sessionmaker(autoflush=False, transactional=True),
scopefunc=lambda:__name__) # a private scope
AppSession = scoped_session(
sessionmaker(autoflush=False, transactional=True))
dsn = 'sqlite:///:memory:'
def test_sa_scoping():
engine = create_engine(dsn)
metadata = MetaData()
sometable = Table('sometable', metadata,
Column('id', Integer, primary_key=True),
Column('keyname', String(30), unique=True))
class SomeObject(object):
pass
metadata.create_all(bind=engine)
PrivateSession.configure(bind=engine)
AppSession.configure(bind=engine)
mapper(SomeObject, sometable)
fixture_session = PrivateSession()
# create some data to test with :
so = SomeObject()
so.keyname = "some unique key name"
fixture_session.save(so)
fixture_session.flush()
app_session = AppSession()
so2 = SomeObject()
so2.keyname = "some unique key name"
app_session.save(so2)
try:
app_session.flush()
except IntegrityError:
# violated unique key
pass
app_session.close()
# after testing application code, I want to tear down
# test even if the app had an error :
assert so in fixture_session
fixture_session.delete(so)
fixture_session.flush()
rs = fixture_session.query(SomeObject).all()
assert rs == [], "unexpected: %s" % rs
if __name__ == '__main__':
test_sa_scoping()
Traceback (most recent call last):
File "test_sa_scoping.py", line 55, in ?
test_sa_scoping()
File "test_sa_scoping.py", line 50, in test_sa_scoping
fixture_session.flush()
File "/Users/kumar/env/sqlalchemy-exp/lib/python2.4/site-packages/SQLAlchemy-0.4.2p3-py2.4.egg/sqlalchemy/orm/session.py",
line 693, in flush
self.uow.flush(self, objects)
File "/Users/kumar/env/sqlalchemy-exp/lib/python2.4/site-packages/SQLAlchemy-0.4.2p3-py2.4.egg/sqlalchemy/orm/unitofwork.py",
line 215, in flush
flush_context.execute()
File "/Users/kumar/env/sqlalchemy-exp/lib/python2.4/site-packages/SQLAlchemy-0.4.2p3-py2.4.egg/sqlalchemy/orm/unitofwork.py",
line 437, in execute
UOWExecutor().execute(self, tasks)
File "/Users/kumar/env/sqlalchemy-exp/lib/python2.4/site-packages/SQLAlchemy-0.4.2p3-py2.4.egg/sqlalchemy/orm/unitofwork.py",
line 930, in execute
self.execute_delete_steps(trans, task)
File "/Users/kumar/env/sqlalchemy-exp/lib/python2.4/site-packages/SQLAlchemy-0.4.2p3-py2.4.egg/sqlalchemy/orm/unitofwork.py",
line 951, in execute_delete_steps
self.delete_objects(trans, task)
File "/Users/kumar/env/sqlalchemy-exp/lib/python2.4/site-packages/SQLAlchemy-0.4.2p3-py2.4.egg/sqlalchemy/orm/unitofwork.py",
line 936, in delete_objects
task.mapper._delete_obj(task.polymorphic_todelete_objects, trans)
File "/Users/kumar/env/sqlalchemy-exp/lib/python2.4/site-packages/SQLAlchemy-0.4.2p3-py2.4.egg/sqlalchemy/orm/mapper.py",
line 1219, in _delete_obj
raise exceptions.ConcurrentModificationError("Deleted rowcount %d
does not match number of objects deleted %d" % (c.rowcount,
len(del_objects)))
sqlalchemy.exceptions.ConcurrentModificationError: Deleted rowcount 0
does not match number of objects deleted 1
NOTE:
When I comment out the code that uses AppSession, this test passes.
Michael Bayer
On Jan 23, 2008, at 5:00 PM, Kumar McMillan wrote:
> Hello, I have not been able to figure this out from the docs.
>
> I would like to setup and teardown test data using mapped classes.
> The problem is that those same mapped classes need to be used by the
> application under test and in case there is an error, the teardown
> still needs to run so that subsequent tests can setup more data. It
> seemed like the setup/teardown could be accomplished with a privately
> scoped session but what I see is that this "private" session collides
> with that of the application.
your teardown code can't have any dependencies on the test code
itself. So at the very least start the teardown phase with
"PrivateSession.close()" so that you start fresh.
the two ways to create tests that use isolated data are to either
create and drop tables local to the unit tests themselves, or to run
the unit tests within an enclosing Transaction (as in, conn =
engine.connect(); trans = conn.begin(); session.bind=conn) which is
rolled back at the end of the unit tests. The SQLA unit tests
themselves use the former method but I have applied the latter method
to Pylons tests (and is also what you usually do with Java/Hibernate
unit tests).
Kumar McMillan
On Jan 23, 2008 4:07 PM, Michael Bayer <mik...@zzzcomputing.com> wrote:
> your teardown code can't have any dependencies on the test code
> itself. So at the very least start the teardown phase with
> "PrivateSession.close()" so that you start fresh.
I tried adding that to the teardown code but then this assertion fails:
assert so in fixture_session
and if I comment out the assertion, I get the same ConcurrentModificationError
without a stored reference to the object that was saved, I'm not sure
how to delete it [without monkeying with last inserted id].
>
> the two ways to create tests that use isolated data are to either
> create and drop tables local to the unit tests themselves, or to run
> the unit tests within an enclosing Transaction (as in, conn =
> engine.connect(); trans = conn.begin(); session.bind=conn) which is
> rolled back at the end of the unit tests. The SQLA unit tests
> themselves use the former method but I have applied the latter method
> to Pylons tests (and is also what you usually do with Java/Hibernate
> unit tests).
ok, I think I see what you're saying. Removing the
PrivateSession.close(), I tried implementing begin/rollback by
changing the app segment to:
conn = engine.connect()
AppSession.configure(bind=conn)
app_session = AppSession()
trans = conn.begin()
so2 = SomeObject()
so2.keyname = "some unique key name"
app_session.save(so2)
try:
app_session.flush()
except IntegrityError:
# violated unique key
trans.rollback()
else:
trans.commit()
app_session.close()
...but it still fails with the same error, Deleted rowcount 0 does not
match number of objects deleted 1. What am I missing? I don't
understand how the teardown code is dependent on the app code if it is
using a different session and a different connection (now) to save the
same mapped class instances.
Here is the altered test case:
from sqlalchemy import *
from sqlalchemy.orm import scoped_session, sessionmaker, mapper
from sqlalchemy.exceptions import IntegrityError
PrivateSession = scoped_session(
sessionmaker(autoflush=False, transactional=True),
scopefunc=lambda:__name__) # a private scope
AppSession = scoped_session(
sessionmaker(autoflush=False, transactional=True))
dsn = 'sqlite:///:memory:'
def test_sa_scoping():
engine = create_engine(dsn)
metadata = MetaData()
sometable = Table('sometable', metadata,
Column('id', Integer, primary_key=True),
Column('keyname', String(30), unique=True))
class SomeObject(object):
pass
metadata.create_all(bind=engine)
PrivateSession.configure(bind=engine)
mapper(SomeObject, sometable)
fixture_session = PrivateSession()
# create some data to test with :
so = SomeObject()
so.keyname = "some unique key name"
fixture_session.save(so)
fixture_session.flush()
conn = engine.connect()
AppSession.configure(bind=conn)
app_session = AppSession()
trans = conn.begin()
so2 = SomeObject()
so2.keyname = "some unique key name"
app_session.save(so2)
try:
app_session.flush()
except IntegrityError:
# violated unique key
trans.rollback()
else:
trans.commit()
app_session.close()
# after testing application code, I want to tear down
# test even if the app had an error :
assert so in fixture_session
fixture_session.delete(so)
fixture_session.flush()
rs = fixture_session.query(SomeObject).all()
assert rs == [], "unexpected: %s" % rs
if __name__ == '__main__':
test_sa_scoping()
same exception...
jason kirtland
> Hello, I have not been able to figure this out from the docs.
>
> I would like to setup and teardown test data using mapped classes.
> The problem is that those same mapped classes need to be used by the
> application under test and in case there is an error, the teardown
> still needs to run so that subsequent tests can setup more data. It
> seemed like the setup/teardown could be accomplished with a privately
> scoped session but what I see is that this "private" session collides
> with that of the application. Here is a failing test case (also
> attached) that illustrates exactly what I need to do (sorry it's a
> little long). The reason it catches the IntegrityError is because
> during testing any kind of error can happen and I need to teardown
> data regardless. Should I give up and use insert statements and
> engine objects for the setup/teardown? Or is there a way to make this
> test case pass? I am using sqlalchemy 0.4.2p3
You're stepping on yourself here with sqlite- you have two transactional
sessions, but an in-memory SQLite database only allows one connection
and it and it's underlying transactional state is being shared by both
sessions. Specifically, the app_session.close() is rolling back the
uncommited row flushed by fixture_session.flush().
This setup will work almost as-is when testing against a
multi-connection database- you just need to commit() your fixture data
after that flush() so that it will be visible to the "app" part of the test.
That will get this going for sqlite as well but you'd want to think
through what a single connection shared between setup/teardown and test
execution means to your tests.
Kumar McMillan
> ...but it still fails with the same error, Deleted rowcount 0 does not
> match number of objects deleted 1. What am I missing? I don't
> understand how the teardown code is dependent on the app code if it is
> using a different session and a different connection (now) to save the
> same mapped class instances.
ah, I just had to put the setup/teardown routines in their own
respective transactions too. Now it passes. Thanks! Next... to see
if I can clean it up a bit and fit it into my app.
Passing test:
from sqlalchemy import *
from sqlalchemy.orm import scoped_session, sessionmaker, mapper
from sqlalchemy.exceptions import IntegrityError
PrivateSession = scoped_session(
sessionmaker(autoflush=False, transactional=True),
scopefunc=lambda:__name__) # a private scope
AppSession = scoped_session(
sessionmaker(autoflush=False, transactional=True))
dsn = 'sqlite:///:memory:'
def test_sa_scoping():
engine = create_engine(dsn)
metadata = MetaData()
sometable = Table('sometable', metadata,
Column('id', Integer, primary_key=True),
Column('keyname', String(30), unique=True))
class SomeObject(object):
pass
metadata.create_all(bind=engine)
mapper(SomeObject, sometable)
conn = engine.connect()
PrivateSession.configure(bind=conn)
trans = conn.begin()
fixture_session = PrivateSession()
# create some data to test with :
so = SomeObject()
so.keyname = "some unique key name"
fixture_session.save(so)
fixture_session.flush()
trans.commit()
conn = engine.connect()
AppSession.configure(bind=conn)
app_session = AppSession()
trans = conn.begin()
so2 = SomeObject()
so2.keyname = "some unique key name"
app_session.save(so2)
try:
app_session.flush()
except IntegrityError:
# violated unique key
trans.rollback()
else:
trans.commit()
app_session.close()
# after testing application code, I want to tear down
# test even if the app had an error :
assert so in fixture_session
trans = conn.begin()
fixture_session.delete(so)
fixture_session.flush()
trans.commit()
svilen
level u'll may do these:
- destroy all your refs to the mappers/tables etc
- sqlalchemy.orm.clearmappes()
- yourmetadata.drop_all()
- yourengine.dispose()
- destroy refs to yourmetadata/yourengine
(dbcook.usage.sa_manager.py/destroy())
if your classes are made-on-the-fly inside some function, then every
time they are _different_ classes. Make sure u do not hold them
somewhere - or they will keep all the stuff asociated with them
(mappers and even session). The reason is the replaced __init__
method is stored as class.__init (in orm.attributes.register_class)
and is not cleared in clearmappers.
for easy check, u can run your test 100 times and watch the memory
used; if it grows then _something_ of all those above is not cleared.
ciao
svilen