blob: aeb65e4570fd9bae5ee3d75ee018fd178a0cb2c7 (
plain)
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
|
import pickle
import sys
import unittest
from nose import case
from nose.plugins import multiprocess
from nose.plugins.skip import SkipTest
from nose.config import Config
from nose.loader import TestLoader
try:
# 2.7+
from unittest.runner import _WritelnDecorator
except ImportError:
from unittest import _WritelnDecorator
class ArgChecker:
def __init__(self, target, args):
self.target = target
self.args = args
# skip the id and queues
pargs = args[7:]
self.pickled = pickle.dumps(pargs)
try:
testQueue = args[1]
testQueue.get(timeout=0)
except:
pass # ok if queue is empty
def start(self,*args):
pass
def is_alive(self):
return False
def setup(mod):
multiprocess._import_mp()
if not multiprocess.Process:
raise SkipTest("multiprocessing not available")
mod.Process = multiprocess.Process
multiprocess.Process = ArgChecker
class T(unittest.TestCase):
__test__ = False
def runTest(self):
pass
def test_mp_process_args_pickleable():
# TODO(Kumar) this test needs to be more succint.
# If you start seeing it timeout then perhaps we need to skip it again.
# raise SkipTest('this currently gets stuck in poll() 90% of the time')
test = case.Test(T('runTest'))
config = Config()
config.multiprocess_workers = 2
config.multiprocess_timeout = 5
runner = multiprocess.MultiProcessTestRunner(
stream=_WritelnDecorator(sys.stdout),
verbosity=10,
loaderClass=TestLoader,
config=config)
runner.run(test)
|
