python-eventlet/tests/subprocess_test.py

   1 import eventlet
   2 from eventlet.green import subprocess
   3 import eventlet.patcher
   4 import sys
   5 import time
   6 original_subprocess = eventlet.patcher.original('subprocess')
   7
   8
   9 def test_subprocess_wait():
  10     # https://bitbucket.org/eventlet/eventlet/issue/89
  11     # In Python 3.3 subprocess.Popen.wait() method acquired `timeout`
  12     # argument.
  13     # RHEL backported it to their Python 2.6 package.
  14     cmd = [sys.executable, "-c", "import time; time.sleep(0.5)"]
  15     p = subprocess.Popen(cmd)
  16     ok = False
  17     t1 = time.time()
  18     try:
  19         p.wait(timeout=0.1)
  20     except subprocess.TimeoutExpired as e:
  21         str(e)  # make sure it doesn't throw
  22         assert e.cmd == cmd
  23         assert e.timeout == 0.1
  24         ok = True
  25     tdiff = time.time() - t1
  26     assert ok, 'did not raise subprocess.TimeoutExpired'
  27     assert 0.1 <= tdiff <= 0.2, 'did not stop within allowed time'
  28
  29
  30 def test_communicate_with_poll():
  31     # This test was being skipped since git 25812fca8, I don't there's
  32     # a need to do this. The original comment:
  33     #
  34     # https://github.com/eventlet/eventlet/pull/24
  35     # `eventlet.green.subprocess.Popen.communicate()` was broken
  36     # in Python 2.7 because the usage of the `select` module was moved from
  37     # `_communicate` into two other methods `_communicate_with_select`
  38     # and `_communicate_with_poll`. Link to 2.7's implementation:
  39     # http://hg.python.org/cpython/file/2145593d108d/Lib/subprocess.py#l1255
  40
  41     p = subprocess.Popen(
  42         [sys.executable, '-c', 'import time; time.sleep(0.5)'],
  43         stdout=subprocess.PIPE, stderr=subprocess.PIPE)
  44     t1 = time.time()
  45     eventlet.with_timeout(0.1, p.communicate, timeout_value=True)
  46     tdiff = time.time() - t1
  47     assert 0.1 <= tdiff <= 0.2, 'did not stop within allowed time'
  48
  49
  50 def test_close_popen_stdin_with_close_fds():
  51     p = subprocess.Popen(
  52         ['ls'],
  53         stdin=subprocess.PIPE,
  54         stdout=subprocess.PIPE,
  55         stderr=subprocess.PIPE,
  56         close_fds=True,
  57         shell=False,
  58         cwd=None,
  59         env=None)
  60
  61     p.communicate(None)
  62
  63     try:
  64         p.stdin.close()
  65     except Exception as e:
  66         assert False, "Exception should not be raised, got %r instead" % e