tests/run-tests.py

24507:a0668a587c04

# subprocess._cleanup can race with any Popen.wait or Popen.poll on py24
# http://bugs.python.org/issue1731717 for details. We shouldn't be producing
# zombies but it's pretty harmless even if we do.
if sys.version_info < (2, 5):
    subprocess._cleanup = lambda: None

closefds = os.name == 'posix'
def Popen4(cmd, wd, timeout, env=None):
    processlock.acquire()
    p = subprocess.Popen(cmd, shell=True, bufsize=-1, cwd=wd, env=env,

    def fail(self, msg):
        # unittest differentiates between errored and failed.
        # Failed is denoted by AssertionError (by default at least).
        raise AssertionError(msg)

class PythonTest(Test):
    """A Python-based test."""

    @property
    def refpath(self):

        py3kswitch = self._py3kwarnings and ' -3' or ''
        cmd = '%s%s "%s"' % (PYTHON, py3kswitch, self.path)
        vlog("# Running", cmd)
        if os.name == 'nt':
            replacements.append((r'\r\n', '\n'))
        result = run(cmd, self._testtmp, replacements, env,
                   debug=self._debug, timeout=self._timeout)
        if self._aborted:
            raise KeyboardInterrupt()

        return result

        f.close()

        cmd = '%s "%s"' % (self._shell, fname)
        vlog("# Running", cmd)

        exitcode, output = run(cmd, self._testtmp, replacements, env,
                               debug=self._debug, timeout=self._timeout)

        if self._aborted:
            raise KeyboardInterrupt()

        # Do not merge output if skipped. Return hghave message instead.

    @staticmethod
    def _stringescape(s):
        return TTest.ESCAPESUB(TTest._escapef, s)


wifexited = getattr(os, "WIFEXITED", lambda x: False)
def run(cmd, wd, replacements, env, debug=False, timeout=None):
    """Run command in a sub-process, capturing the output (stdout and stderr).
    Return a tuple (exitcode, output).  output is None in debug mode."""
    if debug:
        proc = subprocess.Popen(cmd, shell=True, cwd=wd, env=env)
        ret = proc.wait()
        return (ret, None)

    proc = Popen4(cmd, wd, timeout, env)
    def cleanup():
        terminate(proc)
        ret = proc.wait()
        if ret == 0:
            ret = signal.SIGTERM << 8
        killdaemons(env['DAEMON_PIDS'])
        return ret

    output = ''
    proc.tochild.close()

    try:
        output = proc.fromchild.read()
    except KeyboardInterrupt:
        vlog('# Handling keyboard interrupt')
        cleanup()
        raise

    ret = proc.wait()
    if wifexited(ret):
        ret = os.WEXITSTATUS(ret)

    if proc.timeout:
        ret = 'timeout'

    if ret:
        killdaemons(env['DAEMON_PIDS'])

    for s, r in replacements:
        output = re.sub(s, r, output)
    return ret, output.splitlines(True)

iolock = threading.RLock()

class SkipTest(Exception):
    """Raised to indicate that a test is to be skipped."""

24508:fbe2fb71a6e6

# subprocess._cleanup can race with any Popen.wait or Popen.poll on py24
# http://bugs.python.org/issue1731717 for details. We shouldn't be producing
# zombies but it's pretty harmless even if we do.
if sys.version_info < (2, 5):
    subprocess._cleanup = lambda: None

wifexited = getattr(os, "WIFEXITED", lambda x: False)

closefds = os.name == 'posix'
def Popen4(cmd, wd, timeout, env=None):
    processlock.acquire()
    p = subprocess.Popen(cmd, shell=True, bufsize=-1, cwd=wd, env=env,

    def fail(self, msg):
        # unittest differentiates between errored and failed.
        # Failed is denoted by AssertionError (by default at least).
        raise AssertionError(msg)

    def _runcommand(self, cmd, wd, replacements, env, debug=False,
                    timeout=None):
        """Run command in a sub-process, capturing the output (stdout and
        stderr).

        Return a tuple (exitcode, output). output is None in debug mode.
        """
        if debug:
            proc = subprocess.Popen(cmd, shell=True, cwd=wd, env=env)
            ret = proc.wait()
            return (ret, None)

        proc = Popen4(cmd, wd, timeout, env)
        def cleanup():
            terminate(proc)
            ret = proc.wait()
            if ret == 0:
                ret = signal.SIGTERM << 8
            killdaemons(env['DAEMON_PIDS'])
            return ret

        output = ''
        proc.tochild.close()

        try:
            output = proc.fromchild.read()
        except KeyboardInterrupt:
            vlog('# Handling keyboard interrupt')
            cleanup()
            raise

        ret = proc.wait()
        if wifexited(ret):
            ret = os.WEXITSTATUS(ret)

        if proc.timeout:
            ret = 'timeout'

        if ret:
            killdaemons(env['DAEMON_PIDS'])

        for s, r in replacements:
            output = re.sub(s, r, output)
        return ret, output.splitlines(True)

class PythonTest(Test):
    """A Python-based test."""

    @property
    def refpath(self):

        py3kswitch = self._py3kwarnings and ' -3' or ''
        cmd = '%s%s "%s"' % (PYTHON, py3kswitch, self.path)
        vlog("# Running", cmd)
        if os.name == 'nt':
            replacements.append((r'\r\n', '\n'))
        result = self._runcommand(cmd, self._testtmp, replacements, env,
                                  debug=self._debug, timeout=self._timeout)
        if self._aborted:
            raise KeyboardInterrupt()

        return result

        f.close()

        cmd = '%s "%s"' % (self._shell, fname)
        vlog("# Running", cmd)

        exitcode, output = self._runcommand(cmd, self._testtmp, replacements,
                                            env, debug=self._debug,
                                            timeout=self._timeout)

        if self._aborted:
            raise KeyboardInterrupt()

        # Do not merge output if skipped. Return hghave message instead.

    @staticmethod
    def _stringescape(s):
        return TTest.ESCAPESUB(TTest._escapef, s)

iolock = threading.RLock()

class SkipTest(Exception):
    """Raised to indicate that a test is to be skipped."""