# -*- coding: utf-8 -*- """Accessories for automated py.test runner.""" from __future__ import print_function, with_statement # std imports import os import sys import codecs import functools import traceback import contextlib # local from blessed import Terminal from blessed._compat import TextType from .conftest import IS_WINDOWS try: from typing import Callable except ImportError: # py2 pass if IS_WINDOWS: import jinxed as curses else: import curses import pty import termios test_kind = 'vtwin10' if IS_WINDOWS else 'xterm-256color' TestTerminal = functools.partial(Terminal, kind=test_kind) # type: Callable[..., Terminal] SEND_SEMAPHORE = SEMAPHORE = b'SEMAPHORE\n' RECV_SEMAPHORE = b'SEMAPHORE\r\n' def init_subproc_coverage(run_note): """Run coverage on subprocess""" try: import coverage except ImportError: return None _coveragerc = os.path.join( os.path.dirname(__file__), os.pardir, 'tox.ini') cov = coverage.Coverage(config_file=_coveragerc) cov.start() return cov class as_subprocess(object): # pylint: disable=too-few-public-methods """This helper executes test cases in a child process, avoiding a python-internal bug of _curses: setupterm() may not be called more than once per process.""" _CHILD_PID = 0 encoding = 'utf8' def __init__(self, func): self.func = func def __call__(self, *args, **kwargs): # pylint: disable=too-many-locals,too-complex if IS_WINDOWS: self.func(*args, **kwargs) return pid_testrunner = os.getpid() pid, master_fd = pty.fork() # pylint: disable=possibly-used-before-assignment if pid == self._CHILD_PID: # child process executes function, raises exception # if failed, causing a non-zero exit code, using the # protected _exit() function of ``os``; to prevent the # 'SystemExit' exception from being thrown. cov = init_subproc_coverage( "@as_subprocess-{pid};{func_name}(*{args}, **{kwargs})" .format(pid=os.getpid(), func_name=self.func, args=args, kwargs=kwargs)) try: self.func(*args, **kwargs) except Exception: # pylint: disable=broad-except e_type, e_value, e_tb = sys.exc_info() o_err = [line.rstrip().encode('utf-8') for line in traceback.format_tb(e_tb)] o_err.append(('-=' * 20).encode('ascii')) o_err.extend([_exc.rstrip().encode('utf-8') for _exc in traceback.format_exception_only( e_type, e_value)]) os.write(sys.__stdout__.fileno(), b'\n'.join(o_err)) os.close(sys.__stdout__.fileno()) os.close(sys.__stderr__.fileno()) os.close(sys.__stdin__.fileno()) if cov is not None: cov.stop() cov.save() os._exit(1) else: if cov is not None: cov.stop() cov.save() os._exit(0) # detect rare fork in test runner, when bad bugs happen if pid_testrunner != os.getpid(): print('TEST RUNNER HAS FORKED, {0}=>{1}: EXIT' .format(pid_testrunner, os.getpid()), file=sys.stderr) os._exit(1) exc_output = TextType() decoder = codecs.getincrementaldecoder(self.encoding)() while True: try: _exc = os.read(master_fd, 65534) except OSError: # linux EOF break if not _exc: # bsd EOF break exc_output += decoder.decode(_exc) # parent process asserts exit code is 0, causing test # to fail if child process raised an exception/assertion pid, status = os.waitpid(pid, 0) os.close(master_fd) # Display any output written by child process # (esp. any AssertionError exceptions written to stderr). exc_output_msg = 'Output in child process:\n%s\n%s\n%s' % ( u'=' * 40, exc_output, u'=' * 40,) assert exc_output == '', exc_output_msg # Also test exit status is non-zero assert os.WEXITSTATUS(status) == 0 def read_until_semaphore(fd, semaphore=RECV_SEMAPHORE, encoding='utf8'): """ Read file descriptor ``fd`` until ``semaphore`` is found. Used to ensure the child process is awake and ready. For timing tests; without a semaphore, the time to fork() would be (incorrectly) included in the duration of the test, which can be very length on continuous integration servers (such as Travis-CI). """ # note that when a child process writes xyz\\n, the parent # process will read xyz\\r\\n -- this is how pseudo terminals # behave; a virtual terminal requires both carriage return and # line feed, it is only for convenience that \\n does both. outp = TextType() decoder = codecs.getincrementaldecoder(encoding)() semaphore = semaphore.decode('ascii') while not outp.startswith(semaphore): try: _exc = os.read(fd, 1) except OSError: # Linux EOF break if not _exc: # BSD EOF break outp += decoder.decode(_exc, final=False) assert outp.startswith(semaphore), ( 'Semaphore not recv before EOF ' '(expected: %r, got: %r)' % (semaphore, outp,)) return outp[len(semaphore):] def read_until_eof(fd, encoding='utf8'): """ Read file descriptor ``fd`` until EOF. Return decoded string. """ decoder = codecs.getincrementaldecoder(encoding)() outp = TextType() while True: try: _exc = os.read(fd, 100) except OSError: # linux EOF break if not _exc: # bsd EOF break outp += decoder.decode(_exc, final=False) return outp @contextlib.contextmanager def echo_off(fd): """Ensure any bytes written to pty fd are not duplicated as output.""" if not IS_WINDOWS: # pylint: disable=possibly-used-before-assignment try: attrs = termios.tcgetattr(fd) attrs[3] = attrs[3] & ~termios.ECHO termios.tcsetattr(fd, termios.TCSANOW, attrs) yield finally: attrs[3] = attrs[3] | termios.ECHO termios.tcsetattr(fd, termios.TCSANOW, attrs) else: yield def unicode_cap(cap): """Return the result of ``tigetstr`` except as Unicode.""" try: val = curses.tigetstr(cap) except curses.error: val = None return val.decode('latin1') if val else u'' def unicode_parm(cap, *parms): """Return the result of ``tparm(tigetstr())`` except as Unicode.""" try: cap = curses.tigetstr(cap) except curses.error: cap = None if cap: try: val = curses.tparm(cap, *parms) except curses.error: val = None if val: return val.decode('latin1') return u'' class MockTigetstr(object): # pylint: disable=too-few-public-methods """ Wraps curses.tigetstr() to override specific capnames Capnames and return values are provided as keyword arguments """ def __init__(self, **kwargs): self.callable = curses.tigetstr self.kwargs = kwargs def __call__(self, capname): return self.kwargs.get(capname, self.callable(capname))