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