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
|
import os
from eventlet.support import six
from tests.patcher_test import ProcessBase
from tests import skip_with_pyevent
class Socket(ProcessBase):
def test_patched_thread(self):
new_mod = """from eventlet.green import socket
socket.gethostbyname('localhost')
socket.getaddrinfo('localhost', 80)
"""
os.environ['EVENTLET_TPOOL_DNS'] = 'yes'
try:
self.write_to_tempfile("newmod", new_mod)
output, lines = self.launch_subprocess('newmod.py')
self.assertEqual(len(lines), 1, lines)
finally:
del os.environ['EVENTLET_TPOOL_DNS']
class Tpool(ProcessBase):
@skip_with_pyevent
def test_tpool_size(self):
expected = "40"
normal = "20"
new_mod = """from eventlet import tpool
import eventlet
import time
current = [0]
highwater = [0]
def count():
current[0] += 1
time.sleep(0.1)
if current[0] > highwater[0]:
highwater[0] = current[0]
current[0] -= 1
expected = %s
normal = %s
p = eventlet.GreenPool()
for i in range(expected*2):
p.spawn(tpool.execute, count)
p.waitall()
assert highwater[0] > 20, "Highwater %%s <= %%s" %% (highwater[0], normal)
"""
os.environ['EVENTLET_THREADPOOL_SIZE'] = expected
try:
self.write_to_tempfile("newmod", new_mod % (expected, normal))
output, lines = self.launch_subprocess('newmod.py')
self.assertEqual(len(lines), 1, lines)
finally:
del os.environ['EVENTLET_THREADPOOL_SIZE']
def test_tpool_negative(self):
new_mod = """from eventlet import tpool
import eventlet
import time
def do():
print("should not get here")
try:
tpool.execute(do)
except AssertionError:
print("success")
"""
os.environ['EVENTLET_THREADPOOL_SIZE'] = "-1"
try:
self.write_to_tempfile("newmod", new_mod)
output, lines = self.launch_subprocess('newmod.py')
self.assertEqual(len(lines), 2, lines)
self.assertEqual(lines[0], "success", output)
finally:
del os.environ['EVENTLET_THREADPOOL_SIZE']
def test_tpool_zero(self):
new_mod = """from eventlet import tpool
import eventlet
import time
def do():
print("ran it")
tpool.execute(do)
"""
os.environ['EVENTLET_THREADPOOL_SIZE'] = "0"
try:
self.write_to_tempfile("newmod", new_mod)
output, lines = self.launch_subprocess('newmod.py')
self.assertEqual(len(lines), 4, lines)
self.assertEqual(lines[-2], 'ran it', lines)
assert 'Warning' in lines[1] or 'Warning' in lines[0], lines
finally:
del os.environ['EVENTLET_THREADPOOL_SIZE']
class Hub(ProcessBase):
def setUp(self):
super(Hub, self).setUp()
self.old_environ = os.environ.get('EVENTLET_HUB')
os.environ['EVENTLET_HUB'] = 'selects'
def tearDown(self):
if self.old_environ:
os.environ['EVENTLET_HUB'] = self.old_environ
else:
del os.environ['EVENTLET_HUB']
super(Hub, self).tearDown()
def test_eventlet_hub(self):
new_mod = """from eventlet import hubs
print(hubs.get_hub())
"""
self.write_to_tempfile("newmod", new_mod)
output, lines = self.launch_subprocess('newmod.py')
self.assertEqual(len(lines), 2, "\n".join(lines))
assert "selects" in lines[0]
|
