test-stdio.py
245 lines
| 7.4 KiB
| text/x-python
|
PythonLexer
/ tests / test-stdio.py
|
r45583 | #!/usr/bin/env python | ||
""" | ||||
Tests the buffering behavior of stdio streams in `mercurial.utils.procutil`. | ||||
""" | ||||
from __future__ import absolute_import | ||||
import contextlib | ||||
|
r45628 | import errno | ||
|
r45583 | import os | ||
|
r45655 | import signal | ||
|
r45583 | import subprocess | ||
import sys | ||||
import unittest | ||||
from mercurial import pycompat | ||||
|
r45638 | TEST_BUFFERING_CHILD_SCRIPT = r''' | ||
|
r45583 | import os | ||
from mercurial import dispatch | ||||
from mercurial.utils import procutil | ||||
dispatch.initstdio() | ||||
|
r45589 | procutil.{stream}.write(b'aaa') | ||
os.write(procutil.{stream}.fileno(), b'[written aaa]') | ||||
procutil.{stream}.write(b'bbb\n') | ||||
os.write(procutil.{stream}.fileno(), b'[written bbb\\n]') | ||||
|
r45583 | ''' | ||
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' | ||||
|
r45655 | TEST_LARGE_WRITE_CHILD_SCRIPT = r''' | ||
import signal | ||||
import sys | ||||
from mercurial import dispatch | ||||
from mercurial.utils import procutil | ||||
signal.signal(signal.SIGINT, lambda *x: None) | ||||
dispatch.initstdio() | ||||
procutil.{stream}.write(b'x' * 1048576) | ||||
''' | ||||
|
r45583 | @contextlib.contextmanager | ||
def _closing(fds): | ||||
try: | ||||
yield | ||||
finally: | ||||
for fd in fds: | ||||
try: | ||||
os.close(fd) | ||||
except EnvironmentError: | ||||
pass | ||||
@contextlib.contextmanager | ||||
|
r45656 | def _devnull(): | ||
devnull = os.open(os.devnull, os.O_WRONLY) | ||||
with _closing([devnull]): | ||||
yield (None, devnull) | ||||
@contextlib.contextmanager | ||||
|
r45583 | def _pipes(): | ||
rwpair = os.pipe() | ||||
with _closing(rwpair): | ||||
yield rwpair | ||||
@contextlib.contextmanager | ||||
def _ptys(): | ||||
if pycompat.iswindows: | ||||
raise unittest.SkipTest("PTYs are not supported on Windows") | ||||
import pty | ||||
import tty | ||||
rwpair = pty.openpty() | ||||
with _closing(rwpair): | ||||
tty.setraw(rwpair[0]) | ||||
yield rwpair | ||||
|
r45655 | def _readall(fd, buffer_size, initial_buf=None): | ||
buf = initial_buf or [] | ||||
|
r45628 | while True: | ||
try: | ||||
s = os.read(fd, buffer_size) | ||||
except OSError as e: | ||||
if e.errno == errno.EIO: | ||||
# If the child-facing PTY got closed, reading from the | ||||
# parent-facing PTY raises EIO. | ||||
break | ||||
raise | ||||
if not s: | ||||
break | ||||
buf.append(s) | ||||
return b''.join(buf) | ||||
|
r45589 | class TestStdio(unittest.TestCase): | ||
|
r45638 | def _test( | ||
self, | ||||
child_script, | ||||
stream, | ||||
rwpair_generator, | ||||
check_output, | ||||
python_args=[], | ||||
): | ||||
|
r45589 | assert stream in ('stdout', 'stderr') | ||
with rwpair_generator() as (stream_receiver, child_stream), open( | ||||
|
r45583 | os.devnull, 'rb' | ||
) as child_stdin: | ||||
proc = subprocess.Popen( | ||||
|
r45638 | [sys.executable] + python_args + ['-c', child_script], | ||
|
r45583 | stdin=child_stdin, | ||
|
r45589 | stdout=child_stream if stream == 'stdout' else None, | ||
stderr=child_stream if stream == 'stderr' else None, | ||||
|
r45583 | ) | ||
|
r45628 | try: | ||
os.close(child_stream) | ||||
|
r45656 | if stream_receiver is not None: | ||
check_output(stream_receiver, proc) | ||||
|
r45629 | except: # re-raises | ||
proc.terminate() | ||||
raise | ||||
|
r45628 | finally: | ||
retcode = proc.wait() | ||||
|
r45583 | self.assertEqual(retcode, 0) | ||
|
r45638 | def _test_buffering( | ||
self, stream, rwpair_generator, expected_output, python_args=[] | ||||
): | ||||
|
r45655 | def check_output(stream_receiver, proc): | ||
|
r45638 | self.assertEqual(_readall(stream_receiver, 1024), expected_output) | ||
self._test( | ||||
TEST_BUFFERING_CHILD_SCRIPT.format(stream=stream), | ||||
stream, | ||||
rwpair_generator, | ||||
check_output, | ||||
python_args, | ||||
) | ||||
|
r45656 | def test_buffering_stdout_devnull(self): | ||
self._test_buffering('stdout', _devnull, None) | ||||
|
r45630 | def test_buffering_stdout_pipes(self): | ||
|
r45638 | self._test_buffering('stdout', _pipes, FULLY_BUFFERED) | ||
|
r45583 | |||
|
r45630 | def test_buffering_stdout_ptys(self): | ||
|
r45638 | self._test_buffering('stdout', _ptys, LINE_BUFFERED) | ||
|
r45583 | |||
|
r45656 | def test_buffering_stdout_devnull_unbuffered(self): | ||
self._test_buffering('stdout', _devnull, None, python_args=['-u']) | ||||
|
r45630 | def test_buffering_stdout_pipes_unbuffered(self): | ||
|
r45638 | self._test_buffering('stdout', _pipes, UNBUFFERED, python_args=['-u']) | ||
|
r45583 | |||
|
r45630 | def test_buffering_stdout_ptys_unbuffered(self): | ||
|
r45638 | self._test_buffering('stdout', _ptys, UNBUFFERED, python_args=['-u']) | ||
|
r45583 | |||
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. | ||||
|
r45630 | test_buffering_stdout_ptys_unbuffered = unittest.expectedFailure( | ||
test_buffering_stdout_ptys_unbuffered | ||||
|
r45583 | ) | ||
|
r45655 | def _test_large_write(self, stream, rwpair_generator, python_args=[]): | ||
if not pycompat.ispy3 and pycompat.isdarwin: | ||||
# Python 2 doesn't always retry on EINTR, but the libc might retry. | ||||
# So far, it was observed only on macOS that EINTR is raised at the | ||||
# Python level. As Python 2 support will be dropped soon-ish, we | ||||
# won't attempt to fix it. | ||||
raise unittest.SkipTest("raises EINTR on macOS") | ||||
def check_output(stream_receiver, proc): | ||||
if not pycompat.iswindows: | ||||
# On Unix, we can provoke a partial write() by interrupting it | ||||
# by a signal handler as soon as a bit of data was written. | ||||
# We test that write() is called until all data is written. | ||||
buf = [os.read(stream_receiver, 1)] | ||||
proc.send_signal(signal.SIGINT) | ||||
else: | ||||
# On Windows, there doesn't seem to be a way to cause partial | ||||
# writes. | ||||
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, | ||||
) | ||||
|
r45656 | def test_large_write_stdout_devnull(self): | ||
self._test_large_write('stdout', _devnull) | ||||
|
r45655 | def test_large_write_stdout_pipes(self): | ||
self._test_large_write('stdout', _pipes) | ||||
def test_large_write_stdout_ptys(self): | ||||
self._test_large_write('stdout', _ptys) | ||||
|
r45656 | def test_large_write_stdout_devnull_unbuffered(self): | ||
self._test_large_write('stdout', _devnull, python_args=['-u']) | ||||
|
r45655 | def test_large_write_stdout_pipes_unbuffered(self): | ||
self._test_large_write('stdout', _pipes, python_args=['-u']) | ||||
def test_large_write_stdout_ptys_unbuffered(self): | ||||
self._test_large_write('stdout', _ptys, python_args=['-u']) | ||||
|
r45656 | def test_large_write_stderr_devnull(self): | ||
self._test_large_write('stderr', _devnull) | ||||
|
r45655 | def test_large_write_stderr_pipes(self): | ||
self._test_large_write('stderr', _pipes) | ||||
def test_large_write_stderr_ptys(self): | ||||
self._test_large_write('stderr', _ptys) | ||||
|
r45656 | def test_large_write_stderr_devnull_unbuffered(self): | ||
self._test_large_write('stderr', _devnull, python_args=['-u']) | ||||
|
r45655 | def test_large_write_stderr_pipes_unbuffered(self): | ||
self._test_large_write('stderr', _pipes, python_args=['-u']) | ||||
def test_large_write_stderr_ptys_unbuffered(self): | ||||
self._test_large_write('stderr', _ptys, python_args=['-u']) | ||||
|
r45583 | |||
if __name__ == '__main__': | ||||
import silenttestrunner | ||||
silenttestrunner.main(__name__) | ||||