2 from eventlet.support import six
3 from tests.patcher_test import ProcessBase
4 from tests import skip_with_pyevent
7 class Socket(ProcessBase):
8 def test_patched_thread(self):
9 new_mod = """from eventlet.green import socket
10 socket.gethostbyname('localhost')
11 socket.getaddrinfo('localhost', 80)
13 os.environ['EVENTLET_TPOOL_DNS'] = 'yes'
15 self.write_to_tempfile("newmod", new_mod)
16 output, lines = self.launch_subprocess('newmod.py')
17 self.assertEqual(len(lines), 1, lines)
19 del os.environ['EVENTLET_TPOOL_DNS']
22 class Tpool(ProcessBase):
24 def test_tpool_size(self):
27 new_mod = """from eventlet import tpool
35 if current[0] > highwater[0]:
36 highwater[0] = current[0]
40 p = eventlet.GreenPool()
41 for i in range(expected*2):
42 p.spawn(tpool.execute, count)
44 assert highwater[0] > 20, "Highwater %%s <= %%s" %% (highwater[0], normal)
46 os.environ['EVENTLET_THREADPOOL_SIZE'] = expected
48 self.write_to_tempfile("newmod", new_mod % (expected, normal))
49 output, lines = self.launch_subprocess('newmod.py')
50 self.assertEqual(len(lines), 1, lines)
52 del os.environ['EVENTLET_THREADPOOL_SIZE']
54 def test_tpool_negative(self):
55 new_mod = """from eventlet import tpool
59 print("should not get here")
62 except AssertionError:
65 os.environ['EVENTLET_THREADPOOL_SIZE'] = "-1"
67 self.write_to_tempfile("newmod", new_mod)
68 output, lines = self.launch_subprocess('newmod.py')
69 self.assertEqual(len(lines), 2, lines)
70 self.assertEqual(lines[0], "success", output)
72 del os.environ['EVENTLET_THREADPOOL_SIZE']
74 def test_tpool_zero(self):
75 new_mod = """from eventlet import tpool
82 os.environ['EVENTLET_THREADPOOL_SIZE'] = "0"
84 self.write_to_tempfile("newmod", new_mod)
85 output, lines = self.launch_subprocess('newmod.py')
86 self.assertEqual(len(lines), 4, lines)
87 self.assertEqual(lines[-2], 'ran it', lines)
88 assert 'Warning' in lines[1] or 'Warning' in lines[0], lines
90 del os.environ['EVENTLET_THREADPOOL_SIZE']
93 class Hub(ProcessBase):
96 super(Hub, self).setUp()
97 self.old_environ = os.environ.get('EVENTLET_HUB')
98 os.environ['EVENTLET_HUB'] = 'selects'
102 os.environ['EVENTLET_HUB'] = self.old_environ
104 del os.environ['EVENTLET_HUB']
105 super(Hub, self).tearDown()
107 def test_eventlet_hub(self):
108 new_mod = """from eventlet import hubs
109 print(hubs.get_hub())
111 self.write_to_tempfile("newmod", new_mod)
112 output, lines = self.launch_subprocess('newmod.py')
113 self.assertEqual(len(lines), 2, "\n".join(lines))
114 assert "selects" in lines[0]