Source Code for Module elisa.core.tests.test_defer

  1  # -*- coding: utf-8 -*- 
  2  # Moovida - Home multimedia server 
  3  # Copyright (C) 2006-2009 Fluendo Embedded S.L. (www.fluendo.com). 
  4  # All rights reserved. 
  5  # 
  6  # This file is available under one of two license agreements. 
  7  # 
  8  # This file is licensed under the GPL version 3. 
  9  # See "LICENSE.GPL" in the root of this distribution including a special 
 10  # exception to use Moovida with Fluendo's plugins. 
 11  # 
 12  # The GPL part of Moovida is also available under a commercial licensing 
 13  # agreement from Fluendo. 
 14  # See "LICENSE.Moovida" in the root directory of this distribution package 
 15  # for details on that license. 
 16   
 17   
 18  # this both should work without problems at all 
 19   
 20  # things to test: 
 21  from elisa.core.utils import defer 
 22  from elisa.core.utils.defer import Deferred 
 23  from twisted.internet import defer as twisted_defer 
 24   
 25  from twisted.trial import unittest 
 26   
 27 -class DeferFunctionTesting(unittest.TestCase): 
 28   
 29 -    def testDeffered(self): 
 30          # check if the importet Deferred has the cancel-API 
 31          self.failUnless(hasattr(Deferred(), 'cancel')) 
 32          self.failUnless(hasattr(defer.Deferred(), 'cancel')) 
 33   
 34 -    def testSucceed(self): 
 35          # test if the returned Deferred has the cancel-API 
 36          l = [] 
 37          dfr = defer.succeed('Yeah!') 
 38          self.failUnless(hasattr(dfr, 'cancel')) 
 39          dfr.addCallback(l.append) 
 40          self.assertEquals(l[0], "Yeah!") 
 41   
 42 -    def testFail(self): 
 43          # test if the returned Deferred has the cancel-API 
 44          l = [] 
 45          dfr = defer.fail(Exception()) 
 46          self.failUnless(hasattr(dfr, 'cancel')) 
 47          dfr.addErrback(l.append) 
 48          self.assertEquals(l[0].type, Exception) 
 49   
 50          def error_function(): 
 51              something 
 52   
 53          dfr = None 
 54   
 55          try:  
 56              error_function() 
 57          except: 
 58              dfr = defer.fail() 
 59   
 60          self.failUnless(hasattr(dfr, 'cancel')) 
 61          dfr.addErrback(l.append) 
 62          self.assertEquals(l[1].type, NameError) 
 63   
 64 -    def testExecute(self): 
 65   
 66          l = [] 
 67          # works out 
 68          dfr = defer.execute( lambda x,y,z,foo: (x,y,z,foo) ,  
 69                               'one', 'two', 'three', foo='bar') 
 70          self.failUnless(hasattr(dfr, 'cancel')) 
 71          dfr.addCallback(l.append) 
 72          self.assertEquals(l[0], ('one', 'two', 'three', 'bar')) 
 73   
 74          # fails 
 75          dfr = defer.execute( lambda x: y, 
 76                               'yeah', 'yeah', 'yeah', cool=True) 
 77          self.failUnless(hasattr(dfr, 'cancel')) 
 78          dfr.addErrback(l.append) 
 79          self.assertEquals(len(l), 2) 
 80   
 81   
 82 -    def testTimeout(self): 
 83   
 84          l = [] 
 85          dfr = defer.Deferred(canceller=lambda x: x) 
 86          dfr.addErrback(l.append) 
 87          defer.timeout(dfr) 
 88          # we got the right errback 
 89          self.assertEquals(l[0].type, defer.CancelledError) 
 90   
 91          # test the compability 
 92          dfr = twisted_defer.Deferred() 
 93          dfr.addErrback(l.append) 
 94          defer.timeout(dfr) 
 95          self.assertEquals(l[1].type, defer.TimeoutError) 
 96   
 97 -    def testCompability(self): 
 98          # test that all given classes and methods of twisted defer are in our 
 99          # defer 
100          msg = "We miss %s"  
101          for item in twisted_defer.__all__: 
102              self.assertTrue(hasattr(defer, item), msg % item) 
103   
104   
105 -    def testAccumulator(self): 
106          """ 
107          Test the behaviour of deferred_accumulator in various cases. 
108          """ 
109          dfr1 = defer.succeed(None) 
110          dfr2 = defer.succeed(None) 
111          dfr3 = defer.succeed(None) 
112   
113          tests = [ 
114                  ([None, None, None], (True, [])), 
115                  ([None, None, 3   ], (True, [])), 
116                  ([None, None, dfr1], (True, [dfr1])), 
117                  ([None, [],   None], (True, [])), 
118                  ([None, [],   3   ], (True, [])), 
119                  ([None, [],   dfr1], (True, [dfr1])), 
120                  ([None, [dfr1], None], (True, [dfr1])), 
121                  ([None, [dfr1], 3   ], (True, [dfr1])), 
122                  ([None, [dfr1], dfr2], (True, [dfr1, dfr2])), 
123                  ] 
124          for (args, result) in tests: 
125              self.assertEquals(defer.deferred_accumulator(*args), result) 
126