tests/subprocess_test.py

   1 import eventlet
   2 from eventlet.green import subprocess
   3 import eventlet.patcher
   4 from nose.plugins.skip import SkipTest
   5 import sys
   6 import time
   7 original_subprocess = eventlet.patcher.original('subprocess')
   8
   9
  10 def test_subprocess_wait():
  11     # https://bitbucket.org/eventlet/eventlet/issue/89
  12     # In Python 3.3 subprocess.Popen.wait() method acquired `timeout`
  13     # argument.
  14     # RHEL backported it to their Python 2.6 package.
  15     p = subprocess.Popen(
  16         [sys.executable, "-c", "import time; time.sleep(0.5)"])
  17     ok = False
  18     t1 = time.time()
  19     try:
  20         p.wait(timeout=0.1)
  21     except subprocess.TimeoutExpired:
  22         ok = True
  23     tdiff = time.time() - t1
  24     assert ok, 'did not raise subprocess.TimeoutExpired'
  25     assert 0.1 <= tdiff <= 0.2, 'did not stop within allowed time'
  26
  27
  28 def test_communicate_with_poll():
  29     # https://github.com/eventlet/eventlet/pull/24
  30     # `eventlet.green.subprocess.Popen.communicate()` was broken
  31     # in Python 2.7 because the usage of the `select` module was moved from
  32     # `_communicate` into two other methods `_communicate_with_select`
  33     # and `_communicate_with_poll`. Link to 2.7's implementation:
  34     # http://hg.python.org/cpython/file/2145593d108d/Lib/subprocess.py#l1255
  35     if getattr(original_subprocess.Popen, '_communicate_with_poll', None) is None:
  36         raise SkipTest('original subprocess.Popen does not have _communicate_with_poll')
  37
  38     p = subprocess.Popen(
  39         [sys.executable, '-c', 'import time; time.sleep(0.5)'],
  40         stdout=subprocess.PIPE, stderr=subprocess.PIPE)
  41     t1 = time.time()
  42     eventlet.with_timeout(0.1, p.communicate, timeout_value=True)
  43     tdiff = time.time() - t1
  44     assert 0.1 <= tdiff <= 0.2, 'did not stop within allowed time'