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
|
import tests
class Socket(tests.LimitedTestCase):
def test_patched_thread(self):
env = {'EVENTLET_TPOOL_DNS': 'yes'}
output = tests.run_python('tests/env_test_socket_getaddrinfo.py', env=env)
lines = output.splitlines()
self.assertEqual(len(lines), 2, lines)
class Tpool(tests.LimitedTestCase):
longMessage = True
maxDiff = None
@tests.skip_with_pyevent
def test_tpool_size_default(self):
# modify this together with default value in eventlet.tpool
expected = 20
env = {'eventlet_test_limit': str(expected + 20)}
output = tests.run_python('tests/env_test_tpool_size.py', env=env)
lines = output.splitlines()
self.assertEqual(len(lines), 1, output)
highwater = int(output.strip())
self.assertEqual(highwater, expected, output)
@tests.skip_with_pyevent
def test_tpool_size_custom(self):
expected = 40
env = {
'EVENTLET_THREADPOOL_SIZE': str(expected),
'eventlet_test_limit': str(expected + 20),
}
output = tests.run_python('tests/env_test_tpool_size.py', env=env)
lines = output.splitlines()
self.assertEqual(len(lines), 1, output)
highwater = int(output.strip())
self.assertEqual(highwater, expected, output)
def test_tpool_negative(self):
env = {'EVENTLET_THREADPOOL_SIZE': '-1'}
output = tests.run_python('tests/env_test_tpool_size.py', env=env)
lines = output.splitlines()
self.assert_(len(lines) > 1, output)
highwater = int(lines[-1])
self.assert_('AssertionError' in output, output)
self.assertEqual(highwater, 0, output)
def test_tpool_zero(self):
env = {'EVENTLET_THREADPOOL_SIZE': '0'}
output = tests.run_python('tests/env_test_tpool_size.py', env=env)
lines = output.splitlines()
self.assert_(len(lines) > 1, output)
self.assert_('Warning' in output, output)
self.assertEqual(lines[-1], '1', output)
class Hub(tests.LimitedTestCase):
def test_eventlet_hub(self):
for hub in ('selects',):
env = {'EVENTLET_HUB': 'selects'}
output = tests.run_python('tests/env_test_hub.py', env=env)
lines = output.splitlines()
self.assertEqual(len(lines), 1, output)
self.assert_(hub in output, output)
|
