mirrors/cpython
1
0
Fork 0
mirror of https://github.com/python/cpython.git synced 2025-09-26 18:29:57 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 10

Make test_subprocess work. Fix universal newlines in io.py.

Browse source
This commit is contained in:
Guido van Rossum 2007-05-24 04:05:35 +00:00
parent c126e8aae3
commit fa0054aa73
3 changed files with 31 additions and 39 deletions
Download patch file Download diff file Expand all files Collapse all files

20
Lib/io.py
View file

@ -1114,14 +1114,6 @@ class TextIOWrapper(TextIOBase):
self._decoder = decoder self._decoder = decoder
return orig_pos return orig_pos
def _simplify(self, u):
# XXX Hack until str/unicode unification: return str instead
# of unicode if it's all ASCII
try:
return str(u)
except UnicodeEncodeError:
return u
def read(self, n=None): def read(self, n=None):
if n is None: if n is None:
n = -1 n = -1
@ -1131,7 +1123,7 @@ class TextIOWrapper(TextIOBase):
res += decoder.decode(self.buffer.read(), True) res += decoder.decode(self.buffer.read(), True)
self._pending = "" self._pending = ""
self._snapshot = None self._snapshot = None
return self._simplify(res) return res.replace("\r\n", "\n")
else: else:
while len(res) < n: while len(res) < n:
readahead, pending = self._read_chunk() readahead, pending = self._read_chunk()
@ -1139,7 +1131,7 @@ class TextIOWrapper(TextIOBase):
if not readahead: if not readahead:
break break
self._pending = res[n:] self._pending = res[n:]
return self._simplify(res[:n]) return res[:n].replace("\r\n", "\n")
def __next__(self): def __next__(self):
self._telling = False self._telling = False
@ -1155,9 +1147,9 @@ class TextIOWrapper(TextIOBase):
# XXX Hack to support limit argument, for backwards compatibility # XXX Hack to support limit argument, for backwards compatibility
line = self.readline() line = self.readline()
if len(line) <= limit: if len(line) <= limit:
return self._simplify(line) return line
line, self._pending = line[:limit], line[limit:] + self._pending line, self._pending = line[:limit], line[limit:] + self._pending
return self._simplify(line) return line
line = self._pending line = self._pending
start = 0 start = 0
@ -1210,9 +1202,9 @@ class TextIOWrapper(TextIOBase):
# XXX Update self.newlines here if we want to support that # XXX Update self.newlines here if we want to support that
if self._fix_newlines and ending not in ("\n", ""): if self._fix_newlines and ending not in ("\n", ""):
return self._simplify(line[:endpos] + "\n") return line[:endpos] + "\n"
else: else:
return self._simplify(line[:nextpos]) return line[:nextpos]
class StringIO(TextIOWrapper): class StringIO(TextIOWrapper):

26
Lib/subprocess.py
View file

