Mercurial > public > mercurial-scm > hg

comparison tests/test-stdio.py @ 45097:dff208398ede

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
tests: check that procutil.std{out,err}.write() returns correct result On Windows, we currently don?t fully test the case when the stream is connected to a TTY, but we test the child process side by connecting them to NUL, which is recognized as a TTY by Python. To make the large write test a bit more useful besides checking that it doesn?t crash, we can check that the write() method returns the correct result.
author Manuel Jacob <me@manueljacob.de>
date Thu, 09 Jul 2020 12:52:04 +0200
parents e9e452eafbfb
children eb26a9cf7821
comparison
equal deleted inserted replaced
45096:e9e452eafbfb 45097:dff208398ede
8 import errno 8 import errno
9 import os 9 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
16 17
17 18
39 from mercurial import dispatch 40 from mercurial import dispatch
40 from mercurial.utils import procutil 41 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
48 @contextlib.contextmanager 51 @contextlib.contextmanager
49 def _closing(fds): 52 def _closing(fds):
107 child_script, 110 child_script,
108 stream, 111 stream,
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(
115 os.devnull, 'rb' 119 os.devnull, 'rb'
116 ) as child_stdin: 120 ) as child_stdin:
128 proc.terminate() 132 proc.terminate()
129 raise 133 raise
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=[]
136 ): 142 ):
137 def check_output(stream_receiver, proc): 143 def check_output(stream_receiver, proc):
192 buf = [] 198 buf = []
193 self.assertEqual( 199 self.assertEqual(
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)
207 239
208 def test_large_write_stdout_pipes(self): 240 def test_large_write_stdout_pipes(self):