comparison tests/run-tests.py @ 32316:7340465bd788

run-tests: support multiple cases in .t test

Sometimes we want to run similar tests with slightly different
configurations. Previously we duplicate the test files. This patch
introduces special "#testcases" syntax that allows a single .t file to
contain multiple test cases.

Defined cases could be tested using "#if".

For example, if a test should behave the same with or without an
experimental flag, we can add the following to the .t header:

    #testcases default experimental-a
    #if experimental-a
      $ cat >> $HGRCPATH << EOF
      > [experimental]
      > feature=a
      > EOF
    #endif

The "experimental-a" block won't be executed when running the "default" test
case.

32315:67026d65a4fc

            if line:
                entries[line] = filename

        f.close()
    return entries

def getparser():
    """Obtain the OptionParser used by the CLI."""
    parser = optparse.OptionParser("%prog [options] [tests]")

        """
        self.path = path
        self.bname = os.path.basename(path)
        self.name = _strpath(self.bname)
        self._testdir = os.path.dirname(path)
        self.errpath = os.path.join(self._testdir, b'%s.err' % self.bname)

        self._threadtmp = tmpdir
        self._keeptmpdir = keeptmpdir
        self._debug = debug

            os.mkdir(self._threadtmp)
        except OSError as e:
            if e.errno != errno.EEXIST:
                raise

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

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

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

    @property
    def refpath(self):
        return os.path.join(self._testdir, self.bname)

    def _run(self, env):

            return False, stdout

        if 'slow' in reqs:
            self._timeout = self._slowtimeout
        return True, None

    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.

                lsplit = l.split()
                if len(lsplit) < 2 or lsplit[0] != b'#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:])[0]
                after.setdefault(pos, []).append(l)
            elif l.startswith(b'#else'):
                if skipping is None:
                    after.setdefault(pos, []).append('  !!! missing #if\n')
                skipping = not skipping

                stdout, stderr = proc.communicate()
                args = stdout.strip(b'\0').split(b'\0')
            else:
                args = os.listdir(b'.')

        return [{'path': t} for t in args
                if os.path.basename(t).startswith(b'test-')
                    and (t.endswith(b'.py') or t.endswith(b'.t'))]

    def _runtests(self, testdescs):
        def _reloadtest(test, i):
            # convert a test back to its description dict
            desc = {'path': test.path}
            return self._gettest(desc, i)

        try:
            if self._installdir:
                self._installhg()

                self._installchg()

            if self.options.restart:
                orig = list(testdescs)
                while testdescs:
                    if os.path.exists(testdescs[0]['path'] + ".err"):
                        break
                    testdescs.pop(0)
                if not testdescs:
                    print("running all tests")
                    testdescs = orig

                break

        refpath = os.path.join(self._testdir, path)
        tmpdir = os.path.join(self._hgtmp, b'child%d' % count)

        t = testcls(refpath, tmpdir,
                    keeptmpdir=self.options.keep_tmpdir,
                    debug=self.options.debug,
                    timeout=self.options.timeout,
                    startport=self._getport(count),
                    extraconfigopts=self.options.extra_config_opt,
                    py3kwarnings=self.options.py3k_warnings,
                    shell=self.options.shell,
                    hgcommand=self._hgcommand,
                    usechg=bool(self.options.with_chg or self.options.chg),
                    useipv6=useipv6)
        t.should_reload = True
        return t

    def _cleanup(self):
        """Clean up state from this test invocation."""

32316:7340465bd788

            if line:
                entries[line] = filename

        f.close()
    return entries

def parsettestcases(path):
    """read a .t test file, return a set of test case names

    If path does not exist, return an empty set.
    """
    cases = set()
    try:
        with open(path, 'rb') as f:
            for l in f:
                if l.startswith(b'#testcases '):
                    cases.update(l[11:].split())
    except IOError as ex:
        if ex.errno != errno.ENOENT:
            raise
    return cases

def getparser():
    """Obtain the OptionParser used by the CLI."""
    parser = optparse.OptionParser("%prog [options] [tests]")

        """
        self.path = path
        self.bname = os.path.basename(path)
        self.name = _strpath(self.bname)
        self._testdir = os.path.dirname(path)
        self._tmpname = os.path.basename(path)
        self.errpath = os.path.join(self._testdir, b'%s.err' % self.bname)

        self._threadtmp = tmpdir
        self._keeptmpdir = keeptmpdir
        self._debug = debug

            os.mkdir(self._threadtmp)
        except OSError as e:
            if e.errno != errno.EEXIST:
                raise

        name = self._tmpname
        self._testtmp = os.path.join(self._threadtmp, name)
        os.mkdir(self._testtmp)

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

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

    def __init__(self, path, *args, **kwds):
        # accept an extra "case" parameter
        case = None
        if 'case' in kwds:
            case = kwds.pop('case')
        self._case = case
        self._allcases = parsettestcases(path)
        super(TTest, self).__init__(path, *args, **kwds)
        if case:
            self.name = '%s (case %s)' % (self.name, _strpath(case))
            self.errpath = b'%s.%s.err' % (self.errpath[:-4], case)
            self._tmpname += b'-%s' % case

    @property
    def refpath(self):
        return os.path.join(self._testdir, self.bname)

    def _run(self, env):

            return False, stdout

        if 'slow' in reqs:
            self._timeout = self._slowtimeout
        return True, None

    def _iftest(self, args):
        # implements "#if"
        reqs = []
        for arg in args:
            if arg.startswith(b'no-') and arg[3:] in self._allcases:
                if arg[3:] == self._case:
                    return False
            elif arg in self._allcases:
                if arg != self._case:
                    return False
            else:
                reqs.append(arg)
        return self._hghave(reqs)[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.

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

                stdout, stderr = proc.communicate()
                args = stdout.strip(b'\0').split(b'\0')
            else:
                args = os.listdir(b'.')

        tests = []
        for t in args:
            if not (os.path.basename(t).startswith(b'test-')
                    and (t.endswith(b'.py') or t.endswith(b'.t'))):
                continue
            if t.endswith(b'.t'):
                # .t file may contain multiple test cases
                cases = sorted(parsettestcases(t))
                if cases:
                    tests += [{'path': t, 'case': c} for c in sorted(cases)]
                else:
                    tests.append({'path': t})
            else:
                tests.append({'path': t})
        return tests

    def _runtests(self, testdescs):
        def _reloadtest(test, i):
            # convert a test back to its description dict
            desc = {'path': test.path}
            case = getattr(test, '_case', None)
            if case:
                desc['case'] = case
            return self._gettest(desc, i)

        try:
            if self._installdir:
                self._installhg()

                self._installchg()

            if self.options.restart:
                orig = list(testdescs)
                while testdescs:
                    desc = testdescs[0]
                    if 'case' in desc:
                        errpath = b'%s.%s.err' % (desc['path'], desc['case'])
                    else:
                        errpath = b'%s.err' % desc['path']
                    if os.path.exists(errpath):
                        break
                    testdescs.pop(0)
                if not testdescs:
                    print("running all tests")
                    testdescs = orig

                break

        refpath = os.path.join(self._testdir, path)
        tmpdir = os.path.join(self._hgtmp, b'child%d' % count)

        # extra keyword parameters. 'case' is used by .t tests
        kwds = dict((k, testdesc[k]) for k in ['case'] if k in testdesc)

        t = testcls(refpath, tmpdir,
                    keeptmpdir=self.options.keep_tmpdir,
                    debug=self.options.debug,
                    timeout=self.options.timeout,
                    startport=self._getport(count),
                    extraconfigopts=self.options.extra_config_opt,
                    py3kwarnings=self.options.py3k_warnings,
                    shell=self.options.shell,
                    hgcommand=self._hgcommand,
                    usechg=bool(self.options.with_chg or self.options.chg),
                    useipv6=useipv6, **kwds)
        t.should_reload = True
        return t

    def _cleanup(self):
        """Clean up state from this test invocation."""