1""" 2Tests of regrtest.py. 3 4Note: test_regrtest cannot be run twice in parallel. 5""" 6from __future__ import print_function 7 8import collections 9import errno 10import os.path 11import platform 12import re 13import subprocess 14import sys 15import sysconfig 16import tempfile 17import textwrap 18import unittest 19from test import support 20# Use utils alias to use the same code for TestUtils in master and 2.7 branches 21import regrtest as utils 22 23 24Py_DEBUG = hasattr(sys, 'getobjects') 25ROOT_DIR = os.path.join(os.path.dirname(__file__), '..', '..') 26ROOT_DIR = os.path.abspath(os.path.normpath(ROOT_DIR)) 27 28TEST_INTERRUPTED = textwrap.dedent(""" 29 from signal import SIGINT 30 try: 31 from _testcapi import raise_signal 32 raise_signal(SIGINT) 33 except ImportError: 34 import os 35 os.kill(os.getpid(), SIGINT) 36 """) 37 38 39SubprocessRun = collections.namedtuple('SubprocessRun', 40 'returncode stdout stderr') 41 42 43class BaseTestCase(unittest.TestCase): 44 TEST_UNIQUE_ID = 1 45 TESTNAME_PREFIX = 'test_regrtest_' 46 TESTNAME_REGEX = r'test_[a-zA-Z0-9_]+' 47 48 def setUp(self): 49 self.testdir = os.path.realpath(os.path.dirname(__file__)) 50 51 self.tmptestdir = tempfile.mkdtemp() 52 self.addCleanup(support.rmtree, self.tmptestdir) 53 54 def create_test(self, name=None, code=''): 55 if not name: 56 name = 'noop%s' % BaseTestCase.TEST_UNIQUE_ID 57 BaseTestCase.TEST_UNIQUE_ID += 1 58 59 # test_regrtest cannot be run twice in parallel because 60 # of setUp() and create_test() 61 name = self.TESTNAME_PREFIX + name 62 path = os.path.join(self.tmptestdir, name + '.py') 63 64 self.addCleanup(support.unlink, path) 65 # Use O_EXCL to ensure that we do not override existing tests 66 try: 67 fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL) 68 except OSError as exc: 69 if (exc.errno in (errno.EACCES, errno.EPERM) 70 and not sysconfig.is_python_build()): 71 self.skipTest("cannot write %s: %s" % (path, exc)) 72 else: 73 raise 74 else: 75 with os.fdopen(fd, 'w') as fp: 76 fp.write(code) 77 return name 78 79 def regex_search(self, regex, output): 80 match = re.search(regex, output, re.MULTILINE) 81 if not match: 82 self.fail("%r not found in %r" % (regex, output)) 83 return match 84 85 def check_line(self, output, regex): 86 regex = re.compile(r'^' + regex, re.MULTILINE) 87 self.assertRegexpMatches(output, regex) 88 89 def parse_executed_tests(self, output): 90 regex = (r'^[0-9]+:[0-9]+:[0-9]+ (?:load avg: [0-9]+\.[0-9]{2} )?\[ *[0-9]+(?:/ *[0-9]+)*\] (%s)' 91 % self.TESTNAME_REGEX) 92 parser = re.finditer(regex, output, re.MULTILINE) 93 return list(match.group(1) for match in parser) 94 95 def check_executed_tests(self, output, tests, skipped=(), failed=(), 96 env_changed=(), omitted=(), 97 rerun=(), 98 randomize=False, interrupted=False, 99 fail_env_changed=False): 100 if isinstance(tests, str): 101 tests = [tests] 102 if isinstance(skipped, str): 103 skipped = [skipped] 104 if isinstance(failed, str): 105 failed = [failed] 106 if isinstance(env_changed, str): 107 env_changed = [env_changed] 108 if isinstance(omitted, str): 109 omitted = [omitted] 110 if isinstance(rerun, str): 111 rerun = [rerun] 112 113 executed = self.parse_executed_tests(output) 114 if randomize: 115 self.assertEqual(set(executed), set(tests), output) 116 else: 117 self.assertEqual(executed, tests, (executed, tests, output)) 118 119 def plural(count): 120 return 's' if count != 1 else '' 121 122 def list_regex(line_format, tests): 123 count = len(tests) 124 names = ' '.join(sorted(tests)) 125 regex = line_format % (count, plural(count)) 126 regex = r'%s:\n %s$' % (regex, names) 127 return regex 128 129 if skipped: 130 regex = list_regex('%s test%s skipped', skipped) 131 self.check_line(output, regex) 132 133 if failed: 134 regex = list_regex('%s test%s failed', failed) 135 self.check_line(output, regex) 136 137 if env_changed: 138 regex = list_regex('%s test%s altered the execution environment', 139 env_changed) 140 self.check_line(output, regex) 141 142 if omitted: 143 regex = list_regex('%s test%s omitted', omitted) 144 self.check_line(output, regex) 145 146 if rerun: 147 regex = list_regex('%s re-run test%s', rerun) 148 self.check_line(output, regex) 149 self.check_line(output, "Re-running failed tests in verbose mode") 150 for name in rerun: 151 regex = "Re-running test %r in verbose mode" % name 152 self.check_line(output, regex) 153 154 good = (len(tests) - len(skipped) - len(failed) 155 - len(omitted) - len(env_changed)) 156 if good: 157 regex = r'%s test%s OK\.$' % (good, plural(good)) 158 if not skipped and not failed and good > 1: 159 regex = 'All %s' % regex 160 self.check_line(output, regex) 161 162 if interrupted: 163 self.check_line(output, 'Test suite interrupted by signal SIGINT.') 164 165 result = [] 166 if failed: 167 result.append('FAILURE') 168 elif fail_env_changed and env_changed: 169 result.append('ENV CHANGED') 170 if interrupted: 171 result.append('INTERRUPTED') 172 if not result: 173 result.append('SUCCESS') 174 result = ', '.join(result) 175 if rerun: 176 self.check_line(output, 'Tests result: %s' % result) 177 result = 'FAILURE then %s' % result 178 self.check_line(output, 'Tests result: %s' % result) 179 180 def parse_random_seed(self, output): 181 match = self.regex_search(r'Using random seed ([0-9]+)', output) 182 randseed = int(match.group(1)) 183 self.assertTrue(0 <= randseed <= 10000000, randseed) 184 return randseed 185 186 def run_command(self, args, input=None, exitcode=0, **kw): 187 if not input: 188 input = '' 189 if 'stderr' not in kw: 190 kw['stderr'] = subprocess.PIPE 191 proc = subprocess.Popen(args, 192 universal_newlines=True, 193 stdout=subprocess.PIPE, 194 **kw) 195 stdout, stderr = proc.communicate(input=input) 196 if proc.returncode != exitcode: 197 msg = ("Command %s failed with exit code %s\n" 198 "\n" 199 "stdout:\n" 200 "---\n" 201 "%s\n" 202 "---\n" 203 % (str(args), proc.returncode, stdout)) 204 if proc.stderr: 205 msg += ("\n" 206 "stderr:\n" 207 "---\n" 208 "%s" 209 "---\n" 210 % stderr) 211 self.fail(msg) 212 return SubprocessRun(proc.returncode, stdout, stderr) 213 214 def run_python(self, args, **kw): 215 args = [sys.executable] + list(args) 216 proc = self.run_command(args, **kw) 217 return proc.stdout 218 219 220class ProgramsTestCase(BaseTestCase): 221 """ 222 Test various ways to run the Python test suite. Use options close 223 to options used on the buildbot. 224 """ 225 226 NTEST = 4 227 228 def setUp(self): 229 super(ProgramsTestCase, self).setUp() 230 231 # Create NTEST tests doing nothing 232 self.tests = [self.create_test() for index in range(self.NTEST)] 233 234 self.python_args = ['-Wd', '-3', '-E', '-bb', '-tt'] 235 self.regrtest_args = ['-uall', '-rwW', 236 '--testdir=%s' % self.tmptestdir] 237 238 def check_output(self, output): 239 self.parse_random_seed(output) 240 self.check_executed_tests(output, self.tests, randomize=True) 241 242 def run_tests(self, args): 243 output = self.run_python(args) 244 self.check_output(output) 245 246 def test_script_regrtest(self): 247 # Lib/test/regrtest.py 248 script = os.path.join(self.testdir, 'regrtest.py') 249 250 args = self.python_args + [script] + self.regrtest_args + self.tests 251 self.run_tests(args) 252 253 def test_module_test(self): 254 # -m test 255 args = self.python_args + ['-m', 'test'] + self.regrtest_args + self.tests 256 self.run_tests(args) 257 258 def test_module_regrtest(self): 259 # -m test.regrtest 260 args = self.python_args + ['-m', 'test.regrtest'] + self.regrtest_args + self.tests 261 self.run_tests(args) 262 263 def test_module_autotest(self): 264 # -m test.autotest 265 args = self.python_args + ['-m', 'test.autotest'] + self.regrtest_args + self.tests 266 self.run_tests(args) 267 268 def test_module_from_test_autotest(self): 269 # from test import autotest 270 code = 'from test import autotest' 271 args = self.python_args + ['-c', code] + self.regrtest_args + self.tests 272 self.run_tests(args) 273 274 def test_script_autotest(self): 275 # Lib/test/autotest.py 276 script = os.path.join(self.testdir, 'autotest.py') 277 args = self.python_args + [script] + self.regrtest_args + self.tests 278 self.run_tests(args) 279 280 def run_batch(self, *args): 281 proc = self.run_command(args) 282 self.check_output(proc.stdout) 283 284 def need_pcbuild(self): 285 exe = os.path.normpath(os.path.abspath(sys.executable)) 286 parts = exe.split(os.path.sep) 287 if len(parts) < 3: 288 # it's not a python build, python is likely to be installed 289 return 290 291 build_dir = parts[-3] 292 if build_dir.lower() != 'pcbuild': 293 self.skipTest("Tools/buildbot/test.bat requires PCbuild build, " 294 "found %s" % build_dir) 295 296 @unittest.skipUnless(sysconfig.is_python_build(), 297 'test.bat script is not installed') 298 @unittest.skipUnless(sys.platform == 'win32', 'Windows only') 299 def test_tools_buildbot_test(self): 300 self.need_pcbuild() 301 302 # Tools\buildbot\test.bat 303 script = os.path.join(ROOT_DIR, 'Tools', 'buildbot', 'test.bat') 304 test_args = ['--testdir=%s' % self.tmptestdir] 305 if platform.architecture()[0] == '64bit': 306 test_args.append('-x64') # 64-bit build 307 if not Py_DEBUG: 308 test_args.append('+d') # Release build, use python.exe 309 310 args = [script] + test_args + self.tests 311 self.run_batch(*args) 312 313 @unittest.skipUnless(sys.platform == 'win32', 'Windows only') 314 def test_pcbuild_rt(self): 315 self.need_pcbuild() 316 317 # PCbuild\rt.bat 318 script = os.path.join(ROOT_DIR, r'PCbuild\rt.bat') 319 rt_args = ["-q"] # Quick, don't run tests twice 320 if platform.architecture()[0] == '64bit': 321 rt_args.append('-x64') # 64-bit build 322 if Py_DEBUG: 323 rt_args.append('-d') # Debug build, use python_d.exe 324 args = [script] + rt_args + self.regrtest_args + self.tests 325 self.run_batch(*args) 326 327 328class ArgsTestCase(BaseTestCase): 329 """ 330 Test arguments of the Python test suite. 331 """ 332 333 def run_tests(self, *testargs, **kw): 334 cmdargs = ('-m', 'test', '--testdir=%s' % self.tmptestdir) + testargs 335 return self.run_python(cmdargs, **kw) 336 337 def test_failing_test(self): 338 # test a failing test 339 code = textwrap.dedent(""" 340 import unittest 341 from test import support 342 343 class FailingTest(unittest.TestCase): 344 def test_failing(self): 345 self.fail("bug") 346 347 def test_main(): 348 support.run_unittest(FailingTest) 349 """) 350 test_ok = self.create_test('ok') 351 test_failing = self.create_test('failing', code=code) 352 tests = [test_ok, test_failing] 353 354 output = self.run_tests(*tests, exitcode=2) 355 self.check_executed_tests(output, tests, failed=test_failing) 356 357 def test_resources(self): 358 # test -u command line option 359 tests = {} 360 for resource in ('audio', 'network'): 361 code = 'from test import support\nsupport.requires(%r)' % resource 362 tests[resource] = self.create_test(resource, code) 363 test_names = sorted(tests.values()) 364 365 # -u all: 2 resources enabled 366 output = self.run_tests('-u', 'all', *test_names) 367 self.check_executed_tests(output, test_names) 368 369 # -u audio: 1 resource enabled 370 output = self.run_tests('-uaudio', *test_names) 371 self.check_executed_tests(output, test_names, 372 skipped=tests['network']) 373 374 # no option: 0 resources enabled 375 output = self.run_tests(*test_names) 376 self.check_executed_tests(output, test_names, 377 skipped=test_names) 378 379 def test_random(self): 380 # test -r and --randseed command line option 381 code = textwrap.dedent(""" 382 import random 383 print("TESTRANDOM: %s" % random.randint(1, 1000)) 384 """) 385 test = self.create_test('random', code) 386 387 # first run to get the output with the random seed 388 output = self.run_tests('-r', '-v', test) 389 randseed = self.parse_random_seed(output) 390 match = self.regex_search(r'TESTRANDOM: ([0-9]+)', output) 391 test_random = int(match.group(1)) 392 393 # try to reproduce with the random seed 394 output = self.run_tests('-r', '-v', '--randseed=%s' % randseed, test) 395 randseed2 = self.parse_random_seed(output) 396 self.assertEqual(randseed2, randseed) 397 398 match = self.regex_search(r'TESTRANDOM: ([0-9]+)', output) 399 test_random2 = int(match.group(1)) 400 self.assertEqual(test_random2, test_random) 401 402 def test_fromfile(self): 403 # test --fromfile 404 tests = [self.create_test() for index in range(5)] 405 406 # Write the list of files using a format similar to regrtest output: 407 # [1/2] test_1 408 # [2/2] test_2 409 filename = support.TESTFN 410 self.addCleanup(support.unlink, filename) 411 412 # test format 'test_opcodes' 413 with open(filename, "w") as fp: 414 for name in tests: 415 print(name, file=fp) 416 417 output = self.run_tests('--fromfile', filename) 418 self.check_executed_tests(output, tests) 419 420 def test_interrupted(self): 421 code = TEST_INTERRUPTED 422 test = self.create_test('sigint', code=code) 423 output = self.run_tests(test, exitcode=130) 424 self.check_executed_tests(output, test, omitted=test, 425 interrupted=True) 426 427 def test_slowest(self): 428 # test --slow 429 tests = [self.create_test() for index in range(3)] 430 output = self.run_tests("--slowest", *tests) 431 self.check_executed_tests(output, tests) 432 regex = ('10 slowest tests:\n' 433 '(?:- %s: .*\n){%s}' 434 % (self.TESTNAME_REGEX, len(tests))) 435 self.check_line(output, regex) 436 437 def test_slow_interrupted(self): 438 # Issue #25373: test --slowest with an interrupted test 439 code = TEST_INTERRUPTED 440 test = self.create_test("sigint", code=code) 441 442 try: 443 import threading 444 tests = (False, True) 445 except ImportError: 446 tests = (False,) 447 for multiprocessing in tests: 448 if multiprocessing: 449 args = ("--slowest", "-j2", test) 450 else: 451 args = ("--slowest", test) 452 output = self.run_tests(*args, exitcode=130) 453 self.check_executed_tests(output, test, 454 omitted=test, interrupted=True) 455 456 regex = ('10 slowest tests:\n') 457 self.check_line(output, regex) 458 459 def test_coverage(self): 460 # test --coverage 461 test = self.create_test('coverage') 462 output = self.run_tests("--coverage", test) 463 self.check_executed_tests(output, [test]) 464 regex = (r'lines +cov% +module +\(path\)\n' 465 r'(?: *[0-9]+ *[0-9]{1,2}% *[^ ]+ +\([^)]+\)+)+') 466 self.check_line(output, regex) 467 468 def test_forever(self): 469 # test --forever 470 code = textwrap.dedent(""" 471 import __builtin__ 472 import unittest 473 from test import support 474 475 class ForeverTester(unittest.TestCase): 476 def test_run(self): 477 # Store the state in the __builtin__ module, because the test 478 # module is reload at each run 479 if 'RUN' in __builtin__.__dict__: 480 __builtin__.__dict__['RUN'] += 1 481 if __builtin__.__dict__['RUN'] >= 3: 482 self.fail("fail at the 3rd runs") 483 else: 484 __builtin__.__dict__['RUN'] = 1 485 486 def test_main(): 487 support.run_unittest(ForeverTester) 488 """) 489 test = self.create_test('forever', code=code) 490 output = self.run_tests('--forever', test, exitcode=2) 491 self.check_executed_tests(output, [test]*3, failed=test) 492 493 def check_leak(self, code, what): 494 test = self.create_test('huntrleaks', code=code) 495 496 filename = 'reflog.txt' 497 self.addCleanup(support.unlink, filename) 498 output = self.run_tests('--huntrleaks', '3:3:', test, 499 exitcode=2, 500 stderr=subprocess.STDOUT) 501 self.check_executed_tests(output, [test], failed=test) 502 503 line = 'beginning 6 repetitions\n123456\n......\n' 504 self.check_line(output, re.escape(line)) 505 506 line2 = '%s leaked [1, 1, 1] %s, sum=3\n' % (test, what) 507 self.assertIn(line2, output) 508 509 with open(filename) as fp: 510 reflog = fp.read() 511 self.assertIn(line2, reflog) 512 513 @unittest.skipUnless(Py_DEBUG, 'need a debug build') 514 @support.requires_type_collecting 515 def test_huntrleaks(self): 516 # test --huntrleaks 517 code = textwrap.dedent(""" 518 import unittest 519 from test import support 520 521 GLOBAL_LIST = [] 522 523 class RefLeakTest(unittest.TestCase): 524 def test_leak(self): 525 GLOBAL_LIST.append(object()) 526 527 def test_main(): 528 support.run_unittest(RefLeakTest) 529 """) 530 self.check_leak(code, 'references') 531 532 @unittest.skipUnless(Py_DEBUG, 'need a debug build') 533 def test_huntrleaks_fd_leak(self): 534 # test --huntrleaks for file descriptor leak 535 code = textwrap.dedent(""" 536 import os 537 import unittest 538 from test import support 539 540 class FDLeakTest(unittest.TestCase): 541 def test_leak(self): 542 fd = os.open(__file__, os.O_RDONLY) 543 # bug: never close the file descriptor 544 545 def test_main(): 546 support.run_unittest(FDLeakTest) 547 """) 548 self.check_leak(code, 'file descriptors') 549 550 def test_list_tests(self): 551 # test --list-tests 552 tests = [self.create_test() for i in range(5)] 553 output = self.run_tests('--list-tests', *tests) 554 self.assertEqual(output.rstrip().splitlines(), 555 tests) 556 557 def test_list_cases(self): 558 # test --list-cases 559 code = textwrap.dedent(""" 560 import unittest 561 562 class Tests(unittest.TestCase): 563 def test_method1(self): 564 pass 565 def test_method2(self): 566 pass 567 """) 568 testname = self.create_test(code=code) 569 570 # Test --list-cases 571 all_methods = ['%s.Tests.test_method1' % testname, 572 '%s.Tests.test_method2' % testname] 573 output = self.run_tests('--list-cases', testname) 574 self.assertEqual(output.splitlines(), all_methods) 575 576 # Test --list-cases with --match 577 all_methods = ['%s.Tests.test_method1' % testname] 578 output = self.run_tests('--list-cases', 579 '-m', 'test_method1', 580 testname) 581 self.assertEqual(output.splitlines(), all_methods) 582 583 @unittest.skipIf(sys.platform.startswith('aix'), 584 "support._crash_python() doesn't work on AIX") 585 def test_crashed(self): 586 # Any code which causes a crash 587 code = 'import test.support; test.support._crash_python()' 588 crash_test = self.create_test(name="crash", code=code) 589 ok_test = self.create_test(name="ok") 590 591 tests = [crash_test, ok_test] 592 output = self.run_tests("-j2", *tests, exitcode=2) 593 self.check_executed_tests(output, tests, failed=crash_test, 594 randomize=True) 595 596 def parse_methods(self, output): 597 regex = re.compile("^(test[^ ]+).*ok$", flags=re.MULTILINE) 598 return [match.group(1) for match in regex.finditer(output)] 599 600 def test_matchfile(self): 601 # Any code which causes a crash 602 code = textwrap.dedent(""" 603 import unittest 604 from test import support 605 606 class Tests(unittest.TestCase): 607 def test_method1(self): 608 pass 609 def test_method2(self): 610 pass 611 def test_method3(self): 612 pass 613 def test_method4(self): 614 pass 615 616 def test_main(): 617 support.run_unittest(Tests) 618 """) 619 all_methods = ['test_method1', 'test_method2', 620 'test_method3', 'test_method4'] 621 testname = self.create_test(code=code) 622 623 # by default, all methods should be run 624 output = self.run_tests("-v", testname) 625 methods = self.parse_methods(output) 626 self.assertEqual(methods, all_methods) 627 628 # only run a subset 629 filename = support.TESTFN 630 self.addCleanup(support.unlink, filename) 631 632 subset = [ 633 # only match the method name 634 'test_method1', 635 # match the full identifier 636 '%s.Tests.test_method3' % testname] 637 with open(filename, "w") as fp: 638 for name in subset: 639 print(name, file=fp) 640 641 output = self.run_tests("-v", "--matchfile", filename, testname) 642 methods = self.parse_methods(output) 643 subset = ['test_method1', 'test_method3'] 644 self.assertEqual(methods, subset) 645 646 def test_env_changed(self): 647 code = textwrap.dedent(""" 648 import unittest 649 from test import support 650 651 class Tests(unittest.TestCase): 652 def test_env_changed(self): 653 open("env_changed", "w").close() 654 655 def test_main(): 656 support.run_unittest(Tests) 657 """) 658 testname = self.create_test(code=code) 659 660 # don't fail by default 661 output = self.run_tests(testname) 662 self.check_executed_tests(output, [testname], env_changed=testname) 663 664 # fail with --fail-env-changed 665 output = self.run_tests("--fail-env-changed", testname, exitcode=3) 666 self.check_executed_tests(output, [testname], env_changed=testname, 667 fail_env_changed=True) 668 669 def test_rerun_fail(self): 670 code = textwrap.dedent(""" 671 import unittest 672 673 class Tests(unittest.TestCase): 674 def test_bug(self): 675 # test always fail 676 self.fail("bug") 677 678 def test_main(): 679 support.run_unittest(Tests) 680 """) 681 testname = self.create_test(code=code) 682 683 output = self.run_tests("-w", testname, exitcode=2) 684 self.check_executed_tests(output, [testname], 685 failed=testname, rerun=testname) 686 687 688class TestUtils(unittest.TestCase): 689 def test_format_duration(self): 690 self.assertEqual(utils.format_duration(0), 691 '0 ms') 692 self.assertEqual(utils.format_duration(1e-9), 693 '1 ms') 694 self.assertEqual(utils.format_duration(10e-3), 695 '10 ms') 696 self.assertEqual(utils.format_duration(1.5), 697 '1 sec 500 ms') 698 self.assertEqual(utils.format_duration(1), 699 '1 sec') 700 self.assertEqual(utils.format_duration(2 * 60), 701 '2 min') 702 self.assertEqual(utils.format_duration(2 * 60 + 1), 703 '2 min 1 sec') 704 self.assertEqual(utils.format_duration(3 * 3600), 705 '3 hour') 706 self.assertEqual(utils.format_duration(3 * 3600 + 2 * 60 + 1), 707 '3 hour 2 min') 708 self.assertEqual(utils.format_duration(3 * 3600 + 1), 709 '3 hour 1 sec') 710 711 712def test_main(): 713 support.run_unittest(ProgramsTestCase, ArgsTestCase, TestUtils) 714 715 716if __name__ == "__main__": 717 test_main() 718