python-eventlet/tests/backdoor_test.py

   1 import os
   2 import os.path
   3
   4 import eventlet
   5
   6 from eventlet import backdoor
   7 from eventlet.green import socket
   8
   9 from tests import LimitedTestCase, main
  10
  11
  12 class BackdoorTest(LimitedTestCase):
  13     def test_server(self):
  14         listener = socket.socket()
  15         listener.bind(('localhost', 0))
  16         listener.listen(50)
  17         serv = eventlet.spawn(backdoor.backdoor_server, listener)
  18         client = socket.socket()
  19         client.connect(('localhost', listener.getsockname()[1]))
  20         self._run_test_on_client_and_server(client, serv)
  21
  22     def _run_test_on_client_and_server(self, client, server_thread):
  23         f = client.makefile('rw')
  24         assert 'Python' in f.readline()
  25         f.readline()  # build info
  26         f.readline()  # help info
  27         assert 'InteractiveConsole' in f.readline()
  28         self.assertEqual('>>> ', f.read(4))
  29         f.write('print("hi")\n')
  30         f.flush()
  31         self.assertEqual('hi\n', f.readline())
  32         self.assertEqual('>>> ', f.read(4))
  33         f.close()
  34         client.close()
  35         server_thread.kill()
  36         # wait for the console to discover that it's dead
  37         eventlet.sleep(0.1)
  38
  39     def test_server_on_ipv6_socket(self):
  40         listener = socket.socket(socket.AF_INET6)
  41         listener.bind(('::1', 0))
  42         listener.listen(5)
  43         serv = eventlet.spawn(backdoor.backdoor_server, listener)
  44         client = socket.socket(socket.AF_INET6)
  45         client.connect(listener.getsockname())
  46         self._run_test_on_client_and_server(client, serv)
  47
  48     def test_server_on_unix_socket(self):
  49         SOCKET_PATH = '/tmp/eventlet_backdoor_test.socket'
  50         if os.path.exists(SOCKET_PATH):
  51             os.unlink(SOCKET_PATH)
  52         listener = socket.socket(socket.AF_UNIX)
  53         listener.bind(SOCKET_PATH)
  54         listener.listen(5)
  55         serv = eventlet.spawn(backdoor.backdoor_server, listener)
  56         client = socket.socket(socket.AF_UNIX)
  57         client.connect(SOCKET_PATH)
  58         self._run_test_on_client_and_server(client, serv)
  59
  60
  61 if __name__ == '__main__':
  62     main()