##// END OF EJS Templates
Merge pull request #1928 from minrk/nodb...
Merge pull request #1928 from minrk/nodb NoDB is now the default backend for storing Hub results in IPython clusters. This prevents the use of delayed retrieval of results from the Hub, but means that the default use will have significantly reduced memory footprint (and some natural use patterns won't run into memory exhaustion as easily as before). All these advanced features remain available with any of the 'real' storage backends (dictdb, mongodb and sqlitedb). A small bug discovered along the way when using HubResults for execute requests is also fixed and tested.

File last commit:

r7336:aca86345
r7517:e5d53aca merge
Show More
test_hooks.py
75 lines | 2.3 KiB | text/x-python | PythonLexer
# -*- coding: utf-8 -*-
"""Tests for CommandChainDispatcher."""
from __future__ import absolute_import
#-----------------------------------------------------------------------------
# Imports
#-----------------------------------------------------------------------------
import nose.tools as nt
from IPython.core.error import TryNext
from IPython.core.hooks import CommandChainDispatcher
#-----------------------------------------------------------------------------
# Local utilities
#-----------------------------------------------------------------------------
# Define two classes, one which succeeds and one which raises TryNext. Each
# sets the attribute `called` to True when it is called.
class Okay(object):
def __init__(self, message):
self.message = message
self.called = False
def __call__(self):
self.called = True
return self.message
class Fail(object):
def __init__(self, message):
self.message = message
self.called = False
def __call__(self):
self.called = True
raise TryNext(self.message)
#-----------------------------------------------------------------------------
# Test functions
#-----------------------------------------------------------------------------
def test_command_chain_dispatcher_ff():
"""Test two failing hooks"""
fail1 = Fail(u'fail1')
fail2 = Fail(u'fail2')
dp = CommandChainDispatcher([(0, fail1),
(10, fail2)])
try:
dp()
except TryNext as e:
nt.assert_equal(str(e), u'fail2')
else:
assert False, "Expected exception was not raised."
nt.assert_true(fail1.called)
nt.assert_true(fail2.called)
def test_command_chain_dispatcher_fofo():
"""Test a mixture of failing and succeeding hooks."""
fail1 = Fail(u'fail1')
fail2 = Fail(u'fail2')
okay1 = Okay(u'okay1')
okay2 = Okay(u'okay2')
dp = CommandChainDispatcher([(0, fail1),
# (5, okay1), # add this later
(10, fail2),
(15, okay2)])
dp.add(okay1, 5)
nt.assert_equal(dp(), u'okay1')
nt.assert_true(fail1.called)
nt.assert_true(okay1.called)
nt.assert_false(fail2.called)
nt.assert_false(okay2.called)