Update of /cvsroot/sqlobject/SQLObject/tests
In directory sc8-pr-cvs1:/tmp/cvs-serv16262/tests
Modified Files:
SQLObjectTest.py test.py
Log Message:
Merge cascadegeddon-branch. This implements foreign key constraints (postgresql-only) and 'DELETE CASCADE'/'DELETE RESTRICT' (should work on all db's as it's done manually). We don't support 'UPDATE CASCADE'/'UPDATE RESTRICT' yet. To get it working, just pass the 'cascade' keyword argument to ForeignKey. True=CASCADE, False=RESTRICT, None(default)=The old behavior.
Index: SQLObjectTest.py
===================================================================
RCS file: /cvsroot/sqlobject/SQLObject/tests/SQLObjectTest.py,v
retrieving revision 1.19
retrieving revision 1.20
diff -C2 -d -r1.19 -r1.20
*** SQLObjectTest.py 4 Dec 2003 16:36:17 -0000 1.19
--- SQLObjectTest.py 4 Dec 2003 16:44:05 -0000 1.20
***************
*** 118,122 ****
__connection__.query(c.drop)
elif hasattr(c, 'dropTable'):
! c.dropTable(ifExists=True)
if hasattr(c, '%sCreate' % self.databaseName):
--- 118,122 ----
__connection__.query(c.drop)
elif hasattr(c, 'dropTable'):
! c.dropTable(ifExists=True, cascade=True)
if hasattr(c, '%sCreate' % self.databaseName):
Index: test.py
===================================================================
RCS file: /cvsroot/sqlobject/SQLObject/tests/test.py,v
retrieving revision 1.35
retrieving revision 1.36
diff -C2 -d -r1.35 -r1.36
*** test.py 4 Dec 2003 16:36:17 -0000 1.35
--- test.py 4 Dec 2003 16:44:05 -0000 1.36
***************
*** 120,123 ****
--- 120,241 ----
self.assertEqual(tcc2.other, tc4a)
+ class TestSO5(SQLObject):
+ name = StringCol(length=10, dbName='name_col')
+ other = ForeignKey('TestSO6', default=None, cascade=True)
+ another = ForeignKey('TestSO7', default=None, cascade=True)
+
+ class TestSO6(SQLObject):
+ name = StringCol(length=10, dbName='name_col')
+ other = ForeignKey('TestSO7', default=None, cascade=True)
+
+ class TestSO7(SQLObject):
+ name = StringCol(length=10, dbName='name_col')
+
+ class TestCase567(SQLObjectTest):
+
+ classes = [TestSO7, TestSO6, TestSO5]
+
+ def testForeignKeyDestroySelfCascade(self):
+ tc5 = TestSO5.new(name='a')
+ tc6a = TestSO6.new(name='1')
+ tc5.other = tc6a
+ tc7a = TestSO7.new(name='2')
+ tc6a.other = tc7a
+ tc5.another = tc7a
+ self.assertEqual(tc5.other, tc6a)
+ self.assertEqual(tc5.otherID, tc6a.id)
+ self.assertEqual(tc6a.other, tc7a)
+ self.assertEqual(tc6a.otherID, tc7a.id)
+ self.assertEqual(tc5.other.other, tc7a)
+ self.assertEqual(tc5.other.otherID, tc7a.id)
+ self.assertEqual(tc5.another, tc7a)
+ self.assertEqual(tc5.anotherID, tc7a.id)
+ self.assertEqual(tc5.other.other, tc5.another)
+ self.assertEqual(TestSO5.select().count(), 1)
+ self.assertEqual(TestSO6.select().count(), 1)
+ self.assertEqual(TestSO7.select().count(), 1)
+ tc6b = TestSO6.new(name='3')
+ tc6c = TestSO6.new(name='4')
+ tc7b = TestSO7.new(name='5')
+ tc6b.other = tc7b
+ tc6c.other = tc7b
+ self.assertEqual(TestSO5.select().count(), 1)
+ self.assertEqual(TestSO6.select().count(), 3)
+ self.assertEqual(TestSO7.select().count(), 2)
+ tc6b.destroySelf()
+ self.assertEqual(TestSO5.select().count(), 1)
+ self.assertEqual(TestSO6.select().count(), 2)
+ self.assertEqual(TestSO7.select().count(), 2)
+ tc7b.destroySelf()
+ self.assertEqual(TestSO5.select().count(), 1)
+ self.assertEqual(TestSO6.select().count(), 1)
+ self.assertEqual(TestSO7.select().count(), 1)
+ tc7a.destroySelf()
+ self.assertEqual(TestSO5.select().count(), 0)
+ self.assertEqual(TestSO6.select().count(), 0)
+ self.assertEqual(TestSO7.select().count(), 0)
+
+ def testForeignKeyDropTableCascade(self):
+ tc5a = TestSO5.new(name='a')
+ tc6a = TestSO6.new(name='1')
+ tc5a.other = tc6a
+ tc7a = TestSO7.new(name='2')
+ tc6a.other = tc7a
+ tc5a.another = tc7a
+ tc5b = TestSO5.new(name='b')
+ tc5c = TestSO5.new(name='c')
+ tc6b = TestSO6.new(name='3')
+ tc5c.other = tc6b
+ self.assertEqual(TestSO5.select().count(), 3)
+ self.assertEqual(TestSO6.select().count(), 2)
+ self.assertEqual(TestSO7.select().count(), 1)
+ TestSO7.dropTable(cascade=True)
+ self.assertEqual(TestSO5.select().count(), 3)
+ self.assertEqual(TestSO6.select().count(), 2)
+ tc6a.destroySelf()
+ self.assertEqual(TestSO5.select().count(), 2)
+ self.assertEqual(TestSO6.select().count(), 1)
+ tc6b.destroySelf()
+ self.assertEqual(TestSO5.select().count(), 1)
+ self.assertEqual(TestSO6.select().count(), 0)
+ self.assertEqual(iter(TestSO5.select()).next(), tc5b)
+ tc6c = TestSO6.new(name='3')
+ tc5b.other = tc6c
+ self.assertEqual(TestSO5.select().count(), 1)
+ self.assertEqual(TestSO6.select().count(), 1)
+ tc6c.destroySelf()
+ self.assertEqual(TestSO5.select().count(), 0)
+ self.assertEqual(TestSO6.select().count(), 0)
+
+ class TestSO8(SQLObject):
+ name = StringCol(length=10, dbName='name_col')
+ other = ForeignKey('TestSO9', default=None, cascade=False)
+
+ class TestSO9(SQLObject):
+ name = StringCol(length=10, dbName='name_col')
+
+ class TestCase89(SQLObjectTest):
+
+ classes = [TestSO9, TestSO8]
+
+ def testForeignKeyDestroySelfRestrict(self):
+ tc8a = TestSO8.new(name='a')
+ tc9a = TestSO9.new(name='1')
+ tc8a.other = tc9a
+ tc8b = TestSO8.new(name='b')
+ tc9b = TestSO9.new(name='2')
+ self.assertEqual(tc8a.other, tc9a)
+ self.assertEqual(tc8a.otherID, tc9a.id)
+ self.assertEqual(TestSO8.select().count(), 2)
+ self.assertEqual(TestSO9.select().count(), 2)
+ self.assertRaises(Exception, tc9a.destroySelf)
+ tc9b.destroySelf()
+ self.assertEqual(TestSO8.select().count(), 2)
+ self.assertEqual(TestSO9.select().count(), 1)
+ tc8a.destroySelf()
+ tc8b.destroySelf()
+ tc9a.destroySelf()
+ self.assertEqual(TestSO8.select().count(), 0)
+ self.assertEqual(TestSO9.select().count(), 0)
########################################
|