""" Tests of regrtest.py. Note: test_regrtest cannot be run twice in parallel. """ from __future__ import print_function import collections import errno import os.path import platform import re import subprocess import sys import sysconfig import tempfile import textwrap import unittest from test import support # Use utils alias to use the same code for TestUtils in master and 2.7 branches import regrtest as utils Py_DEBUG = hasattr(sys, 'getobjects') ROOT_DIR = os.path.join(os.path.dirname(__file__), '..', '..') ROOT_DIR = os.path.abspath(os.path.normpath(ROOT_DIR)) TEST_INTERRUPTED = textwrap.dedent(""" from signal import SIGINT try: from _testcapi import raise_signal raise_signal(SIGINT) except ImportError: import os os.kill(os.getpid(), SIGINT) """) SubprocessRun = collections.namedtuple('SubprocessRun', 'returncode stdout stderr') class BaseTestCase(unittest.TestCase): TEST_UNIQUE_ID = 1 TESTNAME_PREFIX = 'test_regrtest_' TESTNAME_REGEX = r'test_[a-zA-Z0-9_]+' def setUp(self): self.testdir = os.path.realpath(os.path.dirname(__file__)) self.tmptestdir = tempfile.mkdtemp() self.addCleanup(support.rmtree, self.tmptestdir) def create_test(self, name=None, code=''): if not name: name = 'noop%s' % BaseTestCase.TEST_UNIQUE_ID BaseTestCase.TEST_UNIQUE_ID += 1 # test_regrtest cannot be run twice in parallel because # of setUp() and create_test() name = self.TESTNAME_PREFIX + name path = os.path.join(self.tmptestdir, name + '.py') self.addCleanup(support.unlink, path) # Use O_EXCL to ensure that we do not override existing tests try: fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL) except OSError as exc: if (exc.errno in (errno.EACCES, errno.EPERM) and not sysconfig.is_python_build()): self.skipTest("cannot write %s: %s" % (path, exc)) else: raise else: with os.fdopen(fd, 'w') as fp: fp.write(code) return name def regex_search(self, regex, output): match = re.search(regex, output, re.MULTILINE) if not match: self.fail("%r not found in %r" % (regex, output)) return match def check_line(self, output, regex): regex = re.compile(r'^' + regex, re.MULTILINE) self.assertRegexpMatches(output, regex) def parse_executed_tests(self, output): regex = (r'^[0-9]+:[0-9]+:[0-9]+ (?:load avg: [0-9]+\.[0-9]{2} )?\[ *[0-9]+(?:/ *[0-9]+)*\] (%s)' % self.TESTNAME_REGEX) parser = re.finditer(regex, output, re.MULTILINE) return list(match.group(1) for match in parser) def check_executed_tests(self, output, tests, skipped=(), failed=(), env_changed=(), omitted=(), rerun=(), randomize=False, interrupted=False, fail_env_changed=False): if isinstance(tests, str): tests = [tests] if isinstance(skipped, str): skipped = [skipped] if isinstance(failed, str): failed = [failed] if isinstance(env_changed, str): env_changed = [env_changed] if isinstance(omitted, str): omitted = [omitted] if isinstance(rerun, str): rerun = [rerun] executed = self.parse_executed_tests(output) if randomize: self.assertEqual(set(executed), set(tests), output) else: self.assertEqual(executed, tests, (executed, tests, output)) def plural(count): return 's' if count != 1 else '' def list_regex(line_format, tests): count = len(tests) names = ' '.join(sorted(tests)) regex = line_format % (count, plural(count)) regex = r'%s:\n %s$' % (regex, names) return regex if skipped: regex = list_regex('%s test%s skipped', skipped) self.check_line(output, regex) if failed: regex = list_regex('%s test%s failed', failed) self.check_line(output, regex) if env_changed: regex = list_regex('%s test%s altered the execution environment', env_changed) self.check_line(output, regex) if omitted: regex = list_regex('%s test%s omitted', omitted) self.check_line(output, regex) if rerun: regex = list_regex('%s re-run test%s', rerun) self.check_line(output, regex) self.check_line(output, "Re-running failed tests in verbose mode") for name in rerun: regex = "Re-running test %r in verbose mode" % name self.check_line(output, regex) good = (len(tests) - len(skipped) - len(failed) - len(omitted) - len(env_changed)) if good: regex = r'%s test%s OK\.$' % (good, plural(good)) if not skipped and not failed and good > 1: regex = 'All %s' % regex self.check_line(output, regex) if interrupted: self.check_line(output, 'Test suite interrupted by signal SIGINT.') result = [] if failed: result.append('FAILURE') elif fail_env_changed and env_changed: result.append('ENV CHANGED') if interrupted: result.append('INTERRUPTED') if not result: result.append('SUCCESS') result = ', '.join(result) if rerun: self.check_line(output, 'Tests result: %s' % result) result = 'FAILURE then %s' % result self.check_line(output, 'Tests result: %s' % result) def parse_random_seed(self, output): match = self.regex_search(r'Using random seed ([0-9]+)', output) randseed = int(match.group(1)) self.assertTrue(0 <= randseed <= 10000000, randseed) return randseed def run_command(self, args, input=None, exitcode=0, **kw): if not input: input = '' if 'stderr' not in kw: kw['stderr'] = subprocess.PIPE proc = subprocess.Popen(args, universal_newlines=True, stdout=subprocess.PIPE, **kw) stdout, stderr = proc.communicate(input=input) if proc.returncode != exitcode: msg = ("Command %s failed with exit code %s\n" "\n" "stdout:\n" "---\n" "%s\n" "---\n" % (str(args), proc.returncode, stdout)) if proc.stderr: msg += ("\n" "stderr:\n" "---\n" "%s" "---\n" % stderr) self.fail(msg) return SubprocessRun(proc.returncode, stdout, stderr) def run_python(self, args, **kw): args = [sys.executable] + list(args) proc = self.run_command(args, **kw) return proc.stdout class ProgramsTestCase(BaseTestCase): """ Test various ways to run the Python test suite. Use options close to options used on the buildbot. """ NTEST = 4 def setUp(self): super(ProgramsTestCase, self).setUp() # Create NTEST tests doing nothing self.tests = [self.create_test() for index in range(self.NTEST)] self.python_args = ['-Wd', '-3', '-E', '-bb', '-tt'] self.regrtest_args = ['-uall', '-rwW', '--testdir=%s' % self.tmptestdir] def check_output(self, output): self.parse_random_seed(output) self.check_executed_tests(output, self.tests, randomize=True) def run_tests(self, args): output = self.run_python(args) self.check_output(output) def test_script_regrtest(self): # Lib/test/regrtest.py script = os.path.join(self.testdir, 'regrtest.py') args = self.python_args + [script] + self.regrtest_args + self.tests self.run_tests(args) def test_module_test(self): # -m test args = self.python_args + ['-m', 'test'] + self.regrtest_args + self.tests self.run_tests(args) def test_module_regrtest(self): # -m test.regrtest args = self.python_args + ['-m', 'test.regrtest'] + self.regrtest_args + self.tests self.run_tests(args) def test_module_autotest(self): # -m test.autotest args = self.python_args + ['-m', 'test.autotest'] + self.regrtest_args + self.tests self.run_tests(args) def test_module_from_test_autotest(self): # from test import autotest code = 'from test import autotest' args = self.python_args + ['-c', code] + self.regrtest_args + self.tests self.run_tests(args) def test_script_autotest(self): # Lib/test/autotest.py script = os.path.join(self.testdir, 'autotest.py') args = self.python_args + [script] + self.regrtest_args + self.tests self.run_tests(args) def run_batch(self, *args): proc = self.run_command(args) self.check_output(proc.stdout) 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(sysconfig.is_python_build(), 'test.bat script is not installed') @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] if platform.architecture()[0] == '64bit': test_args.append('-x64') # 64-bit build if not Py_DEBUG: test_args.append('+d') # Release build, use python.exe args = [script] + test_args + self.tests self.run_batch(*args) @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 if platform.architecture()[0] == '64bit': rt_args.append('-x64') # 64-bit build if Py_DEBUG: rt_args.append('-d') # Debug build, use python_d.exe args = [script] + rt_args + self.regrtest_args + self.tests self.run_batch(*args) class ArgsTestCase(BaseTestCase): """ Test arguments of the Python test suite. """ def run_tests(self, *testargs, **kw): cmdargs = ('-m', 'test', '--testdir=%s' % self.tmptestdir) + testargs return self.run_python(cmdargs, **kw) def test_failing_test(self): # test a failing test code = textwrap.dedent(""" import unittest from test import support class FailingTest(unittest.TestCase): def test_failing(self): self.fail("bug") def test_main(): support.run_unittest(FailingTest) """) test_ok = self.create_test('ok') test_failing = self.create_test('failing', code=code) tests = [test_ok, test_failing] output = self.run_tests(*tests, exitcode=2) self.check_executed_tests(output, tests, failed=test_failing) def test_resources(self): # test -u command line option tests = {} for resource in ('audio', 'network'): code = 'from test import support\nsupport.requires(%r)' % resource tests[resource] = self.create_test(resource, code) test_names = sorted(tests.values()) # -u all: 2 resources enabled output = self.run_tests('-u', 'all', *test_names) self.check_executed_tests(output, test_names) # -u audio: 1 resource enabled output = self.run_tests('-uaudio', *test_names) self.check_executed_tests(output, test_names, skipped=tests['network']) # no option: 0 resources enabled output = self.run_tests(*test_names) self.check_executed_tests(output, test_names, skipped=test_names) def test_random(self): # test -r and --randseed command line option code = textwrap.dedent(""" import random print("TESTRANDOM: %s" % random.randint(1, 1000)) """) test = self.create_test('random', code) # first run to get the output with the random seed output = self.run_tests('-r', '-v', test) randseed = self.parse_random_seed(output) match = self.regex_search(r'TESTRANDOM: ([0-9]+)', output) test_random = int(match.group(1)) # try to reproduce with the random seed output = self.run_tests('-r', '-v', '--randseed=%s' % randseed, test) randseed2 = self.parse_random_seed(output) self.assertEqual(randseed2, randseed) match = self.regex_search(r'TESTRANDOM: ([0-9]+)', output) test_random2 = int(match.group(1)) self.assertEqual(test_random2, test_random) def test_fromfile(self): # test --fromfile tests = [self.create_test() for index in range(5)] # Write the list of files using a format similar to regrtest output: # [1/2] test_1 # [2/2] test_2 filename = support.TESTFN self.addCleanup(support.unlink, filename) # test format 'test_opcodes' with open(filename, "w") as fp: for name in tests: print(name, file=fp) output = self.run_tests('--fromfile', filename) self.check_executed_tests(output, tests) def test_interrupted(self): code = TEST_INTERRUPTED test = self.create_test('sigint', code=code) output = self.run_tests(test, exitcode=130) self.check_executed_tests(output, test, omitted=test, interrupted=True) def test_slowest(self): # test --slow tests = [self.create_test() for index in range(3)] output = self.run_tests("--slowest", *tests) self.check_executed_tests(output, tests) regex = ('10 slowest tests:\n' '(?:- %s: .*\n){%s}' % (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=130) 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(""" import __builtin__ import unittest from test import support class ForeverTester(unittest.TestCase): def test_run(self): # Store the state in the __builtin__ module, because the test # module is reload at each run if 'RUN' in __builtin__.__dict__: __builtin__.__dict__['RUN'] += 1 if __builtin__.__dict__['RUN'] >= 3: self.fail("fail at the 3rd runs") else: __builtin__.__dict__['RUN'] = 1 def test_main(): support.run_unittest(ForeverTester) """) test = self.create_test('forever', code=code) output = self.run_tests('--forever', test, exitcode=2) self.check_executed_tests(output, [test]*3, failed=test) def check_leak(self, code, what): test = self.create_test('huntrleaks', code=code) filename = 'reflog.txt' self.addCleanup(support.unlink, filename) output = self.run_tests('--huntrleaks', '3:3:', test, exitcode=2, stderr=subprocess.STDOUT) self.check_executed_tests(output, [test], failed=test) line = 'beginning 6 repetitions\n123456\n......\n' self.check_line(output, re.escape(line)) line2 = '%s leaked [1, 1, 1] %s, sum=3\n' % (test, what) self.assertIn(line2, output) with open(filename) as fp: reflog = fp.read() self.assertIn(line2, reflog) @unittest.skipUnless(Py_DEBUG, 'need a debug build') @support.requires_type_collecting def test_huntrleaks(self): # test --huntrleaks code = textwrap.dedent(""" import unittest from test import support GLOBAL_LIST = [] class RefLeakTest(unittest.TestCase): def test_leak(self): GLOBAL_LIST.append(object()) def test_main(): support.run_unittest(RefLeakTest) """) self.check_leak(code, 'references') @unittest.skipUnless(Py_DEBUG, 'need a debug build') def test_huntrleaks_fd_leak(self): # test --huntrleaks for file descriptor leak code = textwrap.dedent(""" import os import unittest from test import support class FDLeakTest(unittest.TestCase): def test_leak(self): fd = os.open(__file__, os.O_RDONLY) # bug: never close the file descriptor def test_main(): support.run_unittest(FDLeakTest) """) self.check_leak(code, 'file descriptors') def test_list_tests(self): # test --list-tests tests = [self.create_test() for i in range(5)] output = self.run_tests('--list-tests', *tests) self.assertEqual(output.rstrip().splitlines(), tests) def test_list_cases(self): # test --list-cases code = textwrap.dedent(""" import unittest class Tests(unittest.TestCase): def test_method1(self): pass def test_method2(self): pass """) testname = self.create_test(code=code) # Test --list-cases all_methods = ['%s.Tests.test_method1' % testname, '%s.Tests.test_method2' % testname] output = self.run_tests('--list-cases', testname) self.assertEqual(output.splitlines(), all_methods) # Test --list-cases with --match all_methods = ['%s.Tests.test_method1' % testname] output = self.run_tests('--list-cases', '-m', 'test_method1', testname) self.assertEqual(output.splitlines(), all_methods) @unittest.skipIf(sys.platform.startswith('aix'), "support._crash_python() doesn't work on AIX") def test_crashed(self): # Any code which causes a crash code = 'import test.support; test.support._crash_python()' crash_test = self.create_test(name="crash", code=code) ok_test = self.create_test(name="ok") tests = [crash_test, ok_test] output = self.run_tests("-j2", *tests, exitcode=2) self.check_executed_tests(output, tests, failed=crash_test, randomize=True) def parse_methods(self, output): regex = re.compile("^(test[^ ]+).*ok$", flags=re.MULTILINE) return [match.group(1) for match in regex.finditer(output)] def test_matchfile(self): # Any code which causes a crash code = textwrap.dedent(""" import unittest from test import support class Tests(unittest.TestCase): def test_method1(self): pass def test_method2(self): pass def test_method3(self): pass def test_method4(self): pass def test_main(): support.run_unittest(Tests) """) all_methods = ['test_method1', 'test_method2', 'test_method3', 'test_method4'] testname = self.create_test(code=code) # by default, all methods should be run output = self.run_tests("-v", testname) methods = self.parse_methods(output) self.assertEqual(methods, all_methods) # only run a subset filename = support.TESTFN self.addCleanup(support.unlink, filename) subset = [ # only match the method name 'test_method1', # match the full identifier '%s.Tests.test_method3' % testname] with open(filename, "w") as fp: for name in subset: print(name, file=fp) output = self.run_tests("-v", "--matchfile", filename, testname) methods = self.parse_methods(output) subset = ['test_method1', 'test_method3'] self.assertEqual(methods, subset) def test_env_changed(self): code = textwrap.dedent(""" import unittest from test import support class Tests(unittest.TestCase): def test_env_changed(self): open("env_changed", "w").close() def test_main(): support.run_unittest(Tests) """) testname = self.create_test(code=code) # don't fail by default output = self.run_tests(testname) self.check_executed_tests(output, [testname], env_changed=testname) # fail with --fail-env-changed output = self.run_tests("--fail-env-changed", testname, exitcode=3) self.check_executed_tests(output, [testname], env_changed=testname, fail_env_changed=True) def test_rerun_fail(self): code = textwrap.dedent(""" import unittest class Tests(unittest.TestCase): def test_bug(self): # test always fail self.fail("bug") def test_main(): support.run_unittest(Tests) """) testname = self.create_test(code=code) output = self.run_tests("-w", testname, exitcode=2) self.check_executed_tests(output, [testname], failed=testname, rerun=testname) class TestUtils(unittest.TestCase): def test_format_duration(self): self.assertEqual(utils.format_duration(0), '0 ms') self.assertEqual(utils.format_duration(1e-9), '1 ms') self.assertEqual(utils.format_duration(10e-3), '10 ms') self.assertEqual(utils.format_duration(1.5), '1 sec 500 ms') self.assertEqual(utils.format_duration(1), '1 sec') self.assertEqual(utils.format_duration(2 * 60), '2 min') self.assertEqual(utils.format_duration(2 * 60 + 1), '2 min 1 sec') self.assertEqual(utils.format_duration(3 * 3600), '3 hour') self.assertEqual(utils.format_duration(3 * 3600 + 2 * 60 + 1), '3 hour 2 min') self.assertEqual(utils.format_duration(3 * 3600 + 1), '3 hour 1 sec') def test_main(): support.run_unittest(ProgramsTestCase, ArgsTestCase, TestUtils) if __name__ == "__main__": test_main()