##// END OF EJS Templates
tests: check that procutil.std{out,err}.write() returns correct result...
Manuel Jacob -
r45657:dff20839 default
parent child Browse files
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
197 self._test(
198 TEST_LARGE_WRITE_CHILD_SCRIPT.format(stream=stream),
199 stream,
200 rwpair_generator,
201 check_output,
202 python_args,
203 )
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()
221 self._test(
222 TEST_LARGE_WRITE_CHILD_SCRIPT.format(
223 stream=stream, write_result_fn=repr(write_result_fn)
224 ),
225 stream,
226 rwpair_generator,
227 check_output,
228 python_args,
229 post_child_check=post_child_check,
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