Show More
@@ -10,6 +10,7 b' import os' | |||
|
10 | 10 | import signal |
|
11 | 11 | import subprocess |
|
12 | 12 | import sys |
|
13 | import tempfile | |
|
13 | 14 | import unittest |
|
14 | 15 | |
|
15 | 16 | from mercurial import pycompat |
@@ -41,7 +42,9 b' from mercurial.utils import procutil' | |||
|
41 | 42 | |
|
42 | 43 | signal.signal(signal.SIGINT, lambda *x: None) |
|
43 | 44 | dispatch.initstdio() |
|
44 | procutil.{stream}.write(b'x' * 1048576) | |
|
45 | write_result = procutil.{stream}.write(b'x' * 1048576) | |
|
46 | with open({write_result_fn}, 'w') as write_result_f: | |
|
47 | write_result_f.write(str(write_result)) | |
|
45 | 48 | ''' |
|
46 | 49 | |
|
47 | 50 | |
@@ -109,6 +112,7 b' class TestStdio(unittest.TestCase):' | |||
|
109 | 112 | rwpair_generator, |
|
110 | 113 | check_output, |
|
111 | 114 | python_args=[], |
|
115 | post_child_check=None, | |
|
112 | 116 | ): |
|
113 | 117 | assert stream in ('stdout', 'stderr') |
|
114 | 118 | with rwpair_generator() as (stream_receiver, child_stream), open( |
@@ -130,6 +134,8 b' class TestStdio(unittest.TestCase):' | |||
|
130 | 134 | finally: |
|
131 | 135 | retcode = proc.wait() |
|
132 | 136 | self.assertEqual(retcode, 0) |
|
137 | if post_child_check is not None: | |
|
138 | post_child_check() | |
|
133 | 139 | |
|
134 | 140 | def _test_buffering( |
|
135 | 141 | self, stream, rwpair_generator, expected_output, python_args=[] |
@@ -194,13 +200,39 b' class TestStdio(unittest.TestCase):' | |||
|
194 | 200 | _readall(stream_receiver, 131072, buf), b'x' * 1048576 |
|
195 | 201 | ) |
|
196 | 202 | |
|
203 | def post_child_check(): | |
|
204 | with open(write_result_fn, 'r') as write_result_f: | |
|
205 | write_result_str = write_result_f.read() | |
|
206 | if pycompat.ispy3: | |
|
207 | # On Python 3, we test that the correct number of bytes is | |
|
208 | # claimed to have been written. | |
|
209 | expected_write_result_str = '1048576' | |
|
210 | else: | |
|
211 | # On Python 2, we only check that the large write does not | |
|
212 | # crash. | |
|
213 | expected_write_result_str = 'None' | |
|
214 | self.assertEqual(write_result_str, expected_write_result_str) | |
|
215 | ||
|
216 | try: | |
|
217 | # tempfile.mktemp() is unsafe in general, as a malicious process | |
|
218 | # could create the file before we do. But in tests, we're running | |
|
219 | # in a controlled environment. | |
|
220 | write_result_fn = tempfile.mktemp() | |
|
197 | 221 | self._test( |
|
198 |
TEST_LARGE_WRITE_CHILD_SCRIPT.format( |
|
|
222 | TEST_LARGE_WRITE_CHILD_SCRIPT.format( | |
|
223 | stream=stream, write_result_fn=repr(write_result_fn) | |
|
224 | ), | |
|
199 | 225 | stream, |
|
200 | 226 | rwpair_generator, |
|
201 | 227 | check_output, |
|
202 | 228 | python_args, |
|
229 | post_child_check=post_child_check, | |
|
203 | 230 | ) |
|
231 | finally: | |
|
232 | try: | |
|
233 | os.unlink(write_result_fn) | |
|
234 | except OSError: | |
|
235 | pass | |
|
204 | 236 | |
|
205 | 237 | def test_large_write_stdout_devnull(self): |
|
206 | 238 | self._test_large_write('stdout', _devnull) |
General Comments 0
You need to be logged in to leave comments.
Login now