--- /dev/null
+from tests.patcher_test import ProcessBase
+
+
+class ForkTest(ProcessBase):
+ def test_simple(self):
+ newmod = '''
+import eventlet
+import os
+import sys
+import signal
+from eventlet.support import bytes_to_str, six
+mydir = %r
+signal_file = os.path.join(mydir, "output.txt")
+pid = os.fork()
+if (pid != 0):
+ eventlet.Timeout(10)
+ try:
+ port = None
+ while True:
+ try:
+ contents = open(signal_file, "rb").read()
+ port = int(contents.split()[0])
+ break
+ except (IOError, IndexError, ValueError, TypeError):
+ eventlet.sleep(0.1)
+ eventlet.connect(('127.0.0.1', port))
+ while True:
+ try:
+ contents = open(signal_file, "rb").read()
+ result = contents.split()[1]
+ break
+ except (IOError, IndexError):
+ eventlet.sleep(0.1)
+ print('result {0}'.format(bytes_to_str(result)))
+ finally:
+ os.kill(pid, signal.SIGTERM)
+else:
+ try:
+ s = eventlet.listen(('', 0))
+ fd = open(signal_file, "wb")
+ fd.write(six.b(str(s.getsockname()[1])))
+ fd.write(b"\\n")
+ fd.flush()
+ s.accept()
+ fd.write(b"done")
+ fd.flush()
+ finally:
+ fd.close()
+'''
+ self.write_to_tempfile("newmod", newmod % self.tempdir)
+ output, lines = self.launch_subprocess('newmod.py')
+ self.assertEqual(lines[0], "result done", output)