Mercurial > public > mercurial-scm > hg

comparison tests/test-stdio.py @ 45078:a59aab6078eb

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
tests: make subprocess handling reusable for different tests in test-stdio.py
author Manuel Jacob <me@manueljacob.de>
date Tue, 07 Jul 2020 12:13:40 +0200
parents fa270dcbdb55
children 8e04607023e5
comparison
equal deleted inserted replaced
45077:fa270dcbdb55 45078:a59aab6078eb
12 import unittest 12 import unittest
13 13
14 from mercurial import pycompat 14 from mercurial import pycompat
15 15
16 16
17 BUFFERING_CHILD_SCRIPT = r''' 17 TEST_BUFFERING_CHILD_SCRIPT = r'''
18 import os 18 import os
19 19
20 from mercurial import dispatch 20 from mercurial import dispatch
21 from mercurial.utils import procutil 21 from mercurial.utils import procutil
22 22
79 buf.append(s) 79 buf.append(s)
80 return b''.join(buf) 80 return b''.join(buf)
81 81
82 82
83 class TestStdio(unittest.TestCase): 83 class TestStdio(unittest.TestCase):
84 def _test(self, stream, rwpair_generator, expected_output, python_args=[]): 84 def _test(
85 self,
86 child_script,
87 stream,
88 rwpair_generator,
89 check_output,
90 python_args=[],
91 ):
85 assert stream in ('stdout', 'stderr') 92 assert stream in ('stdout', 'stderr')
86 with rwpair_generator() as (stream_receiver, child_stream), open( 93 with rwpair_generator() as (stream_receiver, child_stream), open(
87 os.devnull, 'rb' 94 os.devnull, 'rb'
88 ) as child_stdin: 95 ) as child_stdin:
89 proc = subprocess.Popen( 96 proc = subprocess.Popen(
90 [sys.executable] 97 [sys.executable] + python_args + ['-c', child_script],
91 + python_args
92 + ['-c', BUFFERING_CHILD_SCRIPT.format(stream=stream)],
93 stdin=child_stdin, 98 stdin=child_stdin,
94 stdout=child_stream if stream == 'stdout' else None, 99 stdout=child_stream if stream == 'stdout' else None,
95 stderr=child_stream if stream == 'stderr' else None, 100 stderr=child_stream if stream == 'stderr' else None,
96 ) 101 )
97 try: 102 try:
98 os.close(child_stream) 103 os.close(child_stream)
99 self.assertEqual( 104 check_output(stream_receiver)
100 _readall(stream_receiver, 1024), expected_output
101 )
102 except: # re-raises 105 except: # re-raises
103 proc.terminate() 106 proc.terminate()
104 raise 107 raise
105 finally: 108 finally:
106 retcode = proc.wait() 109 retcode = proc.wait()
107 self.assertEqual(retcode, 0) 110 self.assertEqual(retcode, 0)
108 111
112 def _test_buffering(
113 self, stream, rwpair_generator, expected_output, python_args=[]
114 ):
115 def check_output(stream_receiver):
116 self.assertEqual(_readall(stream_receiver, 1024), expected_output)
117
118 self._test(
119 TEST_BUFFERING_CHILD_SCRIPT.format(stream=stream),
120 stream,
121 rwpair_generator,
122 check_output,
123 python_args,
124 )
125
109 def test_buffering_stdout_pipes(self): 126 def test_buffering_stdout_pipes(self):
110 self._test('stdout', _pipes, FULLY_BUFFERED) 127 self._test_buffering('stdout', _pipes, FULLY_BUFFERED)
111 128
112 def test_buffering_stdout_ptys(self): 129 def test_buffering_stdout_ptys(self):
113 self._test('stdout', _ptys, LINE_BUFFERED) 130 self._test_buffering('stdout', _ptys, LINE_BUFFERED)
114 131
115 def test_buffering_stdout_pipes_unbuffered(self): 132 def test_buffering_stdout_pipes_unbuffered(self):
116 self._test('stdout', _pipes, UNBUFFERED, python_args=['-u']) 133 self._test_buffering('stdout', _pipes, UNBUFFERED, python_args=['-u'])
117 134
118 def test_buffering_stdout_ptys_unbuffered(self): 135 def test_buffering_stdout_ptys_unbuffered(self):
119 self._test('stdout', _ptys, UNBUFFERED, python_args=['-u']) 136 self._test_buffering('stdout', _ptys, UNBUFFERED, python_args=['-u'])
120 137
121 if not pycompat.ispy3 and not pycompat.iswindows: 138 if not pycompat.ispy3 and not pycompat.iswindows:
122 # On Python 2 on non-Windows, we manually open stdout in line-buffered 139 # On Python 2 on non-Windows, we manually open stdout in line-buffered
123 # mode if connected to a TTY. We should check if Python was configured 140 # mode if connected to a TTY. We should check if Python was configured
124 # to use unbuffered stdout, but it's hard to do that. 141 # to use unbuffered stdout, but it's hard to do that.