##// END OF EJS Templates
tests: make subprocess handling reusable for different tests in test-stdio.py
Manuel Jacob -
r45638:a59aab60 default
parent child Browse files
Show More
@@ -14,7 +14,7 b' import unittest'
14 14 from mercurial import pycompat
15 15
16 16
17 BUFFERING_CHILD_SCRIPT = r'''
17 TEST_BUFFERING_CHILD_SCRIPT = r'''
18 18 import os
19 19
20 20 from mercurial import dispatch
@@ -81,24 +81,27 b' def _readall(fd, buffer_size):'
81 81
82 82
83 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 92 assert stream in ('stdout', 'stderr')
86 93 with rwpair_generator() as (stream_receiver, child_stream), open(
87 94 os.devnull, 'rb'
88 95 ) as child_stdin:
89 96 proc = subprocess.Popen(
90 [sys.executable]
91 + python_args
92 + ['-c', BUFFERING_CHILD_SCRIPT.format(stream=stream)],
97 [sys.executable] + python_args + ['-c', child_script],
93 98 stdin=child_stdin,
94 99 stdout=child_stream if stream == 'stdout' else None,
95 100 stderr=child_stream if stream == 'stderr' else None,
96 101 )
97 102 try:
98 103 os.close(child_stream)
99 self.assertEqual(
100 _readall(stream_receiver, 1024), expected_output
101 )
104 check_output(stream_receiver)
102 105 except: # re-raises
103 106 proc.terminate()
104 107 raise
@@ -106,17 +109,31 b' class TestStdio(unittest.TestCase):'
106 109 retcode = proc.wait()
107 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 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 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 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 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 138 if not pycompat.ispy3 and not pycompat.iswindows:
122 139 # On Python 2 on non-Windows, we manually open stdout in line-buffered
General Comments 0
You need to be logged in to leave comments. Login now