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
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. --~--~---------~--~----~------------~-------~--~----~ You received this message because you are subscribed to the Google Groups "sqlalchemy" group. To post to this group, send email to sqlalchemy@googlegroups.com To unsubscribe from this group, send email to [EMAIL PROTECTED] For more options, visit this group at http://groups.google.com/group/sqlalchemy?hl=en -~----------~----~----~----~------~----~------~--~---
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()