def _runtests()

in pylib/mercurial-support/run-tests.py [0:0]

• 76 lines of code
• 20 McCabe index (conditional complexity)

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

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

            tests = [self._gettest(d, i) for i, d in enumerate(testdescs)]
            num_tests = len(tests) * self.options.runs_per_test

            jobs = min(num_tests, self.options.jobs)

            failed = False
            kws = self.options.keywords
            if kws is not None and PYTHON3:
                kws = kws.encode('utf-8')

            suite = TestSuite(
                self._testdir,
                jobs=jobs,
                whitelist=self.options.whitelisted,
                blacklist=self.options.blacklist,
                retest=self.options.retest,
                keywords=kws,
                loop=self.options.loop,
                runs_per_test=self.options.runs_per_test,
                showchannels=self.options.showchannels,
                tests=tests,
                loadtest=_reloadtest,
            )
            verbosity = 1
            if self.options.list_tests:
                verbosity = 0
            elif self.options.verbose:
                verbosity = 2
            runner = TextTestRunner(self, verbosity=verbosity)

            if self.options.list_tests:
                result = runner.listtests(suite)
            else:
                if self._installdir:
                    self._installhg()
                    self._checkhglib("Testing")
                else:
                    self._usecorrectpython()
                if self.options.chg:
                    assert self._installdir
                    self._installchg()

                log(
                    'running %d tests using %d parallel processes'
                    % (num_tests, jobs)
                )

                result = runner.run(suite)

            if result.failures or result.errors:
                failed = True

            result.onEnd()

            if self.options.anycoverage:
                self._outputcoverage()
        except KeyboardInterrupt:
            failed = True
            print("\ninterrupted!")

        if failed:
            return 1