Return to testProxyTM.py CVS log

Up to [Repository] / ExtFile / tests

Annotation of ExtFile/tests/testProxyTM.py, revision 1.1.1.1

1.1       dwinter     1: #
                      2: # Tests TMRegistry and ProxyTM
                      3: #
                      4: 
                      5: import os, sys
                      6: if __name__ == '__main__':
                      7:     execfile(os.path.join(sys.path[0], 'framework.py'))
                      8: 
                      9: from Testing import ZopeTestCase
                     10: 
                     11: from OFS.SimpleItem import SimpleItem
                     12: from Products.ExtFile import transaction
                     13: from Products.ExtFile.TM import TMRegistry, registry
                     14: from Products.ExtFile.TM import ProxyTM
                     15: 
                     16: 
                     17: class Target(SimpleItem):
                     18:     begin_called = 0
                     19:     finish_called = 0
                     20:     abort_called = 0
                     21:     def _begin(self):
                     22:         self.begin_called = 1
                     23:     def _finish(self):
                     24:         self.finish_called = 1
                     25:     def _abort(self):
                     26:         self.abort_called = 1
                     27: 
                     28: def tr():
                     29:     t = transaction.get()
                     30:     if hasattr(t, '_objects'):
                     31:         return t._objects   # ZODB <= 3.2
                     32:     elif hasattr(t, '_resources'):
                     33:         return t._resources # ZODB >= 3.4
                     34: 
                     35: 
                     36: class TestTMRegistry(ZopeTestCase.TestCase):
                     37: 
                     38:     def afterSetUp(self):
                     39:         self.reg = TMRegistry()
                     40: 
                     41:     def testRegister(self):
                     42:         self.assertEqual(len(self.reg), 0)
                     43:         self.reg.register(Target())
                     44:         self.assertEqual(len(self.reg), 1)
                     45:         self.reg.register(Target())
                     46:         self.assertEqual(len(self.reg), 2)
                     47: 
                     48:     def testCount(self):
                     49:         self.assertEqual(self.reg.count(), 0)
                     50:         self.reg.register(Target())
                     51:         self.assertEqual(self.reg.count(), 1)
                     52:         self.reg.register(Target())
                     53:         self.assertEqual(self.reg.count(), 2)
                     54: 
                     55:     def testContains(self):
                     56:         target1, target2 = Target(), Target()
                     57:         self.reg.register(target1)
                     58:         self.reg.register(target2)
                     59:         self.failUnless(self.reg.contains(target1))
                     60:         self.failUnless(self.reg.contains(target2))
                     61: 
                     62:     def testRemove(self):
                     63:         target1, target2 = Target(), Target()
                     64:         self.reg.register(target1)
                     65:         self.reg.register(target2)
                     66:         self.reg.remove(target1)
                     67:         self.reg.remove(target2)
                     68:         self.failIf(self.reg.contains(target1))
                     69:         self.failIf(self.reg.contains(target2))
                     70: 
                     71:     def testGet(self):
                     72:         target1 = Target()
                     73:         self.reg.register(target1)
                     74:         tm = self.reg.get(target1)
                     75:         self.failUnless(isinstance(tm, ProxyTM))
                     76:         self.assertEqual(target1, tm._target)
                     77: 
                     78: 
                     79: class TestProxyTM(ZopeTestCase.Sandboxed, ZopeTestCase.TestCase):
                     80: 
                     81:     def testRegister(self):
                     82:         tm = ProxyTM(Target())
                     83:         self.assertEqual(len(tr()), 0)
                     84:         tm._register()
                     85:         self.assertEqual(len(tr()), 1)
                     86:         try:
                     87:             self.assertEqual(tr()[0].manager, tm)
                     88:         except AttributeError:
                     89:             pass # ZODB <= 3.2
                     90: 
                     91:     def testLastRegisteredComesFirst(self):
                     92:         tm1, tm2 = ProxyTM(Target()), ProxyTM(Target())
                     93:         tm1._register()
                     94:         tm2._register()
                     95:         self.assertEqual(len(tr()), 2)
                     96:         # Now make sure that tm2 comes first in the
                     97:         # transaction's _resources list
                     98:         try:
                     99:             self.assertEqual(tr()[0].manager, tm2)
                    100:             self.assertEqual(tr()[1].manager, tm1)
                    101:         except AttributeError:
                    102:             pass # ZODB <= 3.2
                    103: 
                    104:     def testBeginIsForwarded(self):
                    105:         target = Target()
                    106:         registry.register(target)
                    107:         tm = registry.get(target)
                    108:         tm._begin()
                    109:         self.failUnless(target.begin_called)
                    110: 
                    111:     def testFinishIsForwarded(self):
                    112:         target = Target()
                    113:         registry.register(target)
                    114:         tm = registry.get(target)
                    115:         tm._finish()
                    116:         self.failUnless(target.finish_called)
                    117:         # _finish removes the TM from the registry
                    118:         self.failIf(registry.contains(target))
                    119: 
                    120:     def testAbortIsForwarded(self):
                    121:         target = Target()
                    122:         registry.register(target)
                    123:         tm = registry.get(target)
                    124:         tm._abort()
                    125:         self.failUnless(target.abort_called)
                    126:         # _abort removes the TM from the registry
                    127:         self.failIf(registry.contains(target))
                    128: 
                    129:     def testCommitCallsFinish(self):
                    130:         target = Target()
                    131:         registry.register(target)
                    132:         transaction.commit()
                    133:         self.failUnless(target.finish_called)
                    134:         # _finish removes the TM from the registry
                    135:         self.failIf(registry.contains(target))
                    136: 
                    137:     def testAbortCallsAbort(self):
                    138:         target = Target()
                    139:         registry.register(target)
                    140:         transaction.abort()
                    141:         self.failUnless(target.abort_called)
                    142:         # _abort removes the TM from the registry
                    143:         self.failIf(registry.contains(target))
                    144: 
                    145: 
                    146: def test_suite():
                    147:     from unittest import TestSuite, makeSuite
                    148:     suite = TestSuite()
                    149:     suite.addTest(makeSuite(TestTMRegistry))
                    150:     suite.addTest(makeSuite(TestProxyTM))
                    151:     return suite
                    152: 
                    153: if __name__ == '__main__':
                    154:     framework()
                    155: