tests/run-tests.py

21453:aaf52b78327e

        self._ret = None
        self._out = None
        self._duration = None
        self._result = None
        self._skipped = None

        # If we're not in --debug mode and reference output file exists,
        # check test output against it.
        if runner.options.debug:
            self._refout = None # to match "out is None"

        self._out = None
        self._duration = None
        self._result = None
        self._skipped = None

    def run(self):
        """Run this test instance.

        This will return a tuple describing the result of the test.
        """

        # Remove any previous output files.
        if os.path.exists(self._errpath):
            os.remove(self._errpath)

        testtmp = os.path.join(self._threadtmp, os.path.basename(self._path))
        os.mkdir(testtmp)
        replacements, port = self._getreplacements(testtmp)
        env = self._getenv(testtmp, port)
        self._daemonpids.append(env['DAEMON_PIDS'])
        self._createhgrc(env['HGRCPATH'])

        vlog('# Test', self.name)

        starttime = time.time()
        try:
            ret, out = self._run(testtmp, replacements, env)
            self._duration = time.time() - starttime
            self._finished = True
            self._ret = ret
            self._out = out
        except KeyboardInterrupt:

            raise
        except Exception, e:
            return self.fail('Exception during execution: %s' % e, 255)

        killdaemons(env['DAEMON_PIDS'])

        if not options.keep_tmpdir:
            shutil.rmtree(testtmp)

        def describe(ret):
            if ret < 0:
                return 'killed by signal: %d' % -ret
            return 'returned error code %d' % ret

        return self._result

    def tearDown(self):
        """Tasks to perform after run()."""
        vlog("# Ret was:", self._ret)

        # Don't print progress in unittest mode because that is handled
        # by TestResult.
        if not self._options.verbose and not self._unittest:

            sys.stdout.flush()
            iolock.release()

        self._runner.times.append((self.name, self._duration))

    def _run(self, testtmp, replacements, env):
        # This should be implemented in child classes to run tests.
        return self._skip('unknown test type')

    def _getreplacements(self, testtmp):
        port = self._options.port + self._count * 3
        r = [
            (r':%s\b' % port, ':$HGPORT'),
            (r':%s\b' % (port + 1), ':$HGPORT1'),
            (r':%s\b' % (port + 2), ':$HGPORT2'),

        if os.name == 'nt':
            r.append(
                (''.join(c.isalpha() and '[%s%s]' % (c.lower(), c.upper()) or
                    c in '/\\' and r'[/\\]' or c.isdigit() and c or '\\' + c
                    for c in testtmp), '$TESTTMP'))
        else:
            r.append((re.escape(testtmp), '$TESTTMP'))

        return r, port

    def _getenv(self, testtmp, port):
        env = os.environ.copy()
        env['TESTTMP'] = testtmp
        env['HOME'] = testtmp
        env["HGPORT"] = str(port)
        env["HGPORT1"] = str(port + 1)
        env["HGPORT2"] = str(port + 2)
        env["HGRCPATH"] = os.path.join(self._threadtmp, '.hgrc')
        env["DAEMON_PIDS"] = os.path.join(self._threadtmp, 'daemon.pids')

        return 'i', self.name, msg

class PythonTest(Test):
    """A Python-based test."""
    def _run(self, testtmp, replacements, env):
        py3kswitch = self._options.py3k_warnings and ' -3' or ''
        cmd = '%s%s "%s"' % (PYTHON, py3kswitch, self._path)
        vlog("# Running", cmd)
        if os.name == 'nt':
            replacements.append((r'\r\n', '\n'))
        return run(cmd, testtmp, self._options, replacements, env,
                   self._runner.abort)

class TTest(Test):
    """A "t test" is a test backed by a .t file."""

    ESCAPESUB = re.compile(r'[\x00-\x08\x0b-\x1f\\\x7f-\xff]').sub
    ESCAPEMAP = dict((chr(i), r'\x%02x' % i) for i in range(256)).update(
                     {'\\': '\\\\', '\r': r'\r'})

    def _run(self, testtmp, replacements, env):
        f = open(self._path)
        lines = f.readlines()
        f.close()

        salt, script, after, expected = self._parsetest(lines, testtmp)

        # Write out the generated script.
        fname = '%s.sh' % testtmp
        f = open(fname, 'w')
        for l in script:
            f.write(l)
        f.close()

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

        exitcode, output = run(cmd, testtmp, self._options, replacements, env,
                               self._runner.abort)
        # Do not merge output if skipped. Return hghave message instead.
        # Similarly, with --debug, output is None.
        if exitcode == self.SKIPPED_STATUS or output is None:
            return exitcode, output

        return self._processoutput(exitcode, output, salt, after, expected)

    def _hghave(self, reqs, testtmp):
        # TODO do something smarter when all other uses of hghave are gone.
        tdir = self._testdir.replace('\\', '/')
        proc = Popen4('%s -c "%s/hghave %s"' %
                      (self._options.shell, tdir, ' '.join(reqs)),
                      testtmp, 0)
        stdout, stderr = proc.communicate()
        ret = proc.wait()
        if wifexited(ret):
            ret = os.WEXITSTATUS(ret)
        if ret == 2:
            print stdout
            sys.exit(1)

        return ret == 0

    def _parsetest(self, lines, testtmp):
        # We generate a shell script which outputs unique markers to line
        # up script results with our source. These markers include input
        # line number and the last return code.
        salt = "SALT" + str(time.time())
        def addsalt(line, inpython):

                lsplit = l.split()
                if len(lsplit) < 2 or lsplit[0] != '#if':
                    after.setdefault(pos, []).append('  !!! invalid #if\n')
                if skipping is not None:
                    after.setdefault(pos, []).append('  !!! nested #if\n')
                skipping = not self._hghave(lsplit[1:], testtmp)
                after.setdefault(pos, []).append(l)
            elif l.startswith('#else'):
                if skipping is None:
                    after.setdefault(pos, []).append('  !!! missing #if\n')
                skipping = not skipping

21454:046587aa1c8a

        self._ret = None
        self._out = None
        self._duration = None
        self._result = None
        self._skipped = None
        self._testtmp = None

        # If we're not in --debug mode and reference output file exists,
        # check test output against it.
        if runner.options.debug:
            self._refout = None # to match "out is None"

        self._out = None
        self._duration = None
        self._result = None
        self._skipped = None

        self._testtmp = os.path.join(self._threadtmp,
                                     os.path.basename(self._path))
        os.mkdir(self._testtmp)

    def run(self):
        """Run this test instance.

        This will return a tuple describing the result of the test.
        """

        # Remove any previous output files.
        if os.path.exists(self._errpath):
            os.remove(self._errpath)

        replacements, port = self._getreplacements()
        env = self._getenv(port)
        self._daemonpids.append(env['DAEMON_PIDS'])
        self._createhgrc(env['HGRCPATH'])

        vlog('# Test', self.name)

        starttime = time.time()
        try:
            ret, out = self._run(replacements, env)
            self._duration = time.time() - starttime
            self._finished = True
            self._ret = ret
            self._out = out
        except KeyboardInterrupt:

            raise
        except Exception, e:
            return self.fail('Exception during execution: %s' % e, 255)

        killdaemons(env['DAEMON_PIDS'])

        def describe(ret):
            if ret < 0:
                return 'killed by signal: %d' % -ret
            return 'returned error code %d' % ret

        return self._result

    def tearDown(self):
        """Tasks to perform after run()."""
        if not self._options.keep_tmpdir:
            shutil.rmtree(self._testtmp)

        vlog("# Ret was:", self._ret)

        # Don't print progress in unittest mode because that is handled
        # by TestResult.
        if not self._options.verbose and not self._unittest:

            sys.stdout.flush()
            iolock.release()

        self._runner.times.append((self.name, self._duration))

    def _run(self, replacements, env):
        # This should be implemented in child classes to run tests.
        return self._skip('unknown test type')

    def _getreplacements(self):
        port = self._options.port + self._count * 3
        r = [
            (r':%s\b' % port, ':$HGPORT'),
            (r':%s\b' % (port + 1), ':$HGPORT1'),
            (r':%s\b' % (port + 2), ':$HGPORT2'),

        if os.name == 'nt':
            r.append(
                (''.join(c.isalpha() and '[%s%s]' % (c.lower(), c.upper()) or
                    c in '/\\' and r'[/\\]' or c.isdigit() and c or '\\' + c
                    for c in self._testtmp), '$TESTTMP'))
        else:
            r.append((re.escape(self._testtmp), '$TESTTMP'))

        return r, port

    def _getenv(self, port):
        env = os.environ.copy()
        env['TESTTMP'] = self._testtmp
        env['HOME'] = self._testtmp
        env["HGPORT"] = str(port)
        env["HGPORT1"] = str(port + 1)
        env["HGPORT2"] = str(port + 2)
        env["HGRCPATH"] = os.path.join(self._threadtmp, '.hgrc')
        env["DAEMON_PIDS"] = os.path.join(self._threadtmp, 'daemon.pids')

        return 'i', self.name, msg

class PythonTest(Test):
    """A Python-based test."""
    def _run(self, replacements, env):
        py3kswitch = self._options.py3k_warnings and ' -3' or ''
        cmd = '%s%s "%s"' % (PYTHON, py3kswitch, self._path)
        vlog("# Running", cmd)
        if os.name == 'nt':
            replacements.append((r'\r\n', '\n'))
        return run(cmd, self._testtmp, self._options, replacements, env,
                   self._runner.abort)

class TTest(Test):
    """A "t test" is a test backed by a .t file."""

    ESCAPESUB = re.compile(r'[\x00-\x08\x0b-\x1f\\\x7f-\xff]').sub
    ESCAPEMAP = dict((chr(i), r'\x%02x' % i) for i in range(256)).update(
                     {'\\': '\\\\', '\r': r'\r'})

    def _run(self, replacements, env):
        f = open(self._path)
        lines = f.readlines()
        f.close()

        salt, script, after, expected = self._parsetest(lines)

        # Write out the generated script.
        fname = '%s.sh' % self._testtmp
        f = open(fname, 'w')
        for l in script:
            f.write(l)
        f.close()

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

        exitcode, output = run(cmd, self._testtmp, self._options, replacements,
                               env, self._runner.abort)
        # Do not merge output if skipped. Return hghave message instead.
        # Similarly, with --debug, output is None.
        if exitcode == self.SKIPPED_STATUS or output is None:
            return exitcode, output

        return self._processoutput(exitcode, output, salt, after, expected)

    def _hghave(self, reqs):
        # TODO do something smarter when all other uses of hghave are gone.
        tdir = self._testdir.replace('\\', '/')
        proc = Popen4('%s -c "%s/hghave %s"' %
                      (self._options.shell, tdir, ' '.join(reqs)),
                      self._testtmp, 0)
        stdout, stderr = proc.communicate()
        ret = proc.wait()
        if wifexited(ret):
            ret = os.WEXITSTATUS(ret)
        if ret == 2:
            print stdout
            sys.exit(1)

        return ret == 0

    def _parsetest(self, lines):
        # We generate a shell script which outputs unique markers to line
        # up script results with our source. These markers include input
        # line number and the last return code.
        salt = "SALT" + str(time.time())
        def addsalt(line, inpython):

                lsplit = l.split()
                if len(lsplit) < 2 or lsplit[0] != '#if':
                    after.setdefault(pos, []).append('  !!! invalid #if\n')
                if skipping is not None:
                    after.setdefault(pos, []).append('  !!! nested #if\n')
                skipping = not self._hghave(lsplit[1:])
                after.setdefault(pos, []).append(l)
            elif l.startswith('#else'):
                if skipping is None:
                    after.setdefault(pos, []).append('  !!! missing #if\n')
                skipping = not skipping