tests/fork_test.py

   1 from tests.patcher_test import ProcessBase
   2
   3
   4 class ForkTest(ProcessBase):
   5     def test_simple(self):
   6         newmod = '''
   7 import eventlet
   8 import os
   9 import sys
  10 import signal
  11 mydir = %r
  12 signal_file = os.path.join(mydir, "output.txt")
  13 pid = os.fork()
  14 if (pid != 0):
  15   eventlet.Timeout(10)
  16   try:
  17     port = None
  18     while True:
  19       try:
  20         contents = open(signal_file, "rb").read()
  21         port = int(contents.split()[0])
  22         break
  23       except (IOError, IndexError, ValueError, TypeError):
  24         eventlet.sleep(0.1)
  25     eventlet.connect(('127.0.0.1', port))
  26     while True:
  27       try:
  28         contents = open(signal_file, "rb").read()
  29         result = contents.split()[1]
  30         break
  31       except (IOError, IndexError):
  32         eventlet.sleep(0.1)
  33     print('result {0}'.format(result))
  34   finally:
  35     os.kill(pid, signal.SIGTERM)
  36 else:
  37   try:
  38     s = eventlet.listen(('', 0))
  39     fd = open(signal_file, "wb")
  40     fd.write(str(s.getsockname()[1]))
  41     fd.write("\\n")
  42     fd.flush()
  43     s.accept()
  44     fd.write("done")
  45     fd.flush()
  46   finally:
  47     fd.close()
  48 '''
  49         self.write_to_tempfile("newmod", newmod % self.tempdir)
  50         output, lines = self.launch_subprocess('newmod.py')
  51         self.assertEqual(lines[0], "result done", output)