##// 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 import signal
10 import signal
11 import subprocess
11 import subprocess
12 import sys
12 import sys
13 import tempfile
13 import unittest
14 import unittest
14
15
15 from mercurial import pycompat
16 from mercurial import pycompat
@@ -41,7 +42,9 b' from mercurial.utils import procutil'
41
42
42 signal.signal(signal.SIGINT, lambda *x: None)
43 signal.signal(signal.SIGINT, lambda *x: None)
43 dispatch.initstdio()
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 rwpair_generator,
112 rwpair_generator,
110 check_output,
113 check_output,
111 python_args=[],
114 python_args=[],
115 post_child_check=None,
112 ):
116 ):
113 assert stream in ('stdout', 'stderr')
117 assert stream in ('stdout', 'stderr')
114 with rwpair_generator() as (stream_receiver, child_stream), open(
118 with rwpair_generator() as (stream_receiver, child_stream), open(
@@ -130,6 +134,8 b' class TestStdio(unittest.TestCase):'
130 finally:
134 finally:
131 retcode = proc.wait()
135 retcode = proc.wait()
132 self.assertEqual(retcode, 0)
136 self.assertEqual(retcode, 0)
137 if post_child_check is not None:
138 post_child_check()
133
139
134 def _test_buffering(
140 def _test_buffering(
135 self, stream, rwpair_generator, expected_output, python_args=[]
141 self, stream, rwpair_generator, expected_output, python_args=[]
@@ -194,13 +200,39 b' class TestStdio(unittest.TestCase):'
194 _readall(stream_receiver, 131072, buf), b'x' * 1048576
200 _readall(stream_receiver, 131072, buf), b'x' * 1048576
195 )
201 )
196
202
197 self._test(
203 def post_child_check():
198 TEST_LARGE_WRITE_CHILD_SCRIPT.format(stream=stream),
204 with open(write_result_fn, 'r') as write_result_f:
199 stream,
205 write_result_str = write_result_f.read()
200 rwpair_generator,
206 if pycompat.ispy3:
201 check_output,
207 # On Python 3, we test that the correct number of bytes is
202 python_args,
208 # claimed to have been written.
203 )
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 def test_large_write_stdout_devnull(self):
237 def test_large_write_stdout_devnull(self):
206 self._test_large_write('stdout', _devnull)
238 self._test_large_write('stdout', _devnull)
General Comments 0
You need to be logged in to leave comments. Login now