eventlet/tests/stdlib/test_asyncore.py

   1 from eventlet import patcher
   2 from eventlet.green import asyncore
   3 from eventlet.green import select
   4 from eventlet.green import socket
   5 from eventlet.green import threading
   6 from eventlet.green import time
   7
   8 patcher.inject("test.test_asyncore", globals())
   9
  10
  11 def new_closeall_check(self, usedefault):
  12     # Check that close_all() closes everything in a given map
  13
  14     l = []
  15     testmap = {}
  16     for i in range(10):
  17         c = dummychannel()
  18         l.append(c)
  19         self.assertEqual(c.socket.closed, False)
  20         testmap[i] = c
  21
  22     if usedefault:
  23         # the only change we make is to not assign to asyncore.socket_map
  24         # because doing so fails to assign to the real asyncore's socket_map
  25         # and thus the test fails
  26         socketmap = asyncore.socket_map.copy()
  27         try:
  28             asyncore.socket_map.clear()
  29             asyncore.socket_map.update(testmap)
  30             asyncore.close_all()
  31         finally:
  32             testmap = asyncore.socket_map.copy()
  33             asyncore.socket_map.clear()
  34             asyncore.socket_map.update(socketmap)
  35     else:
  36         asyncore.close_all(testmap)
  37
  38     self.assertEqual(len(testmap), 0)
  39
  40     for c in l:
  41         self.assertEqual(c.socket.closed, True)
  42
  43 HelperFunctionTests.closeall_check = new_closeall_check
  44
  45 try:
  46     # Eventlet's select() emulation doesn't support the POLLPRI flag,
  47     # which this test relies on.  Therefore, nuke it!
  48     BaseTestAPI.test_handle_expt = lambda *a, **kw: None
  49 except NameError:
  50     pass
  51
  52 try:
  53     # temporarily disabling these tests in the python2.7/pyevent configuration
  54     from tests import using_pyevent
  55     import sys
  56     if using_pyevent(None) and sys.version_info >= (2, 7):
  57         TestAPI_UseSelect.test_handle_accept = lambda *a, **kw: None
  58         TestAPI_UseSelect.test_handle_close = lambda *a, **kw: None
  59         TestAPI_UseSelect.test_handle_read = lambda *a, **kw: None
  60 except NameError:
  61     pass
  62
  63 if __name__ == "__main__":
  64     test_main()