Codebase list python-procrunner / d65995e
Incremental decoding of input and add a test for valid and invalid UTF-8 input Markus Gerstel 4 years ago
2 changed file(s) with 26 addition(s) and 4 deletion(s). Raw diff Collapse all Expand all
11
22 from __future__ import absolute_import, division, print_function
33
4 import codecs
45 import copy
56 import logging
67 import os
6667 self._buffer = ''
6768 self._print = print_line
6869 self._callback = callback
70 self._decoder = codecs.getincrementaldecoder('utf-8')('replace')
6971 def add(self, data):
7072 '''Add a single character to buffer. If one or more full lines are found,
7173 print them (if desired) and pass to callback function.'''
72 data = data.decode('utf-8')
74 data = self._decoder.decode(data)
75 if not data: return
7376 self._buffer += data
74 if "\n" in data:
77 if '\n' in data:
7578 to_print, remainder = self._buffer.rsplit('\n')
7679 if self._print:
7780 print(to_print)
8083 self._buffer = remainder
8184 def flush(self):
8285 '''Print/send any remaining data to callback function.'''
86 self._buffer += self._decoder.decode(b'', final=True)
8387 if self._buffer:
8488 if self._print:
8589 print(self._buffer)
33 import sys
44
55 import procrunner
6 import pytest
67
78 def test_simple_command_invocation():
89 if os.name == 'nt':
1617 assert result['stdout'] == b'hello' + os.linesep.encode('utf-8')
1718 assert result['stderr'] == b''
1819
19 def test_input_encoding():
20 def test_decode_invalid_utf8_input(capsys):
2021 command = [sys.executable, '-c', 'import sys; sys.stdout.write("".join(chr(x) for x in '
21 '(0x74,0x65,0x73,0x74,0xa0,0x50,0x73,0x74,0x72,0x69,0x6e,0x67,0x0a)'
22 '(0x74,0x65,0x73,0x74,0xa0,0x73,0x74,0x72,0x69,0x6e,0x67,0x0a)'
2223 '))']
2324 result = procrunner.run(command)
25 assert result['exitcode'] == 0
26 assert not result['stderr']
27 assert result['stdout'] == b'test\xa0string\n'
28 out, err = capsys.readouterr()
29 assert out == u'test\ufffdstring\n'
30 assert err == u''
2431
32 def test_running_wget(tmpdir):
33 tmpdir.chdir()
34 command = ['wget', 'https://www.google.com', '-O', '-']
35 try:
36 result = procrunner.run(command)
37 except OSError as e:
38 if e.errno == 2:
39 pytest.skip('wget not available')
40 raise
2541 assert result['exitcode'] == 0
42 assert b'http' in result['stderr']
43 assert b'google' in result['stdout']