comparison tests/test-stdio.py @ 45097:dff208398ede

tests: check that procutil.std{out,err}.write() returns correct result

On Windows, we currently don’t fully test the case when the stream is connected
to a TTY, but we test the child process side by connecting them to NUL, which
is recognized as a TTY by Python. To make the large write test a bit more
useful besides checking that it doesn’t crash, we can check that the write()
method returns the correct result.

45096:e9e452eafbfb

import errno
import os
import signal
import subprocess
import sys
import unittest

from mercurial import pycompat

from mercurial import dispatch
from mercurial.utils import procutil

signal.signal(signal.SIGINT, lambda *x: None)
dispatch.initstdio()
procutil.{stream}.write(b'x' * 1048576)
'''


@contextlib.contextmanager
def _closing(fds):

        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.terminate()
                raise
            finally:
                retcode = proc.wait()
            self.assertEqual(retcode, 0)

    def _test_buffering(
        self, stream, rwpair_generator, expected_output, python_args=[]
    ):
        def check_output(stream_receiver, proc):

                buf = []
            self.assertEqual(
                _readall(stream_receiver, 131072, buf), b'x' * 1048576
            )

        self._test(
            TEST_LARGE_WRITE_CHILD_SCRIPT.format(stream=stream),
            stream,
            rwpair_generator,
            check_output,
            python_args,
        )

    def test_large_write_stdout_devnull(self):
        self._test_large_write('stdout', _devnull)

    def test_large_write_stdout_pipes(self):

45097:dff208398ede

import errno
import os
import signal
import subprocess
import sys
import tempfile
import unittest

from mercurial import pycompat

from mercurial import dispatch
from mercurial.utils import procutil

signal.signal(signal.SIGINT, lambda *x: None)
dispatch.initstdio()
write_result = procutil.{stream}.write(b'x' * 1048576)
with open({write_result_fn}, 'w') as write_result_f:
    write_result_f.write(str(write_result))
'''


@contextlib.contextmanager
def _closing(fds):

        child_script,
        stream,
        rwpair_generator,
        check_output,
        python_args=[],
        post_child_check=None,
    ):
        assert stream in ('stdout', 'stderr')
        with rwpair_generator() as (stream_receiver, child_stream), open(
            os.devnull, 'rb'
        ) as child_stdin:

                proc.terminate()
                raise
            finally:
                retcode = proc.wait()
            self.assertEqual(retcode, 0)
            if post_child_check is not None:
                post_child_check()

    def _test_buffering(
        self, stream, rwpair_generator, expected_output, python_args=[]
    ):
        def check_output(stream_receiver, proc):

                buf = []
            self.assertEqual(
                _readall(stream_receiver, 131072, buf), b'x' * 1048576
            )

        def post_child_check():
            with open(write_result_fn, 'r') as write_result_f:
                write_result_str = write_result_f.read()
            if pycompat.ispy3:
                # On Python 3, we test that the correct number of bytes is
                # claimed to have been written.
                expected_write_result_str = '1048576'
            else:
                # On Python 2, we only check that the large write does not
                # crash.
                expected_write_result_str = 'None'
            self.assertEqual(write_result_str, expected_write_result_str)

        try:
            # tempfile.mktemp() is unsafe in general, as a malicious process
            # could create the file before we do. But in tests, we're running
            # in a controlled environment.
            write_result_fn = tempfile.mktemp()
            self._test(
                TEST_LARGE_WRITE_CHILD_SCRIPT.format(
                    stream=stream, write_result_fn=repr(write_result_fn)
                ),
                stream,
                rwpair_generator,
                check_output,
                python_args,
                post_child_check=post_child_check,
            )
        finally:
            try:
                os.unlink(write_result_fn)
            except OSError:
                pass

    def test_large_write_stdout_devnull(self):
        self._test_large_write('stdout', _devnull)

    def test_large_write_stdout_pipes(self):