forked from ipython/ipython
-
Notifications
You must be signed in to change notification settings - Fork 0
/
test_backgroundjobs.py
91 lines (76 loc) · 2.87 KB
/
test_backgroundjobs.py
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
88
89
90
91
"""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 sys
import time
# Third-party imports
import nose.tools as nt
# Our own imports
from IPython.lib import backgroundjobs as bg
from IPython.testing import decorators as dec
#-----------------------------------------------------------------------------
# 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_equals(j.result['interval'], t_short)
def test_flush():
"""Test job control"""
jobs = bg.BackgroundJobManager()
j = jobs.new(sleeper)
j.join()
nt.assert_equals(len(jobs.completed), 1)
nt.assert_equals(len(jobs.dead), 0)
jobs.flush()
nt.assert_equals(len(jobs.completed), 0)
def test_dead():
"""Test control of dead jobs"""
jobs = bg.BackgroundJobManager()
j = jobs.new(crasher)
j.join()
nt.assert_equals(len(jobs.completed), 0)
nt.assert_equals(len(jobs.dead), 1)
jobs.flush()
nt.assert_equals(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_equals(len(jobs.running), 1)
nt.assert_equals(len(jobs.completed), 0)
j.join()
nt.assert_equals(len(jobs.running), 0)
nt.assert_equals(len(jobs.completed), 1)