@ -286,6 +286,7 @@ Popen(["/bin/mycmd", "myarg"], env={"PATH": "/usr/bin"})
import sys import sys
mswindows = (sys.platform == "win32") mswindows = (sys.platform == "win32")
import io
import os import os
import traceback import traceback
@ -542,23 +543,23 @@ class Popen(object):
if bufsize == 0: if bufsize == 0:
bufsize = 1 # Nearly unbuffered (XXX for now) bufsize = 1 # Nearly unbuffered (XXX for now)
if p2cwrite is not None: if p2cwrite is not None:
self.stdin = os.fdopen(p2cwrite, 'wb', bufsize) self.stdin = io.open(p2cwrite, 'wb', bufsize)
if self.universal_newlines:
self.stdin = io.TextIOWrapper(self.stdin)
if c2pread is not None: if c2pread is not None:
self.stdout = io.open(c2pread, 'rb', bufsize)
if universal_newlines: if universal_newlines:
self.stdout = os.fdopen(c2pread, 'rU', bufsize) self.stdout = io.TextIOWrapper(self.stdout)
else:
self.stdout = os.fdopen(c2pread, 'rb', bufsize)
if errread is not None: if errread is not None:
self.stderr = io.open(errread, 'rb', bufsize)
if universal_newlines: if universal_newlines:
self.stderr = os.fdopen(errread, 'rU', bufsize) self.stderr = io.TextIOWrapper(self.stderr)
else:
self.stderr = os.fdopen(errread, 'rb', bufsize)
def _translate_newlines(self, data): def _translate_newlines(self, data):
data = data.replace("\r\n", "\n") data = data.replace("\r\n", "\n")
data = data.replace("\r", "\n") data = data.replace("\r", "\n")
return data return str(data)
def __del__(self, sys=sys): def __del__(self, sys=sys):
@ -833,9 +834,9 @@ class Popen(object):
# impossible to combine with select (unless forcing no # impossible to combine with select (unless forcing no
# buffering). # buffering).
if self.universal_newlines: if self.universal_newlines:
if stdout: if stdout is not None:
stdout = self._translate_newlines(stdout) stdout = self._translate_newlines(stdout)
if stderr: if stderr is not None:
stderr = self._translate_newlines(stderr) stderr = self._translate_newlines(stderr)
self.wait() self.wait()
@ -1009,7 +1010,6 @@ class Popen(object):
if data: if data:
os.waitpid(self.pid, 0) os.waitpid(self.pid, 0)
child_exception = pickle.loads(data) child_exception = pickle.loads(data)
print("exc:", child_exception)
raise child_exception raise child_exception
@ -1108,9 +1108,9 @@ class Popen(object):
# impossible to combine with select (unless forcing no # impossible to combine with select (unless forcing no
# buffering). # buffering).
if self.universal_newlines: if self.universal_newlines:
if stdout: if stdout is not None:
stdout = self._translate_newlines(stdout) stdout = self._translate_newlines(stdout)
if stderr: if stderr is not None:
stderr = self._translate_newlines(stderr) stderr = self._translate_newlines(stderr)
self.wait() self.wait()

22
Lib/test/test_subprocess.py
View file

