Skip to content

Commit

Permalink
fix unicode output for python3
Browse files Browse the repository at this point in the history
--HG--
extra : convert_revision : schettino72-20110529121755-vjay0hbpzlk656k0
  • Loading branch information
schettino72 committed May 29, 2011
1 parent e1c0ac4 commit e2c3c02
Show file tree
Hide file tree
Showing 2 changed files with 23 additions and 8 deletions.
9 changes: 7 additions & 2 deletions doit/action.py
Original file line number Diff line number Diff line change
Expand Up @@ -44,19 +44,24 @@ def _print_process_output(self, process, input_, capture, realtime):
"""read 'input_' untill process is terminated
write 'input_' content to 'capture' and 'realtime' streams
"""
if realtime:
if hasattr(realtime, 'encoding'):
encoding = realtime.encoding or 'utf-8'
else:
encoding = 'utf-8'

while True:
# line buffered
try:
line = input_.readline().decode('utf-8')
except:
process.terminate()
input_.read()
raise Exception("got non-unicode output")
raise
# unbuffered ? process.stdout.read(1)
if line:
capture.write(line)
if realtime:
encoding = getattr(realtime, 'encoding') or 'utf-8'
realtime.write(line.encode(encoding))
if not line and process.poll() != None:
break
Expand Down
22 changes: 16 additions & 6 deletions tests/test_action.py
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,7 @@


def create_tempfile():
return tempfile.TemporaryFile('w+')
return tempfile.TemporaryFile('w+b')

def pytest_funcarg__tmpfile(request):
"""crate a temporary file"""
Expand Down Expand Up @@ -108,7 +108,7 @@ def test_noCaptureStderr(self, tmpfile):
action_result = my_action.execute(err=tmpfile)
assert isinstance(action_result, TaskFailed)
tmpfile.seek(0)
got = tmpfile.read()
got = tmpfile.read().decode('utf-8')
assert "err output on failure" == got, repr(got)
assert "err output on failure" == my_action.err, repr(my_action.err)

Expand All @@ -117,7 +117,7 @@ def test_noCaptureStdout(self, tmpfile):
my_action = action.CmdAction("%s hi_stdout hi2" % PROGRAM)
my_action.execute(out=tmpfile)
tmpfile.seek(0)
got = tmpfile.read()
got = tmpfile.read().decode('utf-8')
assert "hi_stdout" == got, repr(got)
assert "hi_stdout" == my_action.out, repr(my_action.out)

Expand Down Expand Up @@ -159,7 +159,17 @@ def test_non_unicode_string(self):

def test_unicode_string(self, tmpfile):
my_action = action.CmdAction("")
unicode_in = StringIO.StringIO(u" 中文 \u2018".encode('utf-8'))
unicode_in = create_tempfile()
unicode_in.write(u" 中文".encode('utf-8'))
unicode_in.seek(0)
my_action._print_process_output(Mock(), unicode_in, Mock(), tmpfile)

def test_unicode_string2(self, tmpfile):
# this \uXXXX has a different behavior!
my_action = action.CmdAction("")
unicode_in = create_tempfile()
unicode_in.write(u" 中文 \u2018".encode('utf-8'))
unicode_in.seek(0)
my_action._print_process_output(Mock(), unicode_in, Mock(), tmpfile)


Expand Down Expand Up @@ -337,7 +347,7 @@ def test_noCaptureStdout(self, capsys):
assert "this is stdout S\n" == got, repr(got)

def test_redirectStderr(self):
tmpfile = create_tempfile()
tmpfile = tempfile.TemporaryFile('w+')
my_action = action.PythonAction(self.write_stderr)
my_action.execute(err=tmpfile)
tmpfile.seek(0)
Expand All @@ -346,7 +356,7 @@ def test_redirectStderr(self):
assert "this is stderr S\n" == got, got

def test_redirectStdout(self):
tmpfile = create_tempfile()
tmpfile = tempfile.TemporaryFile('w+')
my_action = action.PythonAction(self.write_stdout)
my_action.execute(out=tmpfile)
tmpfile.seek(0)
Expand Down

0 comments on commit e2c3c02

Please sign in to comment.