##// 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 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
@@ -81,24 +81,27 b' def _readall(fd, buffer_size):'
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
@@ -106,17 +109,31 b' class TestStdio(unittest.TestCase):'
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
General Comments 0
You need to be logged in to leave comments. Login now