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