print "----------------------------"
objectstore.commit()
-
+#sys.exit()
print "\n\n\n----------------------------"
print "Tree After Save:"
print "----------------------------"
task.mapper.register_dependencies(self)
head = self._sort_dependencies()
- #print "Task dump:\n" + head.dump()
+ print "Task dump:\n" + head.dump()
if head is not None:
head.execute(self)
class UOWTask(object):
def __init__(self, uowtransaction, mapper):
- uowtransaction.tasks[mapper] = self
+ if uowtransaction is not None:
+ uowtransaction.tasks[mapper] = self
self.uowtransaction = uowtransaction
self.mapper = mapper
self.objects = util.OrderedDict()
"""executes this UOWTask. saves objects to be saved, processes all dependencies
that have been registered, and deletes objects to be deleted. """
if self.circular is not None:
+ print "CIRCULAR !"
self.circular.execute(trans)
+ print "CIRCULAR DONE !"
return
+ print "task " + str(self) + " tosave: " + repr(self.tosave_objects())
self.mapper.save_obj(self.tosave_objects(), trans)
for dep in self.save_dependencies():
(processor, targettask, isdelete) = dep
processor.process_dependencies(targettask, [elem.obj for elem in targettask.tosave_elements()], trans, delete = False)
+ print "processed dependencies on " + repr([elem.obj for elem in targettask.tosave_elements()])
for element in self.tosave_elements():
if element.childtask is not None:
+ print "execute elem childtask " + str(element.childtask)
element.childtask.execute(trans)
for dep in self.delete_dependencies():
(processor, targettask, isdelete) = dep
try:
return objecttotask[obj]
except KeyError:
- t = UOWTask(trans, self.mapper)
+ t = UOWTask(None, self.mapper)
objecttotask[obj] = t
return t
try:
l = dp[(processor, isdelete)]
except KeyError:
- l = UOWTask(trans, None)
+ l = UOWTask(None, None)
dp[(processor, isdelete)] = l
return l
t2 = make_task_tree(n, t)
return t
- t = UOWTask(trans, self.mapper)
+ t = UOWTask(None, self.mapper)
make_task_tree(head, t)
return t
return s
saveobj = self.tosave_elements()
if len(saveobj) > 0:
- s += "\n" + indent + " Save Elements:"
+ s += "\n" + indent + " Save Elements:(%d)" % len(saveobj)
for o in saveobj:
if not o.listonly:
s += "\n " + indent + repr(o)
dict(user_id = 8, user_name = 'ed'),
dict(user_id = 9, user_name = 'fred')
)
- db.connection().commit()
+ db.commit()
+ # mapper with just users table
User.mapper = assignmapper(users)
User.mapper.select()
+ oldmapper = User.mapper
+ # now a mapper with the users table plus a relation to the addresses
User.mapper = assignmapper(users, properties = dict(
addresses = relation(Address, addresses, lazy = False)
))
+ self.assert_(oldmapper is not User.mapper)
u = User.mapper.select()
u[0].addresses.append(Address())
u[0].addresses[0].email_address='hi'
+
+ # insure that upon commit, the new mapper with the address relation is used
self.assert_sql(db, lambda: objectstore.commit(),
[
(
echo = True
class PersistTest(unittest.TestCase):
+ """persist base class, provides default setUpAll, tearDownAll and echo functionality"""
def __init__(self, *args, **params):
unittest.TestCase.__init__(self, *args, **params)
def echo(self, text):
pass
class AssertMixin(PersistTest):
+ """given a list-based structure of keys/properties which represent information within an object structure, and
+ a list of actual objects, asserts that the list of objects corresponds to the structure."""
def assert_result(self, result, class_, *objects):
if echo:
print repr(result)
db.set_assert_list(None, None)
class EngineAssert(object):
+ """decorates a SQLEngine object to match the incoming queries against a set of assertions."""
def __init__(self, engine):
self.engine = engine
self.realexec = engine.execute
class TTestSuite(unittest.TestSuite):
- def __init__(self, tests=()):
- if len(tests) >0 and isinstance(tests[0], PersistTest):
- self._initTest = tests[0]
- else:
- self._initTest = None
- unittest.TestSuite.__init__(self, tests)
+ """override unittest.TestSuite to provide per-TestCase class setUpAll() and tearDownAll() functionality"""
+ def __init__(self, tests=()):
+ if len(tests) >0 and isinstance(tests[0], PersistTest):
+ self._initTest = tests[0]
+ else:
+ self._initTest = None
+ unittest.TestSuite.__init__(self, tests)
- def run(self, result):
+ def run(self, result):
+ try:
+ if self._initTest is not None:
+ self._initTest.setUpAll()
+ except:
+ result.addError(self._initTest, self.__exc_info())
+ pass
+ try:
+ return unittest.TestSuite.run(self, result)
+ finally:
try:
if self._initTest is not None:
- self._initTest.setUpAll()
+ self._initTest.tearDownAll()
except:
result.addError(self._initTest, self.__exc_info())
pass
- try:
- return unittest.TestSuite.run(self, result)
- finally:
- try:
- if self._initTest is not None:
- self._initTest.tearDownAll()
- except:
- result.addError(self._initTest, self.__exc_info())
- pass
- def __exc_info(self):
- """Return a version of sys.exc_info() with the traceback frame
- minimised; usually the top level of the traceback frame is not
- needed.
- ripped off out of unittest module since its double __
- """
- exctype, excvalue, tb = sys.exc_info()
- if sys.platform[:4] == 'java': ## tracebacks look different in Jython
- return (exctype, excvalue, tb)
+ def __exc_info(self):
+ """Return a version of sys.exc_info() with the traceback frame
+ minimised; usually the top level of the traceback frame is not
+ needed.
+ ripped off out of unittest module since its double __
+ """
+ exctype, excvalue, tb = sys.exc_info()
+ if sys.platform[:4] == 'java': ## tracebacks look different in Jython
return (exctype, excvalue, tb)
+ return (exctype, excvalue, tb)
unittest.TestLoader.suiteClass = TTestSuite
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