@ -151,7 +151,7 @@ class ProcessTestCase(unittest.TestCase):
p = subprocess.Popen([sys.executable, "-c", p = subprocess.Popen([sys.executable, "-c",
'import sys; sys.stdout.write("orange")'], 'import sys; sys.stdout.write("orange")'],
stdout=subprocess.PIPE) stdout=subprocess.PIPE)
self.assertEqual(p.stdout.read(), "orange") self.assertEqual(p.stdout.read(), b"orange")
def test_stdout_filedes(self): def test_stdout_filedes(self):
# stdout is set to open file descriptor # stdout is set to open file descriptor
@ -172,7 +172,7 @@ class ProcessTestCase(unittest.TestCase):
stdout=tf) stdout=tf)
p.wait() p.wait()
tf.seek(0) tf.seek(0)
self.assertEqual(tf.read(), "orange") self.assertEqual(tf.read(), b"orange")
def test_stderr_pipe(self): def test_stderr_pipe(self):
# stderr redirection # stderr redirection
@ -264,7 +264,7 @@ class ProcessTestCase(unittest.TestCase):
'sys.stdout.write(os.getenv("FRUIT"))'], 'sys.stdout.write(os.getenv("FRUIT"))'],
stdout=subprocess.PIPE, stdout=subprocess.PIPE,
env=newenv) env=newenv)
self.assertEqual(p.stdout.read(), "orange") self.assertEqual(p.stdout.read(), b"orange")
def test_communicate_stdin(self): def test_communicate_stdin(self):
p = subprocess.Popen([sys.executable, "-c", p = subprocess.Popen([sys.executable, "-c",
@ -278,7 +278,7 @@ class ProcessTestCase(unittest.TestCase):
'import sys; sys.stdout.write("pineapple")'], 'import sys; sys.stdout.write("pineapple")'],
stdout=subprocess.PIPE) stdout=subprocess.PIPE)
(stdout, stderr) = p.communicate() (stdout, stderr) = p.communicate()
self.assertEqual(stdout, "pineapple") self.assertEqual(stdout, b"pineapple")
self.assertEqual(stderr, None) self.assertEqual(stderr, None)
def test_communicate_stderr(self): def test_communicate_stderr(self):
@ -353,7 +353,7 @@ class ProcessTestCase(unittest.TestCase):
'import sys,os;' + SETBINARY + 'import sys,os;' + SETBINARY +
'sys.stdout.write("line1\\n");' 'sys.stdout.write("line1\\n");'
'sys.stdout.flush();' 'sys.stdout.flush();'
'sys.stdout.write("line2\\r");' 'sys.stdout.write("line2\\n");'
'sys.stdout.flush();' 'sys.stdout.flush();'
'sys.stdout.write("line3\\r\\n");' 'sys.stdout.write("line3\\r\\n");'
'sys.stdout.flush();' 'sys.stdout.flush();'
@ -373,7 +373,7 @@ class ProcessTestCase(unittest.TestCase):
'import sys,os;' + SETBINARY + 'import sys,os;' + SETBINARY +
'sys.stdout.write("line1\\n");' 'sys.stdout.write("line1\\n");'
'sys.stdout.flush();' 'sys.stdout.flush();'
'sys.stdout.write("line2\\r");' 'sys.stdout.write("line2\\n");'
'sys.stdout.flush();' 'sys.stdout.flush();'
'sys.stdout.write("line3\\r\\n");' 'sys.stdout.write("line3\\r\\n");'
'sys.stdout.flush();' 'sys.stdout.flush();'
@ -385,7 +385,7 @@ class ProcessTestCase(unittest.TestCase):
stdout=subprocess.PIPE, stderr=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
universal_newlines=1) universal_newlines=1)
(stdout, stderr) = p.communicate() (stdout, stderr) = p.communicate()
self.assertEqual(stdout, b"line1\nline2\nline3\nline4\nline5\nline6") self.assertEqual(stdout, "line1\nline2\nline3\nline4\nline5\nline6")
def test_no_leaking(self): def test_no_leaking(self):
# Make sure we leak no resources # Make sure we leak no resources
@ -460,7 +460,7 @@ class ProcessTestCase(unittest.TestCase):
# #
if not mswindows: if not mswindows:
def test_exceptions(self): def test_exceptions(self):
# catched & re-raised exceptions # caught & re-raised exceptions
try: try:
p = subprocess.Popen([sys.executable, "-c", ""], p = subprocess.Popen([sys.executable, "-c", ""],
cwd="/this/path/does/not/exist") cwd="/this/path/does/not/exist")
@ -511,7 +511,7 @@ class ProcessTestCase(unittest.TestCase):
'sys.stdout.write(os.getenv("FRUIT"))'], 'sys.stdout.write(os.getenv("FRUIT"))'],
stdout=subprocess.PIPE, stdout=subprocess.PIPE,
preexec_fn=lambda: os.putenv("FRUIT", "apple")) preexec_fn=lambda: os.putenv("FRUIT", "apple"))
self.assertEqual(p.stdout.read(), "apple") self.assertEqual(p.stdout.read(), b"apple")
def test_args_string(self): def test_args_string(self):
# args is a string # args is a string
@ -544,7 +544,7 @@ class ProcessTestCase(unittest.TestCase):
p = subprocess.Popen(["echo $FRUIT"], shell=1, p = subprocess.Popen(["echo $FRUIT"], shell=1,
stdout=subprocess.PIPE, stdout=subprocess.PIPE,
env=newenv) env=newenv)
self.assertEqual(p.stdout.read().strip(), "apple") self.assertEqual(p.stdout.read().strip(b" \t\r\n\f"), b"apple")
def test_shell_string(self): def test_shell_string(self):
# Run command through the shell (string) # Run command through the shell (string)
@ -553,7 +553,7 @@ class ProcessTestCase(unittest.TestCase):
p = subprocess.Popen("echo $FRUIT", shell=1, p = subprocess.Popen("echo $FRUIT", shell=1,
stdout=subprocess.PIPE, stdout=subprocess.PIPE,
env=newenv) env=newenv)
self.assertEqual(p.stdout.read().strip(), "apple") self.assertEqual(p.stdout.read().strip(b" \t\r\n\f"), b"apple")
def test_call_string(self): def test_call_string(self):
# call() function with string argument on UNIX # call() function with string argument on UNIX