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
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
|
import os
import random
import subprocess
import sys
import unittest
try:
import _ssl
except ImportError:
_ssl = None
try:
import ssl
except ImportError:
ssl = None
try:
import ctypes
except ImportError:
# Python 2.4
ctypes = None
import mitogen
import testlib
import plain_old_module
def _find_ssl_linux():
proc = subprocess.Popen(
['ldd', _ssl.__file__],
stdout=subprocess.PIPE, stderr=subprocess.PIPE,
)
b_stdout, b_stderr = proc.communicate()
assert proc.returncode == 0
assert b_stderr.decode() == ''
for line in b_stdout.decode().splitlines():
bits = line.split()
if bits[0].startswith('libssl'):
return bits[2]
def _find_ssl_darwin():
proc = subprocess.Popen(
['otool', '-l', _ssl.__file__],
stdout=subprocess.PIPE, stderr=subprocess.PIPE,
)
b_stdout, b_stderr = proc.communicate()
assert proc.returncode == 0
assert b_stderr.decode() == ''
for line in b_stdout.decode().splitlines():
bits = line.split()
if bits[0] == 'name' and 'libssl' in bits[1]:
return bits[1]
if ctypes and sys.platform.startswith('linux'):
LIBSSL_PATH = _find_ssl_linux()
elif ctypes and sys.platform == 'darwin':
LIBSSL_PATH = _find_ssl_darwin()
else:
LIBSSL_PATH = None
if ctypes and LIBSSL_PATH:
c_ssl = ctypes.CDLL(LIBSSL_PATH)
c_ssl.RAND_pseudo_bytes.argtypes = [ctypes.c_char_p, ctypes.c_int]
c_ssl.RAND_pseudo_bytes.restype = ctypes.c_int
def ping():
return 123
def random_random():
return random.random()
def RAND_pseudo_bytes(n=32):
buf = ctypes.create_string_buffer(n)
assert 1 == c_ssl.RAND_pseudo_bytes(buf, n)
return buf[:]
def exercise_importer(n):
"""
Ensure the forked child has a sensible importer.
"""
sys.path.remove(testlib.DATA_DIR)
import simple_pkg.a
return simple_pkg.a.subtract_one_add_two(n)
skipIfUnsupported = unittest.skipIf(
condition=(not mitogen.fork.FORK_SUPPORTED),
reason="mitogen.fork unsupported on this platform"
)
class ForkTest(testlib.RouterMixin, testlib.TestCase):
def test_okay(self):
context = self.router.fork()
self.assertNotEqual(context.call(os.getpid), os.getpid())
self.assertEqual(context.call(os.getppid), os.getpid())
def test_random_module_diverges(self):
context = self.router.fork()
self.assertNotEqual(context.call(random_random), random_random())
@unittest.skipIf(
condition=LIBSSL_PATH is None or ctypes is None,
reason='cant test libssl on this platform',
)
def test_ssl_module_diverges(self):
# Ensure generator state is initialized.
RAND_pseudo_bytes()
context = self.router.fork()
self.assertNotEqual(context.call(RAND_pseudo_bytes),
RAND_pseudo_bytes())
def test_importer(self):
context = self.router.fork()
self.assertEqual(2, context.call(exercise_importer, 1))
def test_on_start(self):
recv = mitogen.core.Receiver(self.router)
def on_start(econtext):
sender = mitogen.core.Sender(econtext.parent, recv.handle)
sender.send(123)
context = self.router.fork(on_start=on_start)
self.assertEqual(123, recv.get().unpickle())
ForkTest = skipIfUnsupported(ForkTest)
class DoubleChildTest(testlib.RouterMixin, testlib.TestCase):
def test_okay(self):
# When forking from the master process, Mitogen had nothing to do with
# setting up stdio -- that was inherited wherever the Master is running
# (supervisor, iTerm, etc). When forking from a Mitogen child context
# however, Mitogen owns all of fd 0, 1, and 2, and during the fork
# procedure, it deletes all of these descriptors. That leaves the
# process in a weird state that must be handled by some combination of
# fork.py and ExternalContext.main().
# Below we simply test whether ExternalContext.main() managed to boot
# successfully. In future, we need lots more tests.
c1 = self.router.fork()
c2 = self.router.fork(via=c1)
self.assertEqual(123, c2.call(ping))
def test_importer(self):
c1 = self.router.fork(name='c1')
c2 = self.router.fork(name='c2', via=c1)
self.assertEqual(2, c2.call(exercise_importer, 1))
DoubleChildTest = skipIfUnsupported(DoubleChildTest)
|
