IPython/kernel/tests/test_pendingdeferred.py

#!/usr/bin/env python
# encoding: utf-8

"""Tests for pendingdeferred.py"""

__docformat__ = "restructuredtext en"

#-------------------------------------------------------------------------------
#  Copyright (C) 2008  The IPython Development Team
#
#  Distributed under the terms of the BSD License.  The full license is in
#  the file COPYING, distributed as part of this software.
#-------------------------------------------------------------------------------

#-------------------------------------------------------------------------------
# Imports
#-------------------------------------------------------------------------------

try:
    from twisted.internet import defer
    from twisted.python import failure

    from IPython.testing import tcommon
    from IPython.testing.tcommon import *
    from IPython.testing.util import DeferredTestCase
    import IPython.kernel.pendingdeferred as pd
    from IPython.kernel import error
    from IPython.kernel.util import printer
except ImportError:
    pass
else:

    #-------------------------------------------------------------------------------
    # Setup for inline and standalone doctests
    #-------------------------------------------------------------------------------


    # If you have standalone doctests in a separate file, set their names in the
    # dt_files variable (as a single string  or a list thereof):
    dt_files = []

    # If you have any modules whose docstrings should be scanned for embedded tests
    # as examples accorging to standard doctest practice, set them here (as a
    # single string or a list thereof):
    dt_modules = []

    #-------------------------------------------------------------------------------
    # Regular Unittests
    #-------------------------------------------------------------------------------


    class Foo(object):
    
        def bar(self, bahz):
            return defer.succeed('blahblah: %s' % bahz)

    class TwoPhaseFoo(pd.PendingDeferredManager):
    
        def __init__(self, foo):
            self.foo = foo
            pd.PendingDeferredManager.__init__(self)

        @pd.two_phase
        def bar(self, bahz):
            return self.foo.bar(bahz)

    class PendingDeferredManagerTest(DeferredTestCase):
    
        def setUp(self):
            self.pdm = pd.PendingDeferredManager()
        
        def tearDown(self):
            pass
    
        def testBasic(self):
            dDict = {}
            # Create 10 deferreds and save them
            for i in range(10):
                d = defer.Deferred()
                did = self.pdm.save_pending_deferred(d)
                dDict[did] = d
            # Make sure they are begin saved
            for k in dDict.keys():
                self.assert_(self.pdm.quick_has_id(k))
            # Get the pending deferred (block=True), then callback with 'foo' and compare
            for did in dDict.keys()[0:5]:
                d = self.pdm.get_pending_deferred(did,block=True)
                dDict[did].callback('foo')
                d.addCallback(lambda r: self.assert_(r=='foo'))
            # Get the pending deferreds with (block=False) and make sure ResultNotCompleted is raised
            for did in dDict.keys()[5:10]:
                d = self.pdm.get_pending_deferred(did,block=False)
                d.addErrback(lambda f: self.assertRaises(error.ResultNotCompleted, f.raiseException))
            # Now callback the last 5, get them and compare.
            for did in dDict.keys()[5:10]:
                dDict[did].callback('foo')
                d = self.pdm.get_pending_deferred(did,block=False)
                d.addCallback(lambda r: self.assert_(r=='foo'))
    
        def test_save_then_delete(self):
            d = defer.Deferred()
            did = self.pdm.save_pending_deferred(d)
            self.assert_(self.pdm.quick_has_id(did))
            self.pdm.delete_pending_deferred(did)
            self.assert_(not self.pdm.quick_has_id(did))
    
        def test_save_get_delete(self):
            d = defer.Deferred()
            did = self.pdm.save_pending_deferred(d)
            d2 = self.pdm.get_pending_deferred(did,True)
            d2.addErrback(lambda f: self.assertRaises(error.AbortedPendingDeferredError, f.raiseException))
            self.pdm.delete_pending_deferred(did)
            return d2
   
        def test_double_get(self):
            d = defer.Deferred()
            did = self.pdm.save_pending_deferred(d)
            d2 = self.pdm.get_pending_deferred(did,True)
            d3 = self.pdm.get_pending_deferred(did,True)
            d3.addErrback(lambda f: self.assertRaises(error.InvalidDeferredID, f.raiseException))
    
        def test_get_after_callback(self):
            d = defer.Deferred()
            did = self.pdm.save_pending_deferred(d)
            d.callback('foo')
            d2 = self.pdm.get_pending_deferred(did,True)
            d2.addCallback(lambda r: self.assertEquals(r,'foo'))
            self.assert_(not self.pdm.quick_has_id(did))

        def test_get_before_callback(self):
            d = defer.Deferred()
            did = self.pdm.save_pending_deferred(d)
            d2 = self.pdm.get_pending_deferred(did,True)
            d.callback('foo')
            d2.addCallback(lambda r: self.assertEquals(r,'foo'))
            self.assert_(not self.pdm.quick_has_id(did))
            d = defer.Deferred()
            did = self.pdm.save_pending_deferred(d)
            d2 = self.pdm.get_pending_deferred(did,True)
            d2.addCallback(lambda r: self.assertEquals(r,'foo'))
            d.callback('foo')
            self.assert_(not self.pdm.quick_has_id(did))
    
        def test_get_after_errback(self):
            class MyError(Exception):
                pass
            d = defer.Deferred()
            did = self.pdm.save_pending_deferred(d)
            d.errback(failure.Failure(MyError('foo')))
            d2 = self.pdm.get_pending_deferred(did,True)
            d2.addErrback(lambda f: self.assertRaises(MyError, f.raiseException))
            self.assert_(not self.pdm.quick_has_id(did))

        def test_get_before_errback(self):
            class MyError(Exception):
                pass
            d = defer.Deferred()
            did = self.pdm.save_pending_deferred(d)
            d2 = self.pdm.get_pending_deferred(did,True)
            d.errback(failure.Failure(MyError('foo')))
            d2.addErrback(lambda f: self.assertRaises(MyError, f.raiseException))
            self.assert_(not self.pdm.quick_has_id(did))
            d = defer.Deferred()
            did = self.pdm.save_pending_deferred(d)
            d2 = self.pdm.get_pending_deferred(did,True)
            d2.addErrback(lambda f: self.assertRaises(MyError, f.raiseException))
            d.errback(failure.Failure(MyError('foo')))
            self.assert_(not self.pdm.quick_has_id(did))
        
        def test_noresult_noblock(self):
            d = defer.Deferred()
            did = self.pdm.save_pending_deferred(d)
            d2 = self.pdm.get_pending_deferred(did,False)
            d2.addErrback(lambda f: self.assertRaises(error.ResultNotCompleted, f.raiseException))

        def test_with_callbacks(self):
            d = defer.Deferred()
            d.addCallback(lambda r: r+' foo')
            d.addCallback(lambda r: r+' bar')
            did = self.pdm.save_pending_deferred(d)
            d2 = self.pdm.get_pending_deferred(did,True)
            d.callback('bam')
            d2.addCallback(lambda r: self.assertEquals(r,'bam foo bar'))
    
        def test_with_errbacks(self):
            class MyError(Exception):
                pass
            d = defer.Deferred()
            d.addCallback(lambda r: 'foo')
            d.addErrback(lambda f: 'caught error')
            did = self.pdm.save_pending_deferred(d)
            d2 = self.pdm.get_pending_deferred(did,True)
            d.errback(failure.Failure(MyError('bam')))
            d2.addErrback(lambda f: self.assertRaises(MyError, f.raiseException))
    
        def test_nested_deferreds(self):
            d = defer.Deferred()
            d2 = defer.Deferred()
            d.addCallback(lambda r: d2)
            did = self.pdm.save_pending_deferred(d)
            d.callback('foo')
            d3 = self.pdm.get_pending_deferred(did,False)
            d3.addErrback(lambda f: self.assertRaises(error.ResultNotCompleted, f.raiseException))
            d2.callback('bar')
            d3 = self.pdm.get_pending_deferred(did,False)
            d3.addCallback(lambda r: self.assertEquals(r,'bar'))

#-------------------------------------------------------------------------------
# Regular Unittests
#-------------------------------------------------------------------------------

# This ensures that the code will run either standalone as a script, or that it
# can be picked up by Twisted's `trial` test wrapper to run all the tests.
if tcommon.pexpect is not None:
    if __name__ == '__main__':
        unittest.main(testLoader=IPDocTestLoader(dt_files,dt_modules))
    else:
        testSuite = lambda : makeTestSuite(__name__,dt_files,dt_modules)