Patchwork [4,of,5] tests: make subprocess handling reusable for different tests in test-stdio.py

login
register
mail settings
Submitter Manuel Jacob
Date July 10, 2020, 4:46 a.m.
Message ID <034feda7f6afcb0ae973.1594356401@tmp>
Download mbox | patch
Permalink /patch/46681/
State Accepted
Headers show

Comments

Manuel Jacob - July 10, 2020, 4:46 a.m.
# HG changeset patch
# User Manuel Jacob <me@manueljacob.de>
# Date 1594116820 -7200
#      Tue Jul 07 12:13:40 2020 +0200
# Node ID 034feda7f6afcb0ae973569207a8268487793962
# Parent  8d2f25abde563d51bfc6296ba4136b9e7a86ecd8
# EXP-Topic stdio
tests: make subprocess handling reusable for different tests in test-stdio.py

Patch

diff --git a/tests/test-stdio.py b/tests/test-stdio.py
--- a/tests/test-stdio.py
+++ b/tests/test-stdio.py
@@ -14,7 +14,7 @@ 
 from mercurial import pycompat
 
 
-BUFFERING_CHILD_SCRIPT = r'''
+TEST_BUFFERING_CHILD_SCRIPT = r'''
 import os
 
 from mercurial import dispatch
@@ -81,24 +81,27 @@ 
 
 
 class TestStdio(unittest.TestCase):
-    def _test(self, stream, rwpair_generator, expected_output, python_args=[]):
+    def _test(
+        self,
+        child_script,
+        stream,
+        rwpair_generator,
+        check_output,
+        python_args=[],
+    ):
         assert stream in ('stdout', 'stderr')
         with rwpair_generator() as (stream_receiver, child_stream), open(
             os.devnull, 'rb'
         ) as child_stdin:
             proc = subprocess.Popen(
-                [sys.executable]
-                + python_args
-                + ['-c', BUFFERING_CHILD_SCRIPT.format(stream=stream)],
+                [sys.executable] + python_args + ['-c', child_script],
                 stdin=child_stdin,
                 stdout=child_stream if stream == 'stdout' else None,
                 stderr=child_stream if stream == 'stderr' else None,
             )
             try:
                 os.close(child_stream)
-                self.assertEqual(
-                    _readall(stream_receiver, 1024), expected_output
-                )
+                check_output(stream_receiver)
             except:  # re-raises
                 proc.terminate()
                 raise
@@ -106,17 +109,31 @@ 
                 retcode = proc.wait()
             self.assertEqual(retcode, 0)
 
+    def _test_buffering(
+        self, stream, rwpair_generator, expected_output, python_args=[]
+    ):
+        def check_output(stream_receiver):
+            self.assertEqual(_readall(stream_receiver, 1024), expected_output)
+
+        self._test(
+            TEST_BUFFERING_CHILD_SCRIPT.format(stream=stream),
+            stream,
+            rwpair_generator,
+            check_output,
+            python_args,
+        )
+
     def test_buffering_stdout_pipes(self):
-        self._test('stdout', _pipes, FULLY_BUFFERED)
+        self._test_buffering('stdout', _pipes, FULLY_BUFFERED)
 
     def test_buffering_stdout_ptys(self):
-        self._test('stdout', _ptys, LINE_BUFFERED)
+        self._test_buffering('stdout', _ptys, LINE_BUFFERED)
 
     def test_buffering_stdout_pipes_unbuffered(self):
-        self._test('stdout', _pipes, UNBUFFERED, python_args=['-u'])
+        self._test_buffering('stdout', _pipes, UNBUFFERED, python_args=['-u'])
 
     def test_buffering_stdout_ptys_unbuffered(self):
-        self._test('stdout', _ptys, UNBUFFERED, python_args=['-u'])
+        self._test_buffering('stdout', _ptys, UNBUFFERED, python_args=['-u'])
 
     if not pycompat.ispy3 and not pycompat.iswindows:
         # On Python 2 on non-Windows, we manually open stdout in line-buffered
@@ -127,16 +144,16 @@ 
         )
 
     def test_buffering_stderr_pipes(self):
-        self._test('stderr', _pipes, UNBUFFERED)
+        self._test_buffering('stderr', _pipes, UNBUFFERED)
 
     def test_buffering_stderr_ptys(self):
-        self._test('stderr', _ptys, UNBUFFERED)
+        self._test_buffering('stderr', _ptys, UNBUFFERED)
 
     def test_buffering_stderr_pipes_unbuffered(self):
-        self._test('stderr', _pipes, UNBUFFERED, python_args=['-u'])
+        self._test_buffering('stderr', _pipes, UNBUFFERED, python_args=['-u'])
 
     def test_buffering_stderr_ptys_unbuffered(self):
-        self._test('stderr', _ptys, UNBUFFERED, python_args=['-u'])
+        self._test_buffering('stderr', _ptys, UNBUFFERED, python_args=['-u'])
 
 
 if __name__ == '__main__':