eventlet/tests/env_test.py

   1 import os
   2 from eventlet.support import six
   3 from tests.patcher_test import ProcessBase
   4 from tests import skip_with_pyevent
   5
   6
   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)
  12 """
  13         os.environ['EVENTLET_TPOOL_DNS'] = 'yes'
  14         try:
  15             self.write_to_tempfile("newmod", new_mod)
  16             output, lines = self.launch_subprocess('newmod.py')
  17             self.assertEqual(len(lines), 1, lines)
  18         finally:
  19             del os.environ['EVENTLET_TPOOL_DNS']
  20
  21
  22 class Tpool(ProcessBase):
  23     @skip_with_pyevent
  24     def test_tpool_size(self):
  25         expected = "40"
  26         normal = "20"
  27         new_mod = """from eventlet import tpool
  28 import eventlet
  29 import time
  30 current = [0]
  31 highwater = [0]
  32 def count():
  33     current[0] += 1
  34     time.sleep(0.1)
  35     if current[0] > highwater[0]:
  36         highwater[0] = current[0]
  37     current[0] -= 1
  38 expected = %s
  39 normal = %s
  40 p = eventlet.GreenPool()
  41 for i in range(expected*2):
  42     p.spawn(tpool.execute, count)
  43 p.waitall()
  44 assert highwater[0] > 20, "Highwater %%s  <= %%s" %% (highwater[0], normal)
  45 """
  46         os.environ['EVENTLET_THREADPOOL_SIZE'] = expected
  47         try:
  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)
  51         finally:
  52             del os.environ['EVENTLET_THREADPOOL_SIZE']
  53
  54     def test_tpool_negative(self):
  55         new_mod = """from eventlet import tpool
  56 import eventlet
  57 import time
  58 def do():
  59     print("should not get here")
  60 try:
  61     tpool.execute(do)
  62 except AssertionError:
  63     print("success")
  64 """
  65         os.environ['EVENTLET_THREADPOOL_SIZE'] = "-1"
  66         try:
  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)
  71         finally:
  72             del os.environ['EVENTLET_THREADPOOL_SIZE']
  73
  74     def test_tpool_zero(self):
  75         new_mod = """from eventlet import tpool
  76 import eventlet
  77 import time
  78 def do():
  79     print("ran it")
  80 tpool.execute(do)
  81 """
  82         os.environ['EVENTLET_THREADPOOL_SIZE'] = "0"
  83         try:
  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
  89         finally:
  90             del os.environ['EVENTLET_THREADPOOL_SIZE']
  91
  92
  93 class Hub(ProcessBase):
  94
  95     def setUp(self):
  96         super(Hub, self).setUp()
  97         self.old_environ = os.environ.get('EVENTLET_HUB')
  98         os.environ['EVENTLET_HUB'] = 'selects'
  99
 100     def tearDown(self):
 101         if self.old_environ:
 102             os.environ['EVENTLET_HUB'] = self.old_environ
 103         else:
 104             del os.environ['EVENTLET_HUB']
 105         super(Hub, self).tearDown()
 106
 107     def test_eventlet_hub(self):
 108         new_mod = """from eventlet import hubs
 109 print(hubs.get_hub())
 110 """
 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]