comparison tests/run-tests.py @ 24508:fbe2fb71a6e6

run-tests: move run into Test class

Future patches will change how replacements work. Since the logic in
run() is strongly tied to the operation of individual tests and since
there is potential to make the implementation simpler by giving the
function access to Test attributes, move it into Test.

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."""