test__subprocess.py: add a few more regression tests

greentest/test__subprocess.py

 import sys
 import os
 import greentest
-import time
-from gevent import subprocess, sleep
+from gevent import subprocess
+
+
+if subprocess.mswindows:
+    SETBINARY = ('import msvcrt; msvcrt.setmode(sys.stdout.fileno(), '
+                                                'os.O_BINARY);')
+else:
+    SETBINARY = ''
 
 
 class Test(greentest.TestCase):
@@ -19,16 +25,14 @@ class Test(greentest.TestCase):
         else:
             raise AssertionError('Expected OSError: [Errno 2] No such file or directory')
 
-    if os.path.exists('/proc'):
-        def test_leak(self):
-            fd_directory = '/proc/%d/fd' % os.getpid()
-            num_before = len(os.listdir(fd_directory))
-            p = subprocess.Popen([sys.executable, "-c", "print()"],
-                                 stdout=subprocess.PIPE)
-            p.wait()
-            del p
-            num_after = len(os.listdir(fd_directory))
-            self.assertEqual(num_before, num_after)
+    def test_leak(self):
+        num_before = greentest.get_number_open_files()
+        p = subprocess.Popen([sys.executable, "-c", "print()"],
+                             stdout=subprocess.PIPE)
+        p.wait()
+        del p
+        num_after = greentest.get_number_open_files()
+        self.assertEqual(num_before, num_after)
 
     def test_communicate(self):
         p = subprocess.Popen([sys.executable, "-c",
@@ -45,6 +49,62 @@ class Test(greentest.TestCase):
         else:
             self.assertEqual(stderr, "pineapple")
 
+    def test_universal1(self):
+        p = subprocess.Popen([sys.executable, "-c",
+                          'import sys,os;' + SETBINARY +
+                          'sys.stdout.write("line1\\n");'
+                          'sys.stdout.flush();'
+                          'sys.stdout.write("line2\\r");'
+                          'sys.stdout.flush();'
+                          'sys.stdout.write("line3\\r\\n");'
+                          'sys.stdout.flush();'
+                          'sys.stdout.write("line4\\r");'
+                          'sys.stdout.flush();'
+                          'sys.stdout.write("\\nline5");'
+                          'sys.stdout.flush();'
+                          'sys.stdout.write("\\nline6");'],
+                         stdout=subprocess.PIPE,
+                         universal_newlines=1)
+        try:
+            stdout = p.stdout.read()
+            if hasattr(file, 'newlines'):
+                # Interpreter with universal newline support
+                self.assertEqual(stdout,
+                                 "line1\nline2\nline3\nline4\nline5\nline6")
+            else:
+                # Interpreter without universal newline support
+                self.assertEqual(stdout,
+                                 "line1\nline2\rline3\r\nline4\r\nline5\nline6")
+        finally:
+            p.stdout.close()
+
+    def test_universal2(self):
+        p = subprocess.Popen([sys.executable, "-c",
+                          'import sys,os;' + SETBINARY +
+                          'sys.stdout.write("line1\\n");'
+                          'sys.stdout.flush();'
+                          'sys.stdout.write("line2\\r");'
+                          'sys.stdout.flush();'
+                          'sys.stdout.write("line3\\r\\n");'
+                          'sys.stdout.flush();'
+                          'sys.stdout.write("line4\\r\\nline5");'
+                          'sys.stdout.flush();'
+                          'sys.stdout.write("\\nline6");'],
+                         stdout=subprocess.PIPE,
+                         universal_newlines=1)
+        try:
+            stdout = p.stdout.read()
+            if hasattr(file, 'newlines'):
+                # Interpreter with universal newline support
+                self.assertEqual(stdout,
+                                 "line1\nline2\nline3\nline4\nline5\nline6")
+            else:
+                # Interpreter without universal newline support
+                self.assertEqual(stdout,
+                                 "line1\nline2\rline3\r\nline4\r\nline5\nline6")
+        finally:
+            p.stdout.close()
+
 
 if __name__ == '__main__':
     greentest.main()