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
87
|
"""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
|
