##// END OF EJS Templates
code: format style and py3 compatability
marcink -
r589:bd50ae6f default
parent child Browse files
Show More
@@ -36,8 +36,7 b' def environ():'
36
36
37
37
38 def _get_python_args(script):
38 def _get_python_args(script):
39 return [sys.executable, '-c',
39 return [sys.executable, '-c', 'import sys; import time; import shutil; ' + script]
40 'import sys; import time; import shutil; ' + script]
41
40
42
41
43 def test_raise_exception_on_non_zero_return_code(environ):
42 def test_raise_exception_on_non_zero_return_code(environ):
@@ -48,8 +47,11 b' def test_raise_exception_on_non_zero_ret'
48
47
49 def test_does_not_fail_on_non_zero_return_code(environ):
48 def test_does_not_fail_on_non_zero_return_code(environ):
50 args = _get_python_args('sys.exit(1)')
49 args = _get_python_args('sys.exit(1)')
51 output = ''.join(subprocessio.SubprocessIOChunker(
50 output = ''.join(
52 args, shell=False, fail_on_return_code=False, env=environ))
51 subprocessio.SubprocessIOChunker(
52 args, shell=False, fail_on_return_code=False, env=environ
53 )
54 )
53
55
54 assert output == ''
56 assert output == ''
55
57
@@ -64,49 +66,56 b' def test_raise_exception_on_stderr(envir'
64
66
65 def test_does_not_fail_on_stderr(environ):
67 def test_does_not_fail_on_stderr(environ):
66 args = _get_python_args('sys.stderr.write("X"); time.sleep(1);')
68 args = _get_python_args('sys.stderr.write("X"); time.sleep(1);')
67 output = ''.join(subprocessio.SubprocessIOChunker(
69 output = ''.join(
68 args, shell=False, fail_on_stderr=False, env=environ))
70 subprocessio.SubprocessIOChunker(
71 args, shell=False, fail_on_stderr=False, env=environ
72 )
73 )
69
74
70 assert output == ''
75 assert output == ''
71
76
72
77
73 @pytest.mark.parametrize('size', [1, 10**5])
78 @pytest.mark.parametrize('size', [1, 10 ** 5])
74 def test_output_with_no_input(size, environ):
79 def test_output_with_no_input(size, environ):
75 print type(environ)
80 print(type(environ))
76 data = 'X'
81 data = 'X'
77 args = _get_python_args('sys.stdout.write("%s" * %d)' % (data, size))
82 args = _get_python_args('sys.stdout.write("%s" * %d)' % (data, size))
78 output = ''.join(subprocessio.SubprocessIOChunker(
83 output = ''.join(subprocessio.SubprocessIOChunker(args, shell=False, env=environ))
79 args, shell=False, env=environ))
80
84
81 assert output == data * size
85 assert output == data * size
82
86
83
87
84 @pytest.mark.parametrize('size', [1, 10**5])
88 @pytest.mark.parametrize('size', [1, 10 ** 5])
85 def test_output_with_no_input_does_not_fail(size, environ):
89 def test_output_with_no_input_does_not_fail(size, environ):
86 data = 'X'
90 data = 'X'
87 args = _get_python_args(
91 args = _get_python_args('sys.stdout.write("%s" * %d); sys.exit(1)' % (data, size))
88 'sys.stdout.write("%s" * %d); sys.exit(1)' % (data, size))
92 output = ''.join(
89 output = ''.join(subprocessio.SubprocessIOChunker(
93 subprocessio.SubprocessIOChunker(
90 args, shell=False, fail_on_return_code=False, env=environ))
94 args, shell=False, fail_on_return_code=False, env=environ
95 )
96 )
91
97
92 print len(data * size), len(output)
98 print("{} {}".format(len(data * size), len(output)))
93 assert output == data * size
99 assert output == data * size
94
100
95
101
96 @pytest.mark.parametrize('size', [1, 10**5])
102 @pytest.mark.parametrize('size', [1, 10 ** 5])
97 def test_output_with_input(size, environ):
103 def test_output_with_input(size, environ):
98 data = 'X' * size
104 data = 'X' * size
99 inputstream = io.BytesIO(data)
105 inputstream = io.BytesIO(data)
100 # This acts like the cat command.
106 # This acts like the cat command.
101 args = _get_python_args('shutil.copyfileobj(sys.stdin, sys.stdout)')
107 args = _get_python_args('shutil.copyfileobj(sys.stdin, sys.stdout)')
102 output = ''.join(subprocessio.SubprocessIOChunker(
108 output = ''.join(
103 args, shell=False, inputstream=inputstream, env=environ))
109 subprocessio.SubprocessIOChunker(
110 args, shell=False, inputstream=inputstream, env=environ
111 )
112 )
104
113
105 print len(data), len(output)
114 print("{} {}".format(len(data * size), len(output)))
106 assert output == data
115 assert output == data
107
116
108
117
109 @pytest.mark.parametrize('size', [1, 10**5])
118 @pytest.mark.parametrize('size', [1, 10 ** 5])
110 def test_output_with_input_skipping_iterator(size, environ):
119 def test_output_with_input_skipping_iterator(size, environ):
111 data = 'X' * size
120 data = 'X' * size
112 inputstream = io.BytesIO(data)
121 inputstream = io.BytesIO(data)
@@ -115,8 +124,9 b' def test_output_with_input_skipping_iter'
115
124
116 # Note: assigning the chunker makes sure that it is not deleted too early
125 # Note: assigning the chunker makes sure that it is not deleted too early
117 chunker = subprocessio.SubprocessIOChunker(
126 chunker = subprocessio.SubprocessIOChunker(
118 args, shell=False, inputstream=inputstream, env=environ)
127 args, shell=False, inputstream=inputstream, env=environ
128 )
119 output = ''.join(chunker.output)
129 output = ''.join(chunker.output)
120
130
121 print len(data), len(output)
131 print("{} {}".format(len(data * size), len(output)))
122 assert output == data
132 assert output == data
General Comments 0
You need to be logged in to leave comments. Login now