Spaces:
Runtime error
Runtime error
File size: 2,698 Bytes
4a51346 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 |
"""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
#-----------------------------------------------------------------------------
# Stdlib imports
import time
# 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()
assert j.result["interval"] == t_short
def test_flush():
"""Test job control"""
jobs = bg.BackgroundJobManager()
j = jobs.new(sleeper)
j.join()
assert len(jobs.completed) == 1
assert len(jobs.dead) == 0
jobs.flush()
assert len(jobs.completed) == 0
def test_dead():
"""Test control of dead jobs"""
jobs = bg.BackgroundJobManager()
j = jobs.new(crasher)
j.join()
assert len(jobs.completed) == 0
assert len(jobs.dead) == 1
jobs.flush()
assert 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)
assert len(jobs.running) == 1
assert len(jobs.completed) == 0
j.join()
assert len(jobs.running) == 0
assert len(jobs.completed) == 1
|