[2.7] bpo-30283: regrtest: backport test_slow_interrupted() and test_coverage()

Lib/test/regrtest.py

@@ -316,7 +316,7 @@ def main(tests=None, testdir=None, verbose=0, quiet=False,
              'use=', 'threshold=', 'trace', 'coverdir=', 'nocoverdir',
              'runleaks', 'huntrleaks=', 'memlimit=', 'randseed=',
              'multiprocess=', 'slaveargs=', 'forever', 'header', 'pgo',
-             'failfast', 'match=', 'testdir=', 'list-tests'])
+             'failfast', 'match=', 'testdir=', 'list-tests', 'coverage'])
     except getopt.error, msg:
         usage(2, msg)
 
@@ -531,8 +531,7 @@ def main(tests=None, testdir=None, verbose=0, quiet=False,
 
     if trace:
         import trace
-        tracer = trace.Trace(ignoredirs=[sys.prefix, sys.exec_prefix],
-                             trace=False, count=True)
+        tracer = trace.Trace(trace=False, count=True)
 
     test_times = []
     test_support.use_resources = use_resources
@@ -544,7 +543,7 @@ def accumulate_result(test, result):
             test_times.append((test_time, test))
         if ok == PASSED:
             good.append(test)
-        elif ok == FAILED:
+        elif ok in (FAILED, CHILD_ERROR):
             bad.append(test)
         elif ok == ENV_CHANGED:
             environment_changed.append(test)
@@ -553,9 +552,8 @@ def accumulate_result(test, result):
         elif ok == RESOURCE_DENIED:
             skipped.append(test)
             resource_denieds.append(test)
-        else:
-            # CHILD_ERROR
-            bad.append(test)
+        elif ok != INTERRUPTED:
+            raise ValueError("invalid test result: %r" % ok)
 
     if forever:
         def test_forever(tests=list(selected)):
@@ -740,19 +738,26 @@ def get_running(workers):
                 if previous_test:
                     text = '%s -- %s' % (text, previous_test)
                 display_progress(test_index, text)
+
+            def local_runtest():
+                result = runtest(test, verbose, quiet, huntrleaks, None, pgo,
+                                 failfast=failfast,
+                                 match_tests=match_tests,
+                                 testdir=testdir)
+                accumulate_result(test, result)
+                return result
+
+            start_time = time.time()
             if trace:
                 # If we're tracing code coverage, then we don't exit with status
                 # if on a false return value from main.
-                tracer.runctx('runtest(test, verbose, quiet, testdir=testdir)',
-                              globals=globals(), locals=vars())
+                ns = dict(locals())
+                tracer.runctx('result = local_runtest()',
+                              globals=globals(), locals=ns)
+                result = ns['result']
             else:
-                start_time = time.time()
                 try:
-                    result = runtest(test, verbose, quiet, huntrleaks, None, pgo,
-                                     failfast=failfast,
-                                     match_tests=match_tests,
-                                     testdir=testdir)
-                    accumulate_result(test, result)
+                    result = local_runtest()
                     if verbose3 and result[0] == FAILED:
                         if not pgo:
                             print "Re-running test %r in verbose mode" % test
@@ -764,14 +769,14 @@ def get_running(workers):
                 except:
                     raise
 
-                previous_test = format_test_result(test, result[0])
-                test_time = time.time() - start_time
-                if test_time >= PROGRESS_MIN_TIME:
-                    previous_test = "%s in %s" % (previous_test,
-                                                  format_duration(test_time))
-                elif result[0] == PASSED:
-                    # be quiet: say nothing if the test passed shortly
-                    previous_test = None
+            test_time = time.time() - start_time
+            previous_test = format_test_result(test, result[0])
+            if test_time >= PROGRESS_MIN_TIME:
+                previous_test = "%s in %s" % (previous_test,
+                                              format_duration(test_time))
+            elif result[0] == PASSED:
+                # be quiet: say nothing if the test passed shortly
+                previous_test = None
 
             if findleaks:
                 gc.collect()

Lib/test/test_regrtest.py

@@ -255,10 +255,22 @@ def run_batch(self, *args):
         proc = self.run_command(args)
         self.check_output(proc.stdout)
 
-    @unittest.skipUnless(sysconfig.is_python_build(),
-                         'test.bat script is not installed')
+    def need_pcbuild(self):
+        exe = os.path.normpath(os.path.abspath(sys.executable))
+        parts = exe.split(os.path.sep)
+        if len(parts) < 3:
+            # it's not a python build, python is likely to be installed
+            return
+
+        build_dir = parts[-3]
+        if build_dir.lower() != 'pcbuild':
+            self.skipTest("Tools/buildbot/test.bat requires PCbuild build, "
+                          "found %s" % build_dir)
+
     @unittest.skipUnless(sys.platform == 'win32', 'Windows only')
     def test_tools_buildbot_test(self):
+        self.need_pcbuild()
+
         # Tools\buildbot\test.bat
         script = os.path.join(ROOT_DIR, 'Tools', 'buildbot', 'test.bat')
         test_args = ['--testdir=%s' % self.tmptestdir]
@@ -272,6 +284,8 @@ def test_tools_buildbot_test(self):
 
     @unittest.skipUnless(sys.platform == 'win32', 'Windows only')
     def test_pcbuild_rt(self):
+        self.need_pcbuild()
+
         # PCbuild\rt.bat
         script = os.path.join(ROOT_DIR, r'PCbuild\rt.bat')
         rt_args = ["-q"]             # Quick, don't run tests twice
@@ -392,6 +406,37 @@ def test_slowest(self):
                  % (self.TESTNAME_REGEX, len(tests)))
         self.check_line(output, regex)
 
+    def test_slow_interrupted(self):
+        # Issue #25373: test --slowest with an interrupted test
+        code = TEST_INTERRUPTED
+        test = self.create_test("sigint", code=code)
+
+        try:
+            import threading
+            tests = (False, True)
+        except ImportError:
+            tests = (False,)
+        for multiprocessing in tests:
+            if multiprocessing:
+                args = ("--slowest", "-j2", test)
+            else:
+                args = ("--slowest", test)
+            output = self.run_tests(*args, exitcode=1)
+            self.check_executed_tests(output, test,
+                                      omitted=test, interrupted=True)
+
+            regex = ('10 slowest tests:\n')
+            self.check_line(output, regex)
+
+    def test_coverage(self):
+        # test --coverage
+        test = self.create_test('coverage')
+        output = self.run_tests("--coverage", test)
+        self.check_executed_tests(output, [test])
+        regex = (r'lines +cov% +module +\(path\)\n'
+                 r'(?: *[0-9]+ *[0-9]{1,2}% *[^ ]+ +\([^)]+\)+)+')
+        self.check_line(output, regex)
+
     def test_forever(self):
         # test --forever
         code = textwrap.dedent("""