diff options
Diffstat (limited to 'test/orm/cascade.py')
| -rw-r--r-- | test/orm/cascade.py | 173 |
1 files changed, 173 insertions, 0 deletions
diff --git a/test/orm/cascade.py b/test/orm/cascade.py new file mode 100644 index 000000000..4a997d67f --- /dev/null +++ b/test/orm/cascade.py @@ -0,0 +1,173 @@ +import testbase, tables +import unittest, sys, datetime + +from sqlalchemy.ext.sessioncontext import SessionContext +from sqlalchemy import * + +class O2MCascadeTest(testbase.AssertMixin): + def tearDown(self): + ctx.current.clear() + tables.delete() + + def tearDownAll(self): + clear_mappers() + tables.drop() + + def setUpAll(self): + global ctx, data + ctx = SessionContext(lambda: create_session(echo_uow=True)) + tables.create() + mapper(tables.User, tables.users, properties = dict( + address = relation(mapper(tables.Address, tables.addresses), lazy = False, uselist = False, private = True), + orders = relation( + mapper(tables.Order, tables.orders, properties = dict ( + items = relation(mapper(tables.Item, tables.orderitems), lazy = False, uselist =True, private = True) + )), + lazy = True, uselist = True, private = True) + )) + + def setUp(self): + global data + data = [tables.User, + {'user_name' : 'ed', + 'address' : (tables.Address, {'email_address' : 'foo@bar.com'}), + 'orders' : (tables.Order, [ + {'description' : 'eds 1st order', 'items' : (tables.Item, [{'item_name' : 'eds o1 item'}, {'item_name' : 'eds other o1 item'}])}, + {'description' : 'eds 2nd order', 'items' : (tables.Item, [{'item_name' : 'eds o2 item'}, {'item_name' : 'eds other o2 item'}])} + ]) + }, + {'user_name' : 'jack', + 'address' : (tables.Address, {'email_address' : 'jack@jack.com'}), + 'orders' : (tables.Order, [ + {'description' : 'jacks 1st order', 'items' : (tables.Item, [{'item_name' : 'im a lumberjack'}, {'item_name' : 'and im ok'}])} + ]) + }, + {'user_name' : 'foo', + 'address' : (tables.Address, {'email_address': 'hi@lala.com'}), + 'orders' : (tables.Order, [ + {'description' : 'foo order', 'items' : (tables.Item, [])}, + {'description' : 'foo order 2', 'items' : (tables.Item, [{'item_name' : 'hi'}])}, + {'description' : 'foo order three', 'items' : (tables.Item, [{'item_name' : 'there'}])} + ]) + } + ] + + for elem in data[1:]: + u = tables.User() + ctx.current.save(u) + u.user_name = elem['user_name'] + u.address = tables.Address() + u.address.email_address = elem['address'][1]['email_address'] + u.orders = [] + for order in elem['orders'][1]: + o = tables.Order() + o.isopen = None + o.description = order['description'] + u.orders.append(o) + o.items = [] + for item in order['items'][1]: + i = tables.Item() + i.item_name = item['item_name'] + o.items.append(i) + + ctx.current.flush() + ctx.current.clear() + + + def testdelete(self): + l = ctx.current.query(tables.User).select() + for u in l: + self.echo( repr(u.orders)) + self.assert_result(l, data[0], *data[1:]) + + self.echo("\n\n\n") + ids = (l[0].user_id, l[2].user_id) + ctx.current.delete(l[0]) + ctx.current.delete(l[2]) + + ctx.current.flush() + self.assert_(tables.orders.count(tables.orders.c.user_id.in_(*ids)).scalar() == 0) + self.assert_(tables.orderitems.count(tables.orders.c.user_id.in_(*ids) &(tables.orderitems.c.order_id==tables.orders.c.order_id)).scalar() == 0) + self.assert_(tables.addresses.count(tables.addresses.c.user_id.in_(*ids)).scalar() == 0) + self.assert_(tables.users.count(tables.users.c.user_id.in_(*ids)).scalar() == 0) + + + def testorphan(self): + l = ctx.current.query(tables.User).select() + jack = l[1] + jack.orders[:] = [] + + ids = [jack.user_id] + self.assert_(tables.orders.count(tables.orders.c.user_id.in_(*ids)).scalar() == 1) + self.assert_(tables.orderitems.count(tables.orders.c.user_id.in_(*ids) &(tables.orderitems.c.order_id==tables.orders.c.order_id)).scalar() == 2) + + ctx.current.flush() + + self.assert_(tables.orders.count(tables.orders.c.user_id.in_(*ids)).scalar() == 0) + self.assert_(tables.orderitems.count(tables.orders.c.user_id.in_(*ids) &(tables.orderitems.c.order_id==tables.orders.c.order_id)).scalar() == 0) + + +class M2OCascadeTest(testbase.AssertMixin): + def tearDown(self): + ctx.current.clear() + for t in metadata.table_iterator(reverse=True): + t.delete().execute() + + def tearDownAll(self): + clear_mappers() + metadata.drop_all() + + def setUpAll(self): + global ctx, data, metadata, User, Pref + ctx = SessionContext(create_session) + metadata = BoundMetaData(testbase.db) + prefs = Table('prefs', metadata, + Column('prefs_id', Integer, Sequence('prefs_id_seq', optional=True), primary_key=True), + Column('prefs_data', String(40))) + + users = Table('users', metadata, + Column('user_id', Integer, Sequence('user_id_seq', optional=True), primary_key = True), + Column('user_name', String(40)), + Column('pref_id', Integer, ForeignKey('prefs.prefs_id')) + ) + class User(object): + pass + class Pref(object): + pass + metadata.create_all() + mapper(User, users, properties = dict( + pref = relation(mapper(Pref, prefs), lazy=False, cascade="all, delete-orphan") + )) + + def setUp(self): + global data + data = [User, + {'user_name' : 'ed', + 'pref' : (Pref, {'prefs_data' : 'pref 1'}), + }, + {'user_name' : 'jack', + 'pref' : (Pref, {'prefs_data' : 'pref 2'}), + }, + {'user_name' : 'foo', + 'pref' : (Pref, {'prefs_data' : 'pref 3'}), + } + ] + + for elem in data[1:]: + u = User() + ctx.current.save(u) + u.user_name = elem['user_name'] + u.pref = Pref() + u.pref.prefs_data = elem['pref'][1]['prefs_data'] + + ctx.current.flush() + ctx.current.clear() + + def testorphan(self): + l = ctx.current.query(User).select() + jack = l[1] + jack.pref = None + ctx.current.flush() + +if __name__ == "__main__": + testbase.main() |