[PATCH 1 of 6] tests: add tests for buffering behavior of mercurial.procutil.stdout
Manuel Jacob
me at manueljacob.de
Sun Jul 5 10:50:46 UTC 2020
On 2020-07-05 04:51, Manuel Jacob wrote:
> # HG changeset patch
> # User Manuel Jacob <me at manueljacob.de>
> # Date 1593852424 -7200
> # Sat Jul 04 10:47:04 2020 +0200
> # Node ID cbec874a8b0f9441469cbd5e79db6dcd83b4c4d4
> # Parent 5e758cf2e5c1c2c9b8d11c5dc227b35797ec83a6
> # EXP-Topic stdio
> tests: add tests for buffering behavior of mercurial.procutil.stdout
>
> diff --git a/tests/test-stdio.py b/tests/test-stdio.py
> new file mode 100755
> --- /dev/null
> +++ b/tests/test-stdio.py
> @@ -0,0 +1,104 @@
> +#!/usr/bin/env python
> +"""
> +Tests the buffering behavior of stdio streams in
> `mercurial.utils.procutil`.
> +"""
> +from __future__ import absolute_import
> +
> +import contextlib
> +import os
> +import subprocess
> +import sys
> +import unittest
> +
> +from mercurial import pycompat
> +
> +
> +CHILD_PROCESS = r'''
> +import os
> +
> +from mercurial import dispatch
> +from mercurial.utils import procutil
> +
> +dispatch.initstdio()
> +procutil.stdout.write(b'aaa')
> +os.write(procutil.stdout.fileno(), b'[written aaa]')
> +procutil.stdout.write(b'bbb\n')
> +os.write(procutil.stdout.fileno(), b'[written bbb\\n]')
> +'''
> +UNBUFFERED = b'aaa[written aaa]bbb\n[written bbb\\n]'
> +LINE_BUFFERED = b'[written aaa]aaabbb\n[written bbb\\n]'
> +FULLY_BUFFERED = b'[written aaa][written bbb\\n]aaabbb\n'
> +
> +
> + at contextlib.contextmanager
> +def _closing(fds):
> + try:
> + yield
> + finally:
> + for fd in fds:
> + try:
> + os.close(fd)
> + except EnvironmentError:
> + pass
> +
> +
> + at contextlib.contextmanager
> +def _pipes():
> + rwpair = os.pipe()
> + with _closing(rwpair):
> + yield rwpair
> +
> +
> + at contextlib.contextmanager
> +def _ptys():
> + if pycompat.iswindows:
> + raise unittest.SkipTest("PTYs are not supported on Windows")
> + import pty
> + import tty
I forgot to include a blank line here to please black.
> + rwpair = pty.openpty()
> + with _closing(rwpair):
> + tty.setraw(rwpair[0])
> + yield rwpair
> +
> +
> +class TestStdout(unittest.TestCase):
> + def _test(self, rwpair_generator, expected_output,
> python_args=[]):
> + with rwpair_generator() as (stdout_receiver, child_stdout),
> open(
> + os.devnull, 'rb'
> + ) as child_stdin:
> + proc = subprocess.Popen(
> + [sys.executable] + python_args + ['-c',
> CHILD_PROCESS],
> + stdin=child_stdin,
> + stdout=child_stdout,
> + stderr=None,
> + )
> + retcode = proc.wait()
> + self.assertEqual(retcode, 0)
> + self.assertEqual(os.read(stdout_receiver, 1024),
> expected_output)
> +
> + def test_stdout_pipes(self):
> + self._test(_pipes, FULLY_BUFFERED)
> +
> + def test_stdout_ptys(self):
> + self._test(_ptys, LINE_BUFFERED)
> +
> + def test_stdout_pipes_unbuffered(self):
> + self._test(_pipes, UNBUFFERED, python_args=['-u'])
> +
> + def test_stdout_ptys_unbuffered(self):
> + self._test(_ptys, UNBUFFERED, python_args=['-u'])
> +
> + # On Windows, test_stdout_ptys wouldn't pass, but it's skipped
> anyway.
> + if not pycompat.ispy3 and not pycompat.iswindows:
> + # On Python 2 on non-Windows, we manually open stdout in
> line-buffered
> + # mode if connected to a TTY. We should check if Python was
> configured
> + # to use unbuffered stdout, but it's hard to do that.
> + test_stdout_ptys_unbuffered = unittest.expectedFailure(
> + test_stdout_ptys_unbuffered
> + )
> +
> +
> +if __name__ == '__main__':
> + import silenttestrunner
> +
> + silenttestrunner.main(__name__)
>
> _______________________________________________
> Mercurial-devel mailing list
> Mercurial-devel at mercurial-scm.org
> https://www.mercurial-scm.org/mailman/listinfo/mercurial-devel
More information about the Mercurial-devel
mailing list