comparison tests/run-tests.py @ 24330:799bc18e14d1

run-tests: avoid running the same test instance concurrently

There's a fair amount of mutable state stored on test case
instances. That causes many weird failures if you try to do something
like `run-tests.py -j16 --loop test-help.t`. The quick fix is this
slightly weird test-reloading dance, which ensures that every time a
test is executed it runs on a fresh instance of the TestCase subclass.

24329:e7ca4d4b10e1

class TestSuite(unittest.TestSuite):
    """Custom unittest TestSuite that knows how to execute Mercurial tests."""

    def __init__(self, testdir, jobs=1, whitelist=None, blacklist=None,
                 retest=False, keywords=None, loop=False, runs_per_test=1,
                 *args, **kwargs):
        """Create a new instance that can run tests with a configuration.

        testdir specifies the directory where tests are executed from. This
        is typically the ``tests`` directory from Mercurial's source

        self._blacklist = blacklist
        self._retest = retest
        self._keywords = keywords
        self._loop = loop
        self._runs_per_test = runs_per_test

    def run(self, result):
        # We have a number of filters that need to be applied. We do this
        # here instead of inside Test because it makes the running logic for
        # Test simpler.
        tests = []
        for test in self._tests:
            if not os.path.exists(test.path):
                result.addSkip(test, "Doesn't exist")
                continue

            if not (self._whitelist and test.name in self._whitelist):

                            break

                    if ignored:
                        continue
            for _ in xrange(self._runs_per_test):
                tests.append(test)

        runtests = list(tests)
        done = queue.Queue()
        running = 0

                        continue
                    running -= 1
                if tests and not running == self._jobs:
                    test = tests.pop(0)
                    if self._loop:
                        tests.append(test)
                    t = threading.Thread(target=job, name=test.name,
                                         args=(test, result))
                    t.start()
                    running += 1
        except KeyboardInterrupt:

                              blacklist=self.options.blacklist,
                              retest=self.options.retest,
                              keywords=self.options.keywords,
                              loop=self.options.loop,
                              runs_per_test=self.options.runs_per_test,
                              tests=tests)
            verbosity = 1
            if self.options.verbose:
                verbosity = 2
            runner = TextTestRunner(self, verbosity=verbosity)
            result = runner.run(suite)

                break

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

        return testcls(refpath, tmpdir,
                       keeptmpdir=self.options.keep_tmpdir,
                       debug=self.options.debug,
                       timeout=self.options.timeout,
                       startport=self.options.port + count * 3,
                       extraconfigopts=self.options.extra_config_opt,
                       py3kwarnings=self.options.py3k_warnings,
                       shell=self.options.shell)

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

        if self.options.keep_tmpdir:

24330:799bc18e14d1

class TestSuite(unittest.TestSuite):
    """Custom unittest TestSuite that knows how to execute Mercurial tests."""

    def __init__(self, testdir, jobs=1, whitelist=None, blacklist=None,
                 retest=False, keywords=None, loop=False, runs_per_test=1,
                 loadtest=None,
                 *args, **kwargs):
        """Create a new instance that can run tests with a configuration.

        testdir specifies the directory where tests are executed from. This
        is typically the ``tests`` directory from Mercurial's source

        self._blacklist = blacklist
        self._retest = retest
        self._keywords = keywords
        self._loop = loop
        self._runs_per_test = runs_per_test
        self._loadtest = loadtest

    def run(self, result):
        # We have a number of filters that need to be applied. We do this
        # here instead of inside Test because it makes the running logic for
        # Test simpler.
        tests = []
        num_tests = [0]
        for test in self._tests:
            def get():
                num_tests[0] += 1
                if getattr(test, 'should_reload', False):
                    return self._loadtest(test.name, num_tests[0])
                return test
            if not os.path.exists(test.path):
                result.addSkip(test, "Doesn't exist")
                continue

            if not (self._whitelist and test.name in self._whitelist):

                            break

                    if ignored:
                        continue
            for _ in xrange(self._runs_per_test):
                tests.append(get())

        runtests = list(tests)
        done = queue.Queue()
        running = 0

                        continue
                    running -= 1
                if tests and not running == self._jobs:
                    test = tests.pop(0)
                    if self._loop:
                        if getattr(test, 'should_reload', False):
                            num_tests[0] += 1
                            tests.append(
                                self._loadtest(test.name, num_tests[0]))
                        else:
                            tests.append(test)
                    t = threading.Thread(target=job, name=test.name,
                                         args=(test, result))
                    t.start()
                    running += 1
        except KeyboardInterrupt:

                              blacklist=self.options.blacklist,
                              retest=self.options.retest,
                              keywords=self.options.keywords,
                              loop=self.options.loop,
                              runs_per_test=self.options.runs_per_test,
                              tests=tests, loadtest=self._gettest)
            verbosity = 1
            if self.options.verbose:
                verbosity = 2
            runner = TextTestRunner(self, verbosity=verbosity)
            result = runner.run(suite)

                break

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

        t = testcls(refpath, tmpdir,
                    keeptmpdir=self.options.keep_tmpdir,
                    debug=self.options.debug,
                    timeout=self.options.timeout,
                    startport=self.options.port + count * 3,
                    extraconfigopts=self.options.extra_config_opt,
                    py3kwarnings=self.options.py3k_warnings,
                    shell=self.options.shell)
        t.should_reload = True
        return t

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

        if self.options.keep_tmpdir: