[Python-3000-checkins] r55364 - in python/branches/py3k-struni/Lib: subprocess.py test/test_subprocess.py

[guido.van.rossum]

Author: guido.van.rossum
Date: Wed May 16 01:18:55 2007
New Revision: 55364

Modified:
   python/branches/py3k-struni/Lib/subprocess.py
   python/branches/py3k-struni/Lib/test/test_subprocess.py
Log:
Make test_subprocess pass.  The subprocess module now reads and writes
bytes instead of strings.  We'll see how long that lasts.


Modified: python/branches/py3k-struni/Lib/subprocess.py
==============================================================================
--- python/branches/py3k-struni/Lib/subprocess.py	(original)
+++ python/branches/py3k-struni/Lib/subprocess.py	Wed May 16 01:18:55 2007
@@ -830,7 +830,7 @@
             # object do the translation: It is based on stdio, which is
             # impossible to combine with select (unless forcing no
             # buffering).
-            if self.universal_newlines and hasattr(file, 'newlines'):
+            if self.universal_newlines:
                 if stdout:
                     stdout = self._translate_newlines(stdout)
                 if stderr:
@@ -1044,6 +1044,8 @@
 
 
         def _communicate(self, input):
+            if isinstance(input, str): # Unicode
+                input = input.encode("utf-8") # XXX What else?
             read_set = []
             write_set = []
             stdout = None # Return
@@ -1080,29 +1082,29 @@
 
                 if self.stdout in rlist:
                     data = os.read(self.stdout.fileno(), 1024)
-                    if data == "":
+                    if not data:
                         self.stdout.close()
                         read_set.remove(self.stdout)
                     stdout.append(data)
 
                 if self.stderr in rlist:
                     data = os.read(self.stderr.fileno(), 1024)
-                    if data == "":
+                    if not data:
                         self.stderr.close()
                         read_set.remove(self.stderr)
                     stderr.append(data)
 
             # All data exchanged.  Translate lists into strings.
             if stdout is not None:
-                stdout = ''.join(stdout)
+                stdout = b''.join(stdout)
             if stderr is not None:
-                stderr = ''.join(stderr)
+                stderr = b''.join(stderr)
 
             # Translate newlines, if requested.  We cannot let the file
             # object do the translation: It is based on stdio, which is
             # impossible to combine with select (unless forcing no
             # buffering).
-            if self.universal_newlines and hasattr(file, 'newlines'):
+            if self.universal_newlines:
                 if stdout:
                     stdout = self._translate_newlines(stdout)
                 if stderr:

Modified: python/branches/py3k-struni/Lib/test/test_subprocess.py
==============================================================================
--- python/branches/py3k-struni/Lib/test/test_subprocess.py	(original)
+++ python/branches/py3k-struni/Lib/test/test_subprocess.py	Wed May 16 01:18:55 2007
@@ -24,7 +24,7 @@
 # shutdown time.  That frustrates tests trying to check stderr produced
 # from a spawned Python process.
 def remove_stderr_debug_decorations(stderr):
-    return re.sub(r"\[\d+ refs\]\r?\n?$", "", stderr)
+    return re.sub(r"\[\d+ refs\]\r?\n?$", "", str8(stderr))
 
 class ProcessTestCase(unittest.TestCase):
     def setUp(self):
@@ -162,7 +162,7 @@
                          stdout=d)
         p.wait()
         os.lseek(d, 0, 0)
-        self.assertEqual(os.read(d, 1024), "orange")
+        self.assertEqual(os.read(d, 1024), b"orange")
 
     def test_stdout_fileobj(self):
         # stdout is set to open file object
@@ -300,7 +300,7 @@
                          stdout=subprocess.PIPE,
                          stderr=subprocess.PIPE)
         (stdout, stderr) = p.communicate("banana")
-        self.assertEqual(stdout, "banana")
+        self.assertEqual(stdout, b"banana")
         self.assertEqual(remove_stderr_debug_decorations(stderr),
                          "pineapple")
 
@@ -331,7 +331,7 @@
                          stdin=subprocess.PIPE,
                          stdout=subprocess.PIPE,
                          stderr=subprocess.PIPE)
-        string_to_write = "abc"*pipe_buf
+        string_to_write = b"abc"*pipe_buf
         (stdout, stderr) = p.communicate(string_to_write)
         self.assertEqual(stdout, string_to_write)
 
@@ -345,7 +345,7 @@
                          stderr=subprocess.PIPE)
         p.stdin.write("banana")
         (stdout, stderr) = p.communicate("split")
-        self.assertEqual(stdout, "bananasplit")
+        self.assertEqual(stdout, b"bananasplit")
         self.assertEqual(remove_stderr_debug_decorations(stderr), "")
 
     def test_universal_newlines(self):
@@ -365,14 +365,7 @@
                          stdout=subprocess.PIPE,
                          universal_newlines=1)
         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")
+        self.assertEqual(stdout, "line1\nline2\nline3\nline4\nline5\nline6")
 
     def test_universal_newlines_communicate(self):
         # universal newlines through communicate()
@@ -392,13 +385,7 @@
                          stdout=subprocess.PIPE, stderr=subprocess.PIPE,
                          universal_newlines=1)
         (stdout, stderr) = p.communicate()
-        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")
+        self.assertEqual(stdout, b"line1\nline2\nline3\nline4\nline5\nline6")
 
     def test_no_leaking(self):
         # Make sure we leak no resources
@@ -414,7 +401,7 @@
                     stdout=subprocess.PIPE,
                     stderr=subprocess.PIPE)
             data = p.communicate("lime")[0]
-            self.assertEqual(data, "lime")
+            self.assertEqual(data, b"lime")
 
 
     def test_list2cmdline(self):