tests: subprocess.communicate block on Python 2.7

https://github.com/eventlet/eventlet/pull/24
+ PEP8 fix tabs
This commit is contained in:
Sergey Shepelev
2013-06-10 13:21:38 +04:00
parent e11157a442
commit 25812fca81

View File

@@ -1,6 +1,7 @@
import eventlet import eventlet
from eventlet.green import subprocess from eventlet.green import subprocess
import eventlet.patcher import eventlet.patcher
from nose.plugins.skip import SkipTest
import os import os
import sys import sys
import time import time
@@ -12,8 +13,8 @@ def test_subprocess_wait():
# In Python 3.3 subprocess.Popen.wait() method acquired `timeout` # In Python 3.3 subprocess.Popen.wait() method acquired `timeout`
# argument. # argument.
# RHEL backported it to their Python 2.6 package. # RHEL backported it to their Python 2.6 package.
p = subprocess.Popen([sys.executable, p = subprocess.Popen(
"-c", "import time; time.sleep(0.5)"]) [sys.executable, "-c", "import time; time.sleep(0.5)"])
ok = False ok = False
t1 = time.time() t1 = time.time()
try: try:
@@ -23,3 +24,22 @@ def test_subprocess_wait():
tdiff = time.time() - t1 tdiff = time.time() - t1
assert ok == True, 'did not raise subprocess.TimeoutExpired' assert ok == True, 'did not raise subprocess.TimeoutExpired'
assert 0.1 <= tdiff <= 0.2, 'did not stop within allowed time' assert 0.1 <= tdiff <= 0.2, 'did not stop within allowed time'
def test_communicate_with_poll():
# https://github.com/eventlet/eventlet/pull/24
# `eventlet.green.subprocess.Popen.communicate()` was broken
# in Python 2.7 because the usage of the `select` module was moved from
# `_communicate` into two other methods `_communicate_with_select`
# and `_communicate_with_poll`. Link to 2.7's implementation:
# http://hg.python.org/cpython/file/2145593d108d/Lib/subprocess.py#l1255
if getattr(original_subprocess.Popen, '_communicate_with_poll', None) is None:
raise SkipTest('original subprocess.Popen does not have _communicate_with_poll')
p = subprocess.Popen(
[sys.executable, '-c', 'import time; time.sleep(0.5)'],
stdout=subprocess.PIPE, stderr=subprocess.PIPE)
t1 = time.time()
eventlet.with_timeout(0.1, p.communicate, timeout_value=True)
tdiff = time.time() - t1
assert 0.1 <= tdiff <= 0.2, 'did not stop within allowed time'