Add python-eventlet package to MOS 8.0 repository

[packages/trusty/python-eventlet.git] / python-eventlet / tests / subprocess_test.py

diff --git a/eventlet/tests/subprocess_test.py b/python-eventlet/tests/subprocess_test.py
similarity index 72%
rename from eventlet/tests/subprocess_test.py
rename to python-eventlet/tests/subprocess_test.py
index 6964a742b85306172015789bbd5bacc6adb0de79..085f656c1e8695cafa04c545a7e010f446b2bb40 100644 (file)
--- a/eventlet/tests/subprocess_test.py
+++ b/python-eventlet/tests/subprocess_test.py
@@ -1,7 +1,6 @@
 import eventlet
 from eventlet.green import subprocess
 import eventlet.patcher
-from nose.plugins.skip import SkipTest
 import sys
 import time
 original_subprocess = eventlet.patcher.original('subprocess')
@@ -19,7 +18,7 @@ def test_subprocess_wait():
     try:
         p.wait(timeout=0.1)
     except subprocess.TimeoutExpired as e:
-        str(e)  # make sure it doesnt throw
+        str(e)  # make sure it doesn't throw
         assert e.cmd == cmd
         assert e.timeout == 0.1
         ok = True
@@ -29,14 +28,15 @@ def test_subprocess_wait():
 
 
 def test_communicate_with_poll():
+    # This test was being skipped since git 25812fca8, I don't there's
+    # a need to do this. The original comment:
+    #
     # 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)'],
@@ -45,3 +45,22 @@ def test_communicate_with_poll():
     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'
+
+
+def test_close_popen_stdin_with_close_fds():
+    p = subprocess.Popen(
+        ['ls'],
+        stdin=subprocess.PIPE,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+        close_fds=True,
+        shell=False,
+        cwd=None,
+        env=None)
+
+    p.communicate(None)
+
+    try:
+        p.stdin.close()
+    except Exception as e:
+        assert False, "Exception should not be raised, got %r instead" % e