Show More
@@ -34,6 +34,7 b" FULLY_BUFFERED = b'[written aaa][written" | |||
|
34 | 34 | |
|
35 | 35 | |
|
36 | 36 | TEST_LARGE_WRITE_CHILD_SCRIPT = r''' |
|
37 | import os | |
|
37 | 38 | import signal |
|
38 | 39 | import sys |
|
39 | 40 | |
@@ -43,7 +44,10 b' from mercurial.utils import procutil' | |||
|
43 | 44 | signal.signal(signal.SIGINT, lambda *x: None) |
|
44 | 45 | dispatch.initstdio() |
|
45 | 46 | write_result = procutil.{stream}.write(b'x' * 1048576) |
|
46 | with open({write_result_fn}, 'w') as write_result_f: | |
|
47 | with os.fdopen( | |
|
48 | os.open({write_result_fn!r}, os.O_WRONLY | getattr(os, 'O_TEMPORARY', 0)), | |
|
49 | 'w', | |
|
50 | ) as write_result_f: | |
|
47 | 51 | write_result_f.write(str(write_result)) |
|
48 | 52 | ''' |
|
49 | 53 | |
@@ -201,8 +205,7 b' class TestStdio(unittest.TestCase):' | |||
|
201 | 205 | ) |
|
202 | 206 | |
|
203 | 207 | def post_child_check(): |
|
204 |
w |
|
|
205 | write_result_str = write_result_f.read() | |
|
208 | write_result_str = write_result_f.read() | |
|
206 | 209 | if pycompat.ispy3: |
|
207 | 210 | # On Python 3, we test that the correct number of bytes is |
|
208 | 211 | # claimed to have been written. |
@@ -213,14 +216,10 b' class TestStdio(unittest.TestCase):' | |||
|
213 | 216 | expected_write_result_str = 'None' |
|
214 | 217 | self.assertEqual(write_result_str, expected_write_result_str) |
|
215 | 218 | |
|
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() | |
|
219 | with tempfile.NamedTemporaryFile('r') as write_result_f: | |
|
221 | 220 | self._test( |
|
222 | 221 | TEST_LARGE_WRITE_CHILD_SCRIPT.format( |
|
223 |
stream=stream, write_result_fn= |
|
|
222 | stream=stream, write_result_fn=write_result_f.name | |
|
224 | 223 | ), |
|
225 | 224 | stream, |
|
226 | 225 | rwpair_generator, |
@@ -228,11 +227,6 b' class TestStdio(unittest.TestCase):' | |||
|
228 | 227 | python_args, |
|
229 | 228 | post_child_check=post_child_check, |
|
230 | 229 | ) |
|
231 | finally: | |
|
232 | try: | |
|
233 | os.unlink(write_result_fn) | |
|
234 | except OSError: | |
|
235 | pass | |
|
236 | 230 | |
|
237 | 231 | def test_large_write_stdout_devnull(self): |
|
238 | 232 | self._test_large_write('stdout', _devnull) |
General Comments 0
You need to be logged in to leave comments.
Login now