##// END OF EJS Templates
invoke css
invoke css

File last commit:

r11126:b337c5c3
r19394:5a48398c
Show More
test_backgroundjobs.py
89 lines | 2.8 KiB | text/x-python | PythonLexer
/ IPython / lib / tests / test_backgroundjobs.py
"""Tests for pylab tools module.
"""
#-----------------------------------------------------------------------------
# Copyright (c) 2011, the IPython Development Team.
#
# Distributed under the terms of the Modified BSD License.
#
# The full license is in the file COPYING.txt, distributed with this software.
#-----------------------------------------------------------------------------
#-----------------------------------------------------------------------------
# Imports
#-----------------------------------------------------------------------------
from __future__ import print_function
# Stdlib imports
import time
# Third-party imports
import nose.tools as nt
# Our own imports
from IPython.lib import backgroundjobs as bg
#-----------------------------------------------------------------------------
# Globals and constants
#-----------------------------------------------------------------------------
t_short = 0.0001 # very short interval to wait on jobs
#-----------------------------------------------------------------------------
# Local utilities
#-----------------------------------------------------------------------------
def sleeper(interval=t_short, *a, **kw):
args = dict(interval=interval,
other_args=a,
kw_args=kw)
time.sleep(interval)
return args
def crasher(interval=t_short, *a, **kw):
time.sleep(interval)
raise Exception("Dead job with interval %s" % interval)
#-----------------------------------------------------------------------------
# Classes and functions
#-----------------------------------------------------------------------------
def test_result():
"""Test job submission and result retrieval"""
jobs = bg.BackgroundJobManager()
j = jobs.new(sleeper)
j.join()
nt.assert_equal(j.result['interval'], t_short)
def test_flush():
"""Test job control"""
jobs = bg.BackgroundJobManager()
j = jobs.new(sleeper)
j.join()
nt.assert_equal(len(jobs.completed), 1)
nt.assert_equal(len(jobs.dead), 0)
jobs.flush()
nt.assert_equal(len(jobs.completed), 0)
def test_dead():
"""Test control of dead jobs"""
jobs = bg.BackgroundJobManager()
j = jobs.new(crasher)
j.join()
nt.assert_equal(len(jobs.completed), 0)
nt.assert_equal(len(jobs.dead), 1)
jobs.flush()
nt.assert_equal(len(jobs.dead), 0)
def test_longer():
"""Test control of longer-running jobs"""
jobs = bg.BackgroundJobManager()
# Sleep for long enough for the following two checks to still report the
# job as running, but not so long that it makes the test suite noticeably
# slower.
j = jobs.new(sleeper, 0.1)
nt.assert_equal(len(jobs.running), 1)
nt.assert_equal(len(jobs.completed), 0)
j.join()
nt.assert_equal(len(jobs.running), 0)
nt.assert_equal(len(jobs.completed), 1)