| #!/usr/bin/env python |
| #------------------------------------------------------------------------------------------------------- |
| # Copyright (C) Microsoft. All rights reserved. |
| # Licensed under the MIT license. See LICENSE.txt file in the project root for full license information. |
| #------------------------------------------------------------------------------------------------------- |
| |
| from __future__ import print_function |
| from datetime import datetime |
| from multiprocessing import Pool, Manager |
| from threading import Timer |
| import sys |
| import os |
| import subprocess as SP |
| import traceback |
| import argparse |
| import xml.etree.ElementTree as ET |
| import re |
| import time |
| |
| # handle command line args |
| parser = argparse.ArgumentParser( |
| description='ChakraCore *nix Test Script', |
| formatter_class=argparse.RawDescriptionHelpFormatter, |
| epilog='''\ |
| Samples: |
| |
| test all folders: |
| runtests.py |
| |
| test only Array: |
| runtests.py Array |
| |
| test a single file: |
| runtests.py Basics/hello.js |
| ''') |
| |
| parser.add_argument('folders', metavar='folder', nargs='*', |
| help='folder subset to run tests') |
| parser.add_argument('-b', '--binary', metavar='bin', help='ch full path') |
| parser.add_argument('-d', '--debug', action='store_true', |
| help='use debug build'); |
| parser.add_argument('-t', '--test', action='store_true', help='use test build') |
| parser.add_argument('--include-slow', action='store_true', |
| help='include slow tests') |
| parser.add_argument('--only-slow', action='store_true', |
| help='run only slow tests') |
| parser.add_argument('--nightly', action='store_true', |
| help='run as nightly tests') |
| parser.add_argument('--tag', nargs='*', |
| help='select tests with given tags') |
| parser.add_argument('--not-tag', nargs='*', |
| help='exclude tests with given tags') |
| parser.add_argument('--timeout', type=int, default=60, |
| help='test timeout (default 60 seconds)') |
| parser.add_argument('-l', '--logfile', metavar='logfile', help='file to log results to', default=None) |
| parser.add_argument('--x86', action='store_true', help='use x86 build') |
| parser.add_argument('--x64', action='store_true', help='use x64 build') |
| args = parser.parse_args() |
| |
| |
| test_root = os.path.dirname(os.path.realpath(__file__)) |
| repo_root = os.path.dirname(test_root) |
| |
| # arch: x86, x64 |
| arch = 'x86' if args.x86 else ('x64' if args.x64 else None) |
| if arch == None: |
| arch = os.environ.get('_BuildArch', 'x86') |
| if sys.platform != 'win32': |
| arch = 'x64' # xplat: hard code arch == x64 |
| arch_alias = 'amd64' if arch == 'x64' else None |
| |
| # flavor: debug, test, release |
| type_flavor = {'chk':'Debug', 'test':'Test', 'fre':'Release'} |
| flavor = 'Debug' if args.debug else ('Test' if args.test else None) |
| if flavor == None: |
| print("ERROR: Test build target wasn't defined.") |
| print("Try '-t' (test build) or '-d' (debug build).") |
| sys.exit(1) |
| flavor_alias = 'chk' if flavor == 'Debug' else 'fre' |
| |
| # binary: full ch path |
| binary = args.binary |
| if binary == None: |
| if sys.platform == 'win32': |
| binary = 'Build/VcBuild/bin/{}_{}/ch.exe'.format(arch, flavor) |
| else: |
| binary = 'BuildLinux/{0}/ch'.format(flavor) |
| binary = os.path.join(repo_root, binary) |
| if not os.path.isfile(binary): |
| print('{} not found. Did you run ./build.sh already?'.format(binary)) |
| sys.exit(1) |
| |
| # global tags/not_tags |
| tags = set(args.tag or []) |
| not_tags = set(args.not_tag or []).union(['fail', 'exclude_' + arch]) |
| |
| if arch_alias: |
| not_tags.add('exclude_' + arch_alias) |
| if flavor_alias: |
| not_tags.add('exclude_' + flavor_alias) |
| if args.only_slow: |
| tags.add('Slow') |
| elif not args.include_slow: |
| not_tags.add('Slow') |
| |
| not_tags.add('exclude_nightly' if args.nightly else 'nightly') |
| |
| # xplat: temp hard coded to exclude unsupported tests |
| if sys.platform != 'win32': |
| not_tags.add('exclude_xplat') |
| not_tags.add('Intl') |
| not_tags.add('require_backend') |
| not_tags.add('require_debugger') |
| if sys.platform == 'darwin': |
| not_tags.add('exclude_mac') |
| not_compile_flags = set(['-simdjs']) \ |
| if sys.platform != 'win32' else None |
| |
| # use tags/not_tags/not_compile_flags as case-insensitive |
| def lower_set(s): |
| return set([x.lower() for x in s] if s else []) |
| |
| tags = lower_set(tags) |
| not_tags = lower_set(not_tags) |
| not_compile_flags = lower_set(not_compile_flags) |
| |
| # split tags text into tags set |
| _empty_set = set() |
| def split_tags(text): |
| return set(x.strip() for x in text.lower().split(',')) if text \ |
| else _empty_set |
| |
| class LogFile(object): |
| def __init__(self, log_file_path = None): |
| self.file = None |
| |
| if log_file_path is None: |
| # Set up the log file paths |
| # Make sure the right directory exists and the log file doesn't |
| log_file_name = "testrun.{0}{1}.log".format(arch, flavor) |
| log_file_directory = os.path.join(repo_root, "test", "logs") |
| |
| if not os.path.exists(log_file_directory): |
| os.mkdir(log_file_directory) |
| |
| self.log_file_path = os.path.join(log_file_directory, log_file_name) |
| |
| if os.path.exists(self.log_file_path): |
| os.remove(self.log_file_path) |
| else: |
| self.log_file_path = log_file_path |
| |
| self.file = open(self.log_file_path, "w") |
| |
| def log(self, args): |
| self.file.write(args) |
| |
| def __del__(self): |
| if not (self.file is None): |
| self.file.close() |
| |
| if __name__ == '__main__': |
| log_file = LogFile(args.logfile) |
| |
| def log_message(msg = ""): |
| log_file.log(msg + "\n") |
| |
| def print_and_log(msg = ""): |
| print(msg) |
| log_message(msg) |
| |
| # remove carriage returns at end of line to avoid platform difference |
| def normalize_new_line(text): |
| return re.sub(b'[\r]+\n', b'\n', text) |
| |
| # A test simply contains a collection of test attributes. |
| # Misc attributes added by test run: |
| # filename full path of test file |
| # elapsed_time elapsed time when running the test |
| # |
| class Test(dict): |
| __setattr__ = dict.__setitem__ |
| __delattr__ = dict.__delitem__ |
| |
| # support dot syntax for normal attribute access |
| def __getattr__(self, key): |
| return super(Test, self).__getattr__(key) if key.startswith('__') \ |
| else self.get(key) |
| |
| # mark start of this test run, to compute elapsed_time |
| def start(self): |
| self.start_time = datetime.now() |
| |
| # mark end of this test run, compute elapsed_time |
| def done(self): |
| if not self.elapsed_time: |
| self.elapsed_time = (datetime.now() - self.start_time)\ |
| .total_seconds() |
| |
| # records pass_count/fail_count |
| class PassFailCount(object): |
| def __init__(self): |
| self.pass_count = 0 |
| self.fail_count = 0 |
| |
| def __str__(self): |
| return 'passed {}, failed {}'.format(self.pass_count, self.fail_count) |
| |
| def total_count(self): |
| return self.pass_count + self.fail_count |
| |
| # records total and individual folder's pass_count/fail_count |
| class TestResult(PassFailCount): |
| def __init__(self): |
| super(self.__class__, self).__init__() |
| self.folders = {} |
| |
| def _get_folder_result(self, folder): |
| r = self.folders.get(folder) |
| if not r: |
| r = PassFailCount() |
| self.folders[folder] = r |
| return r |
| |
| def log(self, filename, fail=False): |
| folder = os.path.basename(os.path.dirname(filename)) |
| r = self._get_folder_result(folder) |
| if fail: |
| r.fail_count += 1 |
| self.fail_count += 1 |
| else: |
| r.pass_count += 1 |
| self.pass_count += 1 |
| |
| # test variants: |
| # interpreted: -maxInterpretCount:1 -maxSimpleJitRunCount:1 -bgjit- |
| # dynapogo: -forceNative -off:simpleJit -bgJitDelay:0 |
| class TestVariant(object): |
| def __init__(self, name, compile_flags=[]): |
| self.name = name |
| self.compile_flags = \ |
| ['-WERExceptionSupport', '-ExtendedErrorStackForTestHost', |
| '-BaselineMode'] + compile_flags |
| self.tags = tags.copy() |
| self.not_tags = not_tags.union( |
| ['{}_{}'.format(x, name) for x in ('fails','exclude')]) |
| |
| self.msg_queue = Manager().Queue() # messages from multi processes |
| self.test_result = TestResult() |
| self._print_lines = [] # _print lines buffer |
| self._last_len = 0 |
| |
| # check if this test variant should run a given test |
| def _should_test(self, test): |
| tags = split_tags(test.get('tags')) |
| if not tags.isdisjoint(self.not_tags): |
| return False |
| if self.tags and not self.tags.issubset(tags): |
| return False |
| if not_compile_flags: # exclude unsupported compile-flags if any |
| flags = test.get('compile-flags') |
| if flags and \ |
| not not_compile_flags.isdisjoint(flags.lower().split()): |
| return False |
| return True |
| |
| # print output from multi-process run, to be sent with result message |
| def _print(self, line): |
| self._print_lines.append(str(line)) |
| |
| # queue a test result from multi-process runs |
| def _log_result(self, test, fail): |
| output = '\n'.join(self._print_lines) # collect buffered _print output |
| self._print_lines = [] |
| self.msg_queue.put((test.filename, fail, test.elapsed_time, output)) |
| |
| # (on main process) process one queued message |
| def _process_msg(self, msg): |
| filename, fail, elapsed_time, output = msg |
| self.test_result.log(filename, fail=fail) |
| line = '[{}/{} {:4.2f}] {} -> {}'.format( |
| self.test_result.total_count(), |
| self.test_count, |
| elapsed_time, |
| 'Failed' if fail else 'Passed', |
| self._short_name(filename)) |
| padding = self._last_len - len(line) |
| print(line + ' ' * padding, end='\n' if fail else '\r') |
| log_message(line) |
| self._last_len = len(line) if not fail else 0 |
| if len(output) > 0: |
| print_and_log(output) |
| |
| # get a shorter test file path for display only |
| def _short_name(self, filename): |
| folder = os.path.basename(os.path.dirname(filename)) |
| return os.path.join(folder, os.path.basename(filename)) |
| |
| # (on main process) wait and process one queued message |
| def _process_one_msg(self): |
| self._process_msg(self.msg_queue.get()) |
| |
| # log a failed test with details |
| def _show_failed(self, test, flags, exit_code, output, |
| expected_output=None, timedout=False): |
| if timedout: |
| self._print('ERROR: Test timed out!') |
| self._print('{} {} {}'.format(binary, ' '.join(flags), test.filename)) |
| if expected_output == None or timedout: |
| self._print("\nOutput:") |
| self._print("----------------------------") |
| self._print(output) |
| self._print("----------------------------") |
| else: |
| lst_output = output.split(b'\n') |
| lst_expected = expected_output.split(b'\n') |
| ln = min(len(lst_output), len(lst_expected)) |
| for i in range(0, ln): |
| if lst_output[i] != lst_expected[i]: |
| self._print("Output: (at line " + str(i) + ")") |
| self._print("----------------------------") |
| self._print(lst_output[i]) |
| self._print("----------------------------") |
| self._print("Expected Output:") |
| self._print("----------------------------") |
| self._print(lst_expected[i]) |
| self._print("----------------------------") |
| break |
| |
| self._print("exit code: {}".format(exit_code)) |
| self._log_result(test, fail=True) |
| |
| # temp: try find real file name on hard drive if case mismatch |
| def _check_file(self, folder, filename): |
| path = os.path.join(folder, filename) |
| if os.path.isfile(path): |
| return path # file exists on disk |
| |
| filename_lower = filename.lower() |
| files = os.listdir(folder) |
| for i in range(len(files)): |
| if files[i].lower() == filename_lower: |
| self._print('\nWARNING: {} should be {}\n'.format( |
| path, files[i])) |
| return os.path.join(folder, files[i]) |
| |
| # cann't find the file, just return the path and let it error out |
| return path |
| |
| # run one test under this variant |
| def test_one(self, test): |
| try: |
| test.start() |
| self._run_one_test(test) |
| except Exception: |
| test.done() |
| self._print(traceback.format_exc()) |
| self._log_result(test, fail=True) |
| |
| # internally perform one test run |
| def _run_one_test(self, test): |
| folder = test.folder |
| js_file = test.filename = self._check_file(folder, test.files) |
| js_output = b'' |
| |
| working_path = os.path.dirname(js_file) |
| |
| flags = test.get('compile-flags') |
| flags = self.compile_flags + (flags.split() if flags else []) |
| cmd = [binary] + flags + [os.path.basename(js_file)] |
| |
| test.start() |
| proc = SP.Popen(cmd, stdout=SP.PIPE, stderr=SP.STDOUT, cwd=working_path) |
| timeout_data = [proc, False] |
| def timeout_func(timeout_data): |
| timeout_data[0].kill() |
| timeout_data[1] = True |
| timeout = test.get('timeout', args.timeout) # test override or default |
| timer = Timer(timeout, timeout_func, [timeout_data]) |
| try: |
| timer.start() |
| js_output = normalize_new_line(proc.communicate()[0]) |
| exit_code = proc.wait() |
| finally: |
| timer.cancel() |
| test.done() |
| |
| # shared _show_failed args |
| fail_args = { 'test': test, 'flags': flags, |
| 'exit_code': exit_code, 'output': js_output }; |
| |
| # check timed out |
| if (timeout_data[1]): |
| return self._show_failed(timedout=True, **fail_args) |
| |
| # check ch failed |
| if exit_code != 0: |
| return self._show_failed(**fail_args) |
| |
| # check output |
| if 'baseline' not in test: |
| # output lines must be 'pass' or 'passed' or empty |
| lines = (line.lower() for line in js_output.split(b'\n')) |
| if any(line != b'' and line != b'pass' and line != b'passed' |
| for line in lines): |
| return self._show_failed(**fail_args) |
| else: |
| baseline = test.get('baseline') |
| if baseline: |
| # perform baseline comparison |
| baseline = self._check_file(working_path, baseline) |
| with open(baseline, 'rb') as bs_file: |
| baseline_output = bs_file.read() |
| |
| # Cleanup carriage return |
| # todo: remove carriage return at the end of the line |
| # or better fix ch to output same on all platforms |
| expected_output = normalize_new_line(baseline_output) |
| |
| if expected_output != js_output: |
| return self._show_failed( |
| expected_output=expected_output, **fail_args) |
| |
| # passed |
| self._log_result(test, fail=False) |
| |
| # run tests under this variant, using given multiprocessing Pool |
| def run(self, tests, pool): |
| print_and_log('\n############# Starting {} variant #############'\ |
| .format(self.name)) |
| if self.tags: |
| print_and_log(' tags: {}'.format(self.tags)) |
| for x in self.not_tags: |
| print_and_log(' exclude: {}'.format(x)) |
| print_and_log() |
| |
| # filter tests to run |
| tests = [x for x in tests if self._should_test(x)] |
| self.test_count = len(tests) |
| |
| # run tests in parallel |
| result = pool.map_async(run_one, [(self,test) for test in tests]) |
| while self.test_result.total_count() != self.test_count: |
| self._process_one_msg() |
| |
| # print test result summary |
| def print_summary(self): |
| print_and_log('\n######## Logs for {} variant ########'\ |
| .format(self.name)) |
| for folder, result in sorted(self.test_result.folders.items()): |
| print_and_log('{}: {}'.format(folder, result)) |
| print_and_log("----------------------------") |
| print_and_log('Total: {}'.format(self.test_result)) |
| |
| # global run one test function for multiprocessing, used by TestVariant |
| def run_one(data): |
| try: |
| variant, test = data |
| variant.test_one(test) |
| except Exception: |
| print('ERROR: Unhandled exception!!!') |
| traceback.print_exc() |
| |
| # record folder/tags info from test_root/rlexedirs.xml |
| class FolderTags(object): |
| def __init__(self): |
| xmlpath = os.path.join(test_root, 'rlexedirs.xml') |
| try: |
| xml = ET.parse(xmlpath).getroot() |
| except IOError: |
| print_and_log('ERROR: failed to read {}'.format(xmlpath)) |
| exit(-1) |
| |
| self._folder_tags = {} |
| for x in xml: |
| d = x.find('default') |
| key = d.find('files').text.lower() # avoid case mismatch |
| tags = d.find('tags') |
| self._folder_tags[key] = \ |
| split_tags(tags.text) if tags != None else _empty_set |
| |
| # check if should test a given folder |
| def should_test(self, folder): |
| key = os.path.basename(os.path.normpath(folder)).lower() |
| ftags = self._folder_tags.get(key) |
| |
| # folder listed in rlexedirs.xml and not exlucded by global not_tags |
| return ftags != None and ftags.isdisjoint(not_tags) |
| |
| |
| # load all tests in folder using rlexe.xml file |
| def load_tests(folder, file): |
| try: |
| xmlpath = os.path.join(folder, 'rlexe.xml') |
| xml = ET.parse(xmlpath).getroot() |
| except IOError: |
| return [] |
| |
| def test_override(condition, check_tag, check_value, test): |
| target = condition.find(check_tag) |
| if target != None and target.text == check_value: |
| for override in condition.find('override'): |
| test[override.tag] = override.text |
| |
| def load_test(testXml): |
| test = Test(folder=folder) |
| for c in testXml.find('default'): |
| if c.tag == 'timeout': # timeout seconds |
| test[c.tag] = int(c.text) |
| elif c.tag == 'tags' and c.tag in test: # merge multiple <tags> |
| test[c.tag] = test[c.tag] + ',' + c.text |
| else: |
| test[c.tag] = c.text |
| |
| condition = testXml.find('condition') |
| if condition != None: |
| test_override(condition, 'target', arch_alias, test) |
| |
| return test |
| |
| tests = [load_test(x) for x in xml] |
| if file != None: |
| tests = [x for x in tests if x.files == file] |
| if len(tests) == 0 and is_jsfile(file): |
| tests = [Test(folder=folder, files=file, baseline='')] |
| return tests |
| |
| def is_jsfile(path): |
| return os.path.splitext(path)[1] == '.js' |
| |
| def main(): |
| # By default run all tests |
| if len(args.folders) == 0: |
| files = (os.path.join(test_root, x) for x in os.listdir(test_root)) |
| args.folders = [f for f in sorted(files) if not os.path.isfile(f)] |
| |
| # Set the right timezone, the tests need Pacific Standard Time |
| os.environ['TZ'] = 'US/Pacific' |
| time.tzset() |
| |
| # load all tests |
| tests = [] |
| folder_tags = FolderTags() |
| for path in args.folders: |
| if os.path.isfile(path): |
| folder, file = os.path.dirname(path), os.path.basename(path) |
| else: |
| folder, file = path, None |
| if folder_tags.should_test(folder): |
| tests += load_tests(folder, file) |
| |
| # test variants |
| variants = [ |
| TestVariant('interpreted', [ |
| '-maxInterpretCount:1', '-maxSimpleJitRunCount:1', '-bgjit-']) |
| ] |
| |
| # run each variant |
| pool = Pool(1) # Use a multiprocessing process Pool -- TODO: forcing this to single threaded until we have support for sequential tag |
| start_time = datetime.now() |
| for variant in variants: |
| variant.run(tests, pool) |
| elapsed_time = datetime.now() - start_time |
| |
| # print summary |
| for variant in variants: |
| variant.print_summary() |
| |
| print() |
| failed = any(variant.test_result.fail_count > 0 for variant in variants) |
| print('[{}] {}'.format( |
| str(elapsed_time), 'Success!' if not failed else 'Failed!')) |
| |
| return 1 if failed else 0 |
| |
| if __name__ == '__main__': |
| sys.exit(main()) |
