tests/env_test.py

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