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()
