I've been struggling with this for a few days now, and I've tried a whole
slew of different approaches, but I'm just missing something.
I started with the example
at
http://docs.sqlalchemy.org/en/latest/orm/session_transaction.html#joining-a-session-into-an-external-transaction-such-as-for-test-suites
and have been trying variations on it to suit my goals.
We have a webapp which, in simplest form, could be represented as this:
# in the db setup file
dsn = 'mysql://...t'
engine = create_engine(dsn)
session_factory = sessionmaker(bind=engine)
scoped = scoped_session(session_factory)
# the app
class Webapp(object):
def dispatch(self, name):
view = getattr(self, name)
session = scoped()
try:
rsp = view()
session.commit()
return rsp
except:
session.rollback()
return 'error'
finally:
session.close()
pass
# the views files
def add_user(self, session):
user = User(username='test')
session = scoped()
session.add(user)
session.flush()
return 'user %s' % user.id
The app uses commit, rollback, and close on a scoped session.
I'd like to know how to adjust the example at the above URL so the test is
able to "enclose" the webapp, so the calls to sessionmaker, and all
operation that occur within the webapp, are within the scope of the test
transaction, so everything that happens in the app can be rolled back at
the end of the test. The example at the url acquires a connection, and
binds the session to the connection, and I'm not sure how to force the
sessionmaker to work with that. Could 'scopefunc' be used to somehow force
the returned session to be within the context of the transaction?
There is also fixture loading code, which starts by acquiring a session
from the scoped_session, adding fixtures, then committing. A simple example
could be this function:
def load_fixture(username):
session = scoped()
session.add(User(username=username))
session.commit()
I'd like to use this in unittests, and have it rolled back along with the
transaction.
I attached a full (nonworking) example, can you tell me what I am doing
wrong? Is this even possible with a scope session, or is this sort of
testing limited to to sessions bound to connections? Would I need to
rewrite the session acquisition method to return a globally stored
connection-bound session before defaulting to the scoped (engine-bound)
session?
An example I saw
at
https://web.archive.org/web/20140419235219/http://sontek.net/blog/detail/writing-tests-for-pyramid-and-sqlalchemy
seems to indicate that person overwrote the app session from within the
unittest, to ensure the app used that session. Is that the approach I would
need to take?
--
You received this message because you are subscribed to the Google Groups
"sqlalchemy" group.
To unsubscribe from this group and stop receiving emails from it, send an email
to sqlalchemy+unsubscr...@googlegroups.com.
To post to this group, send email to sqlalchemy@googlegroups.com.
Visit this group at https://groups.google.com/group/sqlalchemy.
For more options, visit https://groups.google.com/d/optout.
from sqlalchemy import *
from sqlalchemy import event
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import scoped_session
from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
import unittest
dsn = 'mysql://...'
engine = create_engine(dsn)
session_factory = sessionmaker(bind=engine)
scoped = scoped_session(session_factory)
Base = declarative_base()
class User(Base):
__tablename__ = 'test_users'
username = Column(String(50), primary_key=True)
def __repr__(self):
return "'%s'" % self.username
class Webapp(object):
def dispatch(self, name):
# in the real app, views are spread out over the fs
# and are looked up via a cached url configuration,
# they are not attributes of the base app
view = getattr(self, name)
session = scoped()
try:
rsp = view()
session.commit()
return rsp
except:
session.rollback()
return 'error'
finally:
session.close()
pass
def add_user(self):
session = scoped()
user = User(username='app')
session.add(user)
session.flush()
return 'user %s' % user.username
def view_user(self):
session = scoped()
user = session.query(User).get('setup')
return 'user %s' % user.username
def delete_user(self):
session = scoped()
user = session.query(User).get('setup')
session.delete(user)
session.flush()
return 'deleted'
def load_fixture(username):
session = scoped()
session.add(User(username=username))
session.commit()
def setUpModule():
print 'setup module'
Base.metadata.bind = engine
Base.metadata.drop_all()
Base.metadata.create_all()
def tearDownModule():
print 'teardown module'
class SomeTest(unittest.TestCase):
@classmethod
def setUpClass(cls):
print 'setup class'
cls.app = Webapp()
cls.session = scoped()
cls.trans1 = cls.session.begin_nested()
load_fixture('setupclass')
print 'post-setupclass users:', cls.session.query(User).all()
@classmethod
def tearDownClass(cls):
print 'pre-teardown class users:', cls.session.query(User).all()
cls.trans1.rollback()
cls.session.close()
print 'post-teardown class users:', cls.session.query(User).all()
def setUp(self):
print '\npre-setup users:', self.session.query(User).all()
self.trans2 = self.session.begin_nested()
load_fixture('setup')
print 'post-setup users:', self.session.query(User).all()
def tearDown(self):
print 'pre-teardown users:', self.session.query(User).all()
self.trans2.rollback()
print 'post-teardown users:', self.session.query(User).all()
#def test_view(self):
# rsp = self.app.dispatch('view_user')
# print rsp
def test_add(self):
rsp = self.app.dispatch('add_user')
print rsp
#def test_delete(self):
# rsp = self.app.dispatch('delete_user')
# print rsp
if __name__ == "__main__":
unittest.main()
