| #!/usr/bin/env python |
| # Copyright 2016 The Chromium Authors. All rights reserved. |
| # Use of this source code is governed by a BSD-style license that can be |
| # found in the LICENSE file. |
| |
| """Script to generate the majority of the JSON files in the src/testing/buildbot |
| directory. Maintaining these files by hand is too unwieldy. |
| """ |
| |
| import argparse |
| import ast |
| import collections |
| import copy |
| import difflib |
| import itertools |
| import json |
| import os |
| import string |
| import sys |
| import traceback |
| |
| THIS_DIR = os.path.dirname(os.path.abspath(__file__)) |
| |
| |
| class BBGenErr(Exception): |
| def __init__(self, message): |
| super(BBGenErr, self).__init__(message) |
| |
| |
| # This class is only present to accommodate certain machines on |
| # chromium.android.fyi which run certain tests as instrumentation |
| # tests, but not as gtests. If this discrepancy were fixed then the |
| # notion could be removed. |
| class TestSuiteTypes(object): |
| GTEST = 'gtest' |
| |
| |
| class BaseGenerator(object): |
| def __init__(self, bb_gen): |
| self.bb_gen = bb_gen |
| |
| def generate(self, waterfall, tester_name, tester_config, input_tests): |
| raise NotImplementedError() |
| |
| def sort(self, tests): |
| raise NotImplementedError() |
| |
| |
| def cmp_tests(a, b): |
| # Prefer to compare based on the "test" key. |
| val = cmp(a['test'], b['test']) |
| if val != 0: |
| return val |
| if 'name' in a and 'name' in b: |
| return cmp(a['name'], b['name']) # pragma: no cover |
| if 'name' not in a and 'name' not in b: |
| return 0 # pragma: no cover |
| # Prefer to put variants of the same test after the first one. |
| if 'name' in a: |
| return 1 |
| # 'name' is in b. |
| return -1 # pragma: no cover |
| |
| |
| class GPUTelemetryTestGenerator(BaseGenerator): |
| def __init__(self, bb_gen): |
| super(GPUTelemetryTestGenerator, self).__init__(bb_gen) |
| |
| def generate(self, waterfall, tester_name, tester_config, input_tests): |
| isolated_scripts = [] |
| for test_name, test_config in sorted(input_tests.iteritems()): |
| test = self.bb_gen.generate_gpu_telemetry_test( |
| waterfall, tester_name, tester_config, test_name, test_config) |
| if test: |
| isolated_scripts.append(test) |
| return isolated_scripts |
| |
| def sort(self, tests): |
| return sorted(tests, key=lambda x: x['name']) |
| |
| |
| class GTestGenerator(BaseGenerator): |
| def __init__(self, bb_gen): |
| super(GTestGenerator, self).__init__(bb_gen) |
| |
| def generate(self, waterfall, tester_name, tester_config, input_tests): |
| # The relative ordering of some of the tests is important to |
| # minimize differences compared to the handwritten JSON files, since |
| # Python's sorts are stable and there are some tests with the same |
| # key (see gles2_conform_d3d9_test and similar variants). Avoid |
| # losing the order by avoiding coalescing the dictionaries into one. |
| gtests = [] |
| for test_name, test_config in sorted(input_tests.iteritems()): |
| test = self.bb_gen.generate_gtest( |
| waterfall, tester_name, tester_config, test_name, test_config) |
| if test: |
| # generate_gtest may veto the test generation on this tester. |
| gtests.append(test) |
| return gtests |
| |
| def sort(self, tests): |
| return sorted(tests, cmp=cmp_tests) |
| |
| |
| class IsolatedScriptTestGenerator(BaseGenerator): |
| def __init__(self, bb_gen): |
| super(IsolatedScriptTestGenerator, self).__init__(bb_gen) |
| |
| def generate(self, waterfall, tester_name, tester_config, input_tests): |
| isolated_scripts = [] |
| for test_name, test_config in sorted(input_tests.iteritems()): |
| test = self.bb_gen.generate_isolated_script_test( |
| waterfall, tester_name, tester_config, test_name, test_config) |
| if test: |
| isolated_scripts.append(test) |
| return isolated_scripts |
| |
| def sort(self, tests): |
| return sorted(tests, key=lambda x: x['name']) |
| |
| |
| class ScriptGenerator(BaseGenerator): |
| def __init__(self, bb_gen): |
| super(ScriptGenerator, self).__init__(bb_gen) |
| |
| def generate(self, waterfall, tester_name, tester_config, input_tests): |
| scripts = [] |
| for test_name, test_config in sorted(input_tests.iteritems()): |
| test = self.bb_gen.generate_script_test( |
| waterfall, tester_name, tester_config, test_name, test_config) |
| if test: |
| scripts.append(test) |
| return scripts |
| |
| def sort(self, tests): |
| return sorted(tests, key=lambda x: x['name']) |
| |
| |
| class JUnitGenerator(BaseGenerator): |
| def __init__(self, bb_gen): |
| super(JUnitGenerator, self).__init__(bb_gen) |
| |
| def generate(self, waterfall, tester_name, tester_config, input_tests): |
| scripts = [] |
| for test_name, test_config in sorted(input_tests.iteritems()): |
| test = self.bb_gen.generate_junit_test( |
| waterfall, tester_name, tester_config, test_name, test_config) |
| if test: |
| scripts.append(test) |
| return scripts |
| |
| def sort(self, tests): |
| return sorted(tests, key=lambda x: x['test']) |
| |
| |
| class CTSGenerator(BaseGenerator): |
| def __init__(self, bb_gen): |
| super(CTSGenerator, self).__init__(bb_gen) |
| |
| def generate(self, waterfall, tester_name, tester_config, input_tests): |
| # These only contain one entry and it's the contents of the input tests' |
| # dictionary, verbatim. |
| cts_tests = [] |
| cts_tests.append(input_tests) |
| return cts_tests |
| |
| def sort(self, tests): |
| return tests |
| |
| |
| class InstrumentationTestGenerator(BaseGenerator): |
| def __init__(self, bb_gen): |
| super(InstrumentationTestGenerator, self).__init__(bb_gen) |
| |
| def generate(self, waterfall, tester_name, tester_config, input_tests): |
| scripts = [] |
| for test_name, test_config in sorted(input_tests.iteritems()): |
| test = self.bb_gen.generate_instrumentation_test( |
| waterfall, tester_name, tester_config, test_name, test_config) |
| if test: |
| scripts.append(test) |
| return scripts |
| |
| def sort(self, tests): |
| return sorted(tests, cmp=cmp_tests) |
| |
| |
| class BBJSONGenerator(object): |
| def __init__(self): |
| self.this_dir = THIS_DIR |
| self.args = None |
| self.waterfalls = None |
| self.test_suites = None |
| self.exceptions = None |
| self.mixins = None |
| |
| def generate_abs_file_path(self, relative_path): |
| return os.path.join(self.this_dir, relative_path) # pragma: no cover |
| |
| def print_line(self, line): |
| # Exists so that tests can mock |
| print line # pragma: no cover |
| |
| def read_file(self, relative_path): |
| with open(self.generate_abs_file_path( |
| relative_path)) as fp: # pragma: no cover |
| return fp.read() # pragma: no cover |
| |
| def write_file(self, relative_path, contents): |
| with open(self.generate_abs_file_path( |
| relative_path), 'wb') as fp: # pragma: no cover |
| fp.write(contents) # pragma: no cover |
| |
| def pyl_file_path(self, filename): |
| if self.args and self.args.pyl_files_dir: |
| return os.path.join(self.args.pyl_files_dir, filename) |
| return filename |
| |
| def load_pyl_file(self, filename): |
| try: |
| return ast.literal_eval(self.read_file( |
| self.pyl_file_path(filename))) |
| except (SyntaxError, ValueError) as e: # pragma: no cover |
| raise BBGenErr('Failed to parse pyl file "%s": %s' % |
| (filename, e)) # pragma: no cover |
| |
| # TOOD(kbr): require that os_type be specified for all bots in waterfalls.pyl. |
| # Currently it is only mandatory for bots which run GPU tests. Change these to |
| # use [] instead of .get(). |
| def is_android(self, tester_config): |
| return tester_config.get('os_type') == 'android' |
| |
| def is_chromeos(self, tester_config): |
| return tester_config.get('os_type') == 'chromeos' |
| |
| def is_linux(self, tester_config): |
| return tester_config.get('os_type') == 'linux' |
| |
| def get_exception_for_test(self, test_name, test_config): |
| # gtests may have both "test" and "name" fields, and usually, if the "name" |
| # field is specified, it means that the same test is being repurposed |
| # multiple times with different command line arguments. To handle this case, |
| # prefer to lookup per the "name" field of the test itself, as opposed to |
| # the "test_name", which is actually the "test" field. |
| if 'name' in test_config: |
| return self.exceptions.get(test_config['name']) |
| else: |
| return self.exceptions.get(test_name) |
| |
| def should_run_on_tester(self, waterfall, tester_name,test_name, test_config): |
| # Currently, the only reason a test should not run on a given tester is that |
| # it's in the exceptions. (Once the GPU waterfall generation script is |
| # incorporated here, the rules will become more complex.) |
| exception = self.get_exception_for_test(test_name, test_config) |
| if not exception: |
| return True |
| remove_from = None |
| remove_from = exception.get('remove_from') |
| if remove_from: |
| if tester_name in remove_from: |
| return False |
| # TODO(kbr): this code path was added for some tests (including |
| # android_webview_unittests) on one machine (Nougat Phone |
| # Tester) which exists with the same name on two waterfalls, |
| # chromium.android and chromium.fyi; the tests are run on one |
| # but not the other. Once the bots are all uniquely named (a |
| # different ongoing project) this code should be removed. |
| # TODO(kbr): add coverage. |
| return (tester_name + ' ' + waterfall['name'] |
| not in remove_from) # pragma: no cover |
| return True |
| |
| def get_test_modifications(self, test, test_name, tester_name): |
| exception = self.get_exception_for_test(test_name, test) |
| if not exception: |
| return None |
| return exception.get('modifications', {}).get(tester_name) |
| |
| def merge_command_line_args(self, arr, prefix, splitter): |
| prefix_len = len(prefix) |
| idx = 0 |
| first_idx = -1 |
| accumulated_args = [] |
| while idx < len(arr): |
| flag = arr[idx] |
| delete_current_entry = False |
| if flag.startswith(prefix): |
| arg = flag[prefix_len:] |
| accumulated_args.extend(arg.split(splitter)) |
| if first_idx < 0: |
| first_idx = idx |
| else: |
| delete_current_entry = True |
| if delete_current_entry: |
| del arr[idx] |
| else: |
| idx += 1 |
| if first_idx >= 0: |
| arr[first_idx] = prefix + splitter.join(accumulated_args) |
| return arr |
| |
| def maybe_fixup_args_array(self, arr): |
| # The incoming array of strings may be an array of command line |
| # arguments. To make it easier to turn on certain features per-bot or |
| # per-test-suite, look specifically for certain flags and merge them |
| # appropriately. |
| # --enable-features=Feature1 --enable-features=Feature2 |
| # are merged to: |
| # --enable-features=Feature1,Feature2 |
| # and: |
| # --extra-browser-args=arg1 --extra-browser-args=arg2 |
| # are merged to: |
| # --extra-browser-args=arg1 arg2 |
| arr = self.merge_command_line_args(arr, '--enable-features=', ',') |
| arr = self.merge_command_line_args(arr, '--extra-browser-args=', ' ') |
| return arr |
| |
| def dictionary_merge(self, a, b, path=None, update=True): |
| """http://stackoverflow.com/questions/7204805/ |
| python-dictionaries-of-dictionaries-merge |
| merges b into a |
| """ |
| if path is None: |
| path = [] |
| for key in b: |
| if key in a: |
| if isinstance(a[key], dict) and isinstance(b[key], dict): |
| self.dictionary_merge(a[key], b[key], path + [str(key)]) |
| elif a[key] == b[key]: |
| pass # same leaf value |
| elif isinstance(a[key], list) and isinstance(b[key], list): |
| # Args arrays are lists of strings. Just concatenate them, |
| # and don't sort them, in order to keep some needed |
| # arguments adjacent (like --time-out-ms [arg], etc.) |
| if all(isinstance(x, str) |
| for x in itertools.chain(a[key], b[key])): |
| a[key] = self.maybe_fixup_args_array(a[key] + b[key]) |
| else: |
| # TODO(kbr): this only works properly if the two arrays are |
| # the same length, which is currently always the case in the |
| # swarming dimension_sets that we have to merge. It will fail |
| # to merge / override 'args' arrays which are different |
| # length. |
| for idx in xrange(len(b[key])): |
| try: |
| a[key][idx] = self.dictionary_merge(a[key][idx], b[key][idx], |
| path + [str(key), str(idx)], |
| update=update) |
| except (IndexError, TypeError): # pragma: no cover |
| raise BBGenErr('Error merging list keys ' + str(key) + |
| ' and indices ' + str(idx) + ' between ' + |
| str(a) + ' and ' + str(b)) # pragma: no cover |
| elif update: # pragma: no cover |
| a[key] = b[key] # pragma: no cover |
| else: |
| raise BBGenErr('Conflict at %s' % '.'.join( |
| path + [str(key)])) # pragma: no cover |
| else: |
| a[key] = b[key] |
| return a |
| |
| def initialize_args_for_test( |
| self, generated_test, tester_config, additional_arg_keys=None): |
| |
| args = [] |
| args.extend(generated_test.get('args', [])) |
| args.extend(tester_config.get('args', [])) |
| |
| def add_conditional_args(key, fn): |
| val = generated_test.pop(key, []) |
| if fn(tester_config): |
| args.extend(val) |
| |
| add_conditional_args('desktop_args', lambda cfg: not self.is_android(cfg)) |
| add_conditional_args('linux_args', self.is_linux) |
| add_conditional_args('android_args', self.is_android) |
| |
| for key in additional_arg_keys or []: |
| args.extend(generated_test.pop(key, [])) |
| args.extend(tester_config.get(key, [])) |
| |
| if args: |
| generated_test['args'] = self.maybe_fixup_args_array(args) |
| |
| def initialize_swarming_dictionary_for_test(self, generated_test, |
| tester_config): |
| if 'swarming' not in generated_test: |
| generated_test['swarming'] = {} |
| if not 'can_use_on_swarming_builders' in generated_test['swarming']: |
| generated_test['swarming'].update({ |
| 'can_use_on_swarming_builders': tester_config.get('use_swarming', True) |
| }) |
| if 'swarming' in tester_config: |
| if ('dimension_sets' not in generated_test['swarming'] and |
| 'dimension_sets' in tester_config['swarming']): |
| generated_test['swarming']['dimension_sets'] = copy.deepcopy( |
| tester_config['swarming']['dimension_sets']) |
| self.dictionary_merge(generated_test['swarming'], |
| tester_config['swarming']) |
| # Apply any Android-specific Swarming dimensions after the generic ones. |
| if 'android_swarming' in generated_test: |
| if self.is_android(tester_config): # pragma: no cover |
| self.dictionary_merge( |
| generated_test['swarming'], |
| generated_test['android_swarming']) # pragma: no cover |
| del generated_test['android_swarming'] # pragma: no cover |
| |
| def clean_swarming_dictionary(self, swarming_dict): |
| # Clean out redundant entries from a test's "swarming" dictionary. |
| # This is really only needed to retain 100% parity with the |
| # handwritten JSON files, and can be removed once all the files are |
| # autogenerated. |
| if 'shards' in swarming_dict: |
| if swarming_dict['shards'] == 1: # pragma: no cover |
| del swarming_dict['shards'] # pragma: no cover |
| if 'hard_timeout' in swarming_dict: |
| if swarming_dict['hard_timeout'] == 0: # pragma: no cover |
| del swarming_dict['hard_timeout'] # pragma: no cover |
| if not swarming_dict.get('can_use_on_swarming_builders', False): |
| # Remove all other keys. |
| for k in swarming_dict.keys(): # pragma: no cover |
| if k != 'can_use_on_swarming_builders': # pragma: no cover |
| del swarming_dict[k] # pragma: no cover |
| |
| def update_and_cleanup_test(self, test, test_name, tester_name, tester_config, |
| waterfall): |
| # Apply swarming mixins. |
| test = self.apply_all_mixins( |
| test, waterfall, tester_name, tester_config) |
| # See if there are any exceptions that need to be merged into this |
| # test's specification. |
| modifications = self.get_test_modifications(test, test_name, tester_name) |
| if modifications: |
| test = self.dictionary_merge(test, modifications) |
| if 'swarming' in test: |
| self.clean_swarming_dictionary(test['swarming']) |
| return test |
| |
| def add_common_test_properties(self, test, tester_config): |
| if tester_config.get('use_multi_dimension_trigger_script'): |
| # Assumes update_and_cleanup_test has already been called, so the |
| # builder's mixins have been flattened into the test. |
| test['trigger_script'] = { |
| 'script': '//testing/trigger_scripts/trigger_multiple_dimensions.py', |
| 'args': [ |
| '--multiple-trigger-configs', |
| json.dumps(test['swarming']['dimension_sets'] + |
| tester_config.get('alternate_swarming_dimensions', [])), |
| '--multiple-dimension-script-verbose', |
| 'True' |
| ], |
| } |
| elif self.is_chromeos(tester_config) and tester_config.get('use_swarming', |
| True): |
| # The presence of the "device_type" dimension indicates that the tests |
| # are targetting CrOS hardware and so need the special trigger script. |
| dimension_sets = tester_config['swarming']['dimension_sets'] |
| if all('device_type' in ds for ds in dimension_sets): |
| test['trigger_script'] = { |
| 'script': '//testing/trigger_scripts/chromeos_device_trigger.py', |
| } |
| |
| def add_android_presentation_args(self, tester_config, test_name, result): |
| args = result.get('args', []) |
| args.append('--gs-results-bucket=chromium-result-details') |
| if (result['swarming']['can_use_on_swarming_builders'] and not |
| tester_config.get('skip_merge_script', False)): |
| result['merge'] = { |
| 'args': [ |
| '--bucket', |
| 'chromium-result-details', |
| '--test-name', |
| test_name |
| ], |
| 'script': '//build/android/pylib/results/presentation/' |
| 'test_results_presentation.py', |
| } |
| if not tester_config.get('skip_cipd_packages', False): |
| cipd_packages = result['swarming'].get('cipd_packages', []) |
| cipd_packages.append( |
| { |
| 'cipd_package': 'infra/tools/luci/logdog/butler/${platform}', |
| 'location': 'bin', |
| 'revision': 'git_revision:ff387eadf445b24c935f1cf7d6ddd279f8a6b04c', |
| } |
| ) |
| result['swarming']['cipd_packages'] = cipd_packages |
| if not tester_config.get('skip_output_links', False): |
| result['swarming']['output_links'] = [ |
| { |
| 'link': [ |
| 'https://luci-logdog.appspot.com/v/?s', |
| '=android%2Fswarming%2Flogcats%2F', |
| '${TASK_ID}%2F%2B%2Funified_logcats', |
| ], |
| 'name': 'shard #${SHARD_INDEX} logcats', |
| }, |
| ] |
| if args: |
| result['args'] = args |
| |
| def generate_gtest(self, waterfall, tester_name, tester_config, test_name, |
| test_config): |
| if not self.should_run_on_tester( |
| waterfall, tester_name, test_name, test_config): |
| return None |
| result = copy.deepcopy(test_config) |
| if 'test' in result: |
| result['name'] = test_name |
| else: |
| result['test'] = test_name |
| self.initialize_swarming_dictionary_for_test(result, tester_config) |
| |
| self.initialize_args_for_test( |
| result, tester_config, additional_arg_keys=['gtest_args']) |
| if self.is_android(tester_config) and tester_config.get('use_swarming', |
| True): |
| self.add_android_presentation_args(tester_config, test_name, result) |
| result['args'] = result.get('args', []) + ['--recover-devices'] |
| |
| result = self.update_and_cleanup_test( |
| result, test_name, tester_name, tester_config, waterfall) |
| self.add_common_test_properties(result, tester_config) |
| return result |
| |
| def generate_isolated_script_test(self, waterfall, tester_name, tester_config, |
| test_name, test_config): |
| if not self.should_run_on_tester(waterfall, tester_name, test_name, |
| test_config): |
| return None |
| result = copy.deepcopy(test_config) |
| result['isolate_name'] = result.get('isolate_name', test_name) |
| result['name'] = test_name |
| self.initialize_swarming_dictionary_for_test(result, tester_config) |
| self.initialize_args_for_test(result, tester_config) |
| if tester_config.get('use_android_presentation', False): |
| self.add_android_presentation_args(tester_config, test_name, result) |
| result = self.update_and_cleanup_test( |
| result, test_name, tester_name, tester_config, waterfall) |
| self.add_common_test_properties(result, tester_config) |
| return result |
| |
| def generate_script_test(self, waterfall, tester_name, tester_config, |
| test_name, test_config): |
| if not self.should_run_on_tester(waterfall, tester_name, test_name, |
| test_config): |
| return None |
| result = { |
| 'name': test_name, |
| 'script': test_config['script'] |
| } |
| result = self.update_and_cleanup_test( |
| result, test_name, tester_name, tester_config, waterfall) |
| return result |
| |
| def generate_junit_test(self, waterfall, tester_name, tester_config, |
| test_name, test_config): |
| del tester_config |
| if not self.should_run_on_tester(waterfall, tester_name, test_name, |
| test_config): |
| return None |
| result = { |
| 'test': test_name, |
| } |
| return result |
| |
| def generate_instrumentation_test(self, waterfall, tester_name, tester_config, |
| test_name, test_config): |
| if not self.should_run_on_tester(waterfall, tester_name, test_name, |
| test_config): |
| return None |
| result = copy.deepcopy(test_config) |
| if 'test' in result and result['test'] != test_name: |
| result['name'] = test_name |
| else: |
| result['test'] = test_name |
| result = self.update_and_cleanup_test( |
| result, test_name, tester_name, tester_config, waterfall) |
| return result |
| |
| def substitute_gpu_args(self, tester_config, swarming_config, args): |
| substitutions = { |
| # Any machine in waterfalls.pyl which desires to run GPU tests |
| # must provide the os_type key. |
| 'os_type': tester_config['os_type'], |
| 'gpu_vendor_id': '0', |
| 'gpu_device_id': '0', |
| } |
| dimension_set = swarming_config['dimension_sets'][0] |
| if 'gpu' in dimension_set: |
| # First remove the driver version, then split into vendor and device. |
| gpu = dimension_set['gpu'] |
| gpu = gpu.split('-')[0].split(':') |
| substitutions['gpu_vendor_id'] = gpu[0] |
| substitutions['gpu_device_id'] = gpu[1] |
| return [string.Template(arg).safe_substitute(substitutions) for arg in args] |
| |
| def generate_gpu_telemetry_test(self, waterfall, tester_name, tester_config, |
| test_name, test_config): |
| # These are all just specializations of isolated script tests with |
| # a bunch of boilerplate command line arguments added. |
| |
| # The step name must end in 'test' or 'tests' in order for the |
| # results to automatically show up on the flakiness dashboard. |
| # (At least, this was true some time ago.) Continue to use this |
| # naming convention for the time being to minimize changes. |
| step_name = test_config.get('name', test_name) |
| if not (step_name.endswith('test') or step_name.endswith('tests')): |
| step_name = '%s_tests' % step_name |
| result = self.generate_isolated_script_test( |
| waterfall, tester_name, tester_config, step_name, test_config) |
| if not result: |
| return None |
| result['isolate_name'] = 'telemetry_gpu_integration_test' |
| args = result.get('args', []) |
| test_to_run = result.pop('telemetry_test_name', test_name) |
| |
| # These tests upload and download results from cloud storage and therefore |
| # aren't idempotent yet. https://crbug.com/549140. |
| result['swarming']['idempotent'] = False |
| |
| # The GPU tests act much like integration tests for the entire browser, and |
| # tend to uncover flakiness bugs more readily than other test suites. In |
| # order to surface any flakiness more readily to the developer of the CL |
| # which is introducing it, we disable retries with patch on the commit |
| # queue. |
| result['should_retry_with_patch'] = False |
| |
| args = [ |
| test_to_run, |
| '--show-stdout', |
| '--browser=%s' % tester_config['browser_config'], |
| # --passthrough displays more of the logging in Telemetry when |
| # run via typ, in particular some of the warnings about tests |
| # being expected to fail, but passing. |
| '--passthrough', |
| '-v', |
| '--extra-browser-args=--enable-logging=stderr --js-flags=--expose-gc', |
| ] + args |
| result['args'] = self.maybe_fixup_args_array(self.substitute_gpu_args( |
| tester_config, result['swarming'], args)) |
| return result |
| |
| def get_test_generator_map(self): |
| return { |
| 'cts_tests': CTSGenerator(self), |
| 'gpu_telemetry_tests': GPUTelemetryTestGenerator(self), |
| 'gtest_tests': GTestGenerator(self), |
| 'instrumentation_tests': InstrumentationTestGenerator(self), |
| 'isolated_scripts': IsolatedScriptTestGenerator(self), |
| 'junit_tests': JUnitGenerator(self), |
| 'scripts': ScriptGenerator(self), |
| } |
| |
| def get_test_type_remapper(self): |
| return { |
| # These are a specialization of isolated_scripts with a bunch of |
| # boilerplate command line arguments added to each one. |
| 'gpu_telemetry_tests': 'isolated_scripts', |
| } |
| |
| def check_composition_test_suites(self): |
| # Pre-pass to catch errors reliably. |
| for name, value in self.test_suites.iteritems(): |
| if isinstance(value, list): |
| for entry in value: |
| if isinstance(self.test_suites[entry], list): |
| raise BBGenErr('Composition test suites may not refer to other ' |
| 'composition test suites (error found while ' |
| 'processing %s)' % name) |
| |
| def flatten_test_suites(self): |
| new_test_suites = {} |
| for name, value in self.test_suites.get('basic_suites', {}).iteritems(): |
| new_test_suites[name] = value |
| for name, value in self.test_suites.get('compound_suites', {}).iteritems(): |
| if name in new_test_suites: |
| raise BBGenErr('Composition test suite names may not duplicate basic ' |
| 'test suite names (error found while processsing %s' % ( |
| name)) |
| new_test_suites[name] = value |
| self.test_suites = new_test_suites |
| |
| def resolve_composition_test_suites(self): |
| self.flatten_test_suites() |
| |
| self.check_composition_test_suites() |
| for name, value in self.test_suites.iteritems(): |
| if isinstance(value, list): |
| # Resolve this to a dictionary. |
| full_suite = {} |
| for entry in value: |
| suite = self.test_suites[entry] |
| full_suite.update(suite) |
| self.test_suites[name] = full_suite |
| |
| def link_waterfalls_to_test_suites(self): |
| for waterfall in self.waterfalls: |
| for tester_name, tester in waterfall['machines'].iteritems(): |
| for suite, value in tester.get('test_suites', {}).iteritems(): |
| if not value in self.test_suites: |
| # Hard / impossible to cover this in the unit test. |
| raise self.unknown_test_suite( |
| value, tester_name, waterfall['name']) # pragma: no cover |
| tester['test_suites'][suite] = self.test_suites[value] |
| |
| def load_configuration_files(self): |
| self.waterfalls = self.load_pyl_file('waterfalls.pyl') |
| self.test_suites = self.load_pyl_file('test_suites.pyl') |
| self.exceptions = self.load_pyl_file('test_suite_exceptions.pyl') |
| self.mixins = self.load_pyl_file('mixins.pyl') |
| |
| def resolve_configuration_files(self): |
| self.resolve_composition_test_suites() |
| self.link_waterfalls_to_test_suites() |
| |
| def unknown_bot(self, bot_name, waterfall_name): |
| return BBGenErr( |
| 'Unknown bot name "%s" on waterfall "%s"' % (bot_name, waterfall_name)) |
| |
| def unknown_test_suite(self, suite_name, bot_name, waterfall_name): |
| return BBGenErr( |
| 'Test suite %s from machine %s on waterfall %s not present in ' |
| 'test_suites.pyl' % (suite_name, bot_name, waterfall_name)) |
| |
| def unknown_test_suite_type(self, suite_type, bot_name, waterfall_name): |
| return BBGenErr( |
| 'Unknown test suite type ' + suite_type + ' in bot ' + bot_name + |
| ' on waterfall ' + waterfall_name) |
| |
| def apply_all_mixins(self, test, waterfall, builder_name, builder): |
| """Applies all present swarming mixins to the test for a given builder. |
| |
| Checks in the waterfall, builder, and test objects for mixins. |
| """ |
| def valid_mixin(mixin_name): |
| """Asserts that the mixin is valid.""" |
| if mixin_name not in self.mixins: |
| raise BBGenErr("bad mixin %s" % mixin_name) |
| def must_be_list(mixins, typ, name): |
| """Asserts that given mixins are a list.""" |
| if not isinstance(mixins, list): |
| raise BBGenErr("'%s' in %s '%s' must be a list" % (mixins, typ, name)) |
| |
| if 'mixins' in waterfall: |
| must_be_list(waterfall['mixins'], 'waterfall', waterfall['name']) |
| for mixin in waterfall['mixins']: |
| valid_mixin(mixin) |
| test = self.apply_mixin(self.mixins[mixin], test) |
| |
| if 'mixins' in builder: |
| must_be_list(builder['mixins'], 'builder', builder_name) |
| for mixin in builder['mixins']: |
| valid_mixin(mixin) |
| test = self.apply_mixin(self.mixins[mixin], test) |
| |
| if not 'mixins' in test: |
| return test |
| |
| test_name = test.get('name') |
| if not test_name: |
| test_name = test.get('test') |
| if not test_name: # pragma: no cover |
| # Not the best name, but we should say something. |
| test_name = str(test) |
| must_be_list(test['mixins'], 'test', test_name) |
| for mixin in test['mixins']: |
| valid_mixin(mixin) |
| test = self.apply_mixin(self.mixins[mixin], test) |
| del test['mixins'] |
| return test |
| |
| def apply_mixin(self, mixin, test): |
| """Applies a mixin to a test. |
| |
| Mixins will not override an existing key. This is to ensure exceptions can |
| override a setting a mixin applies. |
| |
| Swarming dimensions are handled in a special way. Instead of specifying |
| 'dimension_sets', which is how normal test suites specify their dimensions, |
| you specify a 'dimensions' key, which maps to a dictionary. This dictionary |
| is then applied to every dimension set in the test. |
| |
| """ |
| new_test = copy.deepcopy(test) |
| mixin = copy.deepcopy(mixin) |
| |
| if 'swarming' in mixin: |
| swarming_mixin = mixin['swarming'] |
| new_test.setdefault('swarming', {}) |
| if 'dimensions' in swarming_mixin: |
| new_test['swarming'].setdefault('dimension_sets', [{}]) |
| for dimension_set in new_test['swarming']['dimension_sets']: |
| dimension_set.update(swarming_mixin['dimensions']) |
| del swarming_mixin['dimensions'] |
| |
| # python dict update doesn't do recursion at all. Just hard code the |
| # nested update we need (mixin['swarming'] shouldn't clobber |
| # test['swarming'], but should update it). |
| new_test['swarming'].update(swarming_mixin) |
| del mixin['swarming'] |
| |
| if '$mixin_append' in mixin: |
| # Values specified under $mixin_append should be appended to existing |
| # lists, rather than replacing them. |
| mixin_append = mixin['$mixin_append'] |
| for key in mixin_append: |
| new_test.setdefault(key, []) |
| if not isinstance(mixin_append[key], list): |
| raise BBGenErr( |
| 'Key "' + key + '" in $mixin_append must be a list.') |
| if not isinstance(new_test[key], list): |
| raise BBGenErr( |
| 'Cannot apply $mixin_append to non-list "' + key + '".') |
| new_test[key].extend(mixin_append[key]) |
| if 'args' in mixin_append: |
| new_test['args'] = self.maybe_fixup_args_array(new_test['args']) |
| del mixin['$mixin_append'] |
| |
| new_test.update(mixin) |
| |
| return new_test |
| |
| def generate_waterfall_json(self, waterfall): |
| all_tests = {} |
| generator_map = self.get_test_generator_map() |
| test_type_remapper = self.get_test_type_remapper() |
| for name, config in waterfall['machines'].iteritems(): |
| tests = {} |
| # Copy only well-understood entries in the machine's configuration |
| # verbatim into the generated JSON. |
| if 'additional_compile_targets' in config: |
| tests['additional_compile_targets'] = config[ |
| 'additional_compile_targets'] |
| for test_type, input_tests in config.get('test_suites', {}).iteritems(): |
| if test_type not in generator_map: |
| raise self.unknown_test_suite_type( |
| test_type, name, waterfall['name']) # pragma: no cover |
| test_generator = generator_map[test_type] |
| # Let multiple kinds of generators generate the same kinds |
| # of tests. For example, gpu_telemetry_tests are a |
| # specialization of isolated_scripts. |
| new_tests = test_generator.generate( |
| waterfall, name, config, input_tests) |
| remapped_test_type = test_type_remapper.get(test_type, test_type) |
| tests[remapped_test_type] = test_generator.sort( |
| tests.get(remapped_test_type, []) + new_tests) |
| all_tests[name] = tests |
| all_tests['AAAAA1 AUTOGENERATED FILE DO NOT EDIT'] = {} |
| all_tests['AAAAA2 See generate_buildbot_json.py to make changes'] = {} |
| return json.dumps(all_tests, indent=2, separators=(',', ': '), |
| sort_keys=True) + '\n' |
| |
| def generate_waterfalls(self): # pragma: no cover |
| self.load_configuration_files() |
| self.resolve_configuration_files() |
| filters = self.args.waterfall_filters |
| suffix = '.json' |
| if self.args.new_files: |
| suffix = '.new' + suffix |
| for waterfall in self.waterfalls: |
| should_gen = not filters or waterfall['name'] in filters |
| if should_gen: |
| file_path = waterfall['name'] + suffix |
| self.write_file(self.pyl_file_path(file_path), |
| self.generate_waterfall_json(waterfall)) |
| |
| def get_valid_bot_names(self): |
| # Extract bot names from infra/config/global/luci-milo.cfg. |
| # NOTE: This reference can cause issues; if a file changes there, the |
| # presubmit here won't be run by default. A manually maintained list there |
| # tries to run presubmit here when luci-milo.cfg is changed. If any other |
| # references to configs outside of this directory are added, please change |
| # their presubmit to run `generate_buildbot_json.py -c`, so that the tree |
| # never ends up in an invalid state. |
| bot_names = set() |
| infra_config_dir = os.path.abspath( |
| os.path.join(os.path.dirname(__file__), |
| '..', '..', 'infra', 'config', 'global')) |
| milo_configs = [ |
| os.path.join(infra_config_dir, 'luci-milo.cfg'), |
| os.path.join(infra_config_dir, 'luci-milo-dev.cfg'), |
| ] |
| for c in milo_configs: |
| for l in self.read_file(c).splitlines(): |
| if (not 'name: "buildbucket/luci.chromium.' in l and |
| not 'name: "buildbot/chromium.' in l and |
| not 'name: "buildbot/tryserver.chromium.' in l): |
| continue |
| # l looks like |
| # `name: "buildbucket/luci.chromium.try/win_chromium_dbg_ng"` |
| # Extract win_chromium_dbg_ng part. |
| bot_names.add(l[l.rindex('/') + 1:l.rindex('"')]) |
| return bot_names |
| |
| def get_bots_that_do_not_actually_exist(self): |
| # Some of the bots on the chromium.gpu.fyi waterfall in particular |
| # are defined only to be mirrored into trybots, and don't actually |
| # exist on any of the waterfalls or consoles. |
| return [ |
| 'ANGLE GPU Win10 Release (Intel HD 630)', |
| 'ANGLE GPU Win10 Release (NVIDIA)', |
| 'Dawn GPU Linux Release (Intel HD 630)', |
| 'Dawn GPU Linux Release (NVIDIA)', |
| 'Dawn GPU Mac Release (Intel)', |
| 'Dawn GPU Mac Retina Release (AMD)', |
| 'Dawn GPU Mac Retina Release (NVIDIA)', |
| 'Dawn GPU Win10 Release (Intel HD 630)', |
| 'Dawn GPU Win10 Release (NVIDIA)', |
| 'Optional Android Release (Nexus 5X)', |
| 'Optional Linux Release (Intel HD 630)', |
| 'Optional Linux Release (NVIDIA)', |
| 'Optional Mac Release (Intel)', |
| 'Optional Mac Retina Release (AMD)', |
| 'Optional Mac Retina Release (NVIDIA)', |
| 'Optional Win10 Release (Intel HD 630)', |
| 'Optional Win10 Release (NVIDIA)', |
| 'Win7 ANGLE Tryserver (AMD)', |
| # chromium.fyi |
| 'linux-blink-rel-dummy', |
| 'mac10.10-blink-rel-dummy', |
| 'mac10.11-blink-rel-dummy', |
| 'mac10.12-blink-rel-dummy', |
| 'mac10.13_retina-blink-rel-dummy', |
| 'mac10.13-blink-rel-dummy', |
| 'win7-blink-rel-dummy', |
| 'win10-blink-rel-dummy', |
| 'Dummy WebKit Mac10.13', |
| 'WebKit Linux composite_after_paint Dummy Builder', |
| 'WebKit Linux layout_ng Dummy Builder', |
| 'WebKit Linux root_layer_scrolls Dummy Builder', |
| # chromium, due to https://crbug.com/878915 |
| 'win-dbg', |
| 'win32-dbg', |
| ] |
| |
| def check_input_file_consistency(self, verbose=False): |
| self.check_input_files_sorting(verbose) |
| |
| self.load_configuration_files() |
| self.flatten_test_suites() |
| self.check_composition_test_suites() |
| |
| # All bots should exist. |
| bot_names = self.get_valid_bot_names() |
| bots_that_dont_exist = self.get_bots_that_do_not_actually_exist() |
| for waterfall in self.waterfalls: |
| for bot_name in waterfall['machines']: |
| if bot_name in bots_that_dont_exist: |
| continue # pragma: no cover |
| if bot_name not in bot_names: |
| if waterfall['name'] in ['client.v8.chromium', 'client.v8.fyi']: |
| # TODO(thakis): Remove this once these bots move to luci. |
| continue # pragma: no cover |
| if waterfall['name'] in ['tryserver.webrtc', |
| 'webrtc.chromium.fyi.experimental']: |
| # These waterfalls have their bot configs in a different repo. |
| # so we don't know about their bot names. |
| continue # pragma: no cover |
| raise self.unknown_bot(bot_name, waterfall['name']) |
| |
| # All test suites must be referenced. |
| suites_seen = set() |
| generator_map = self.get_test_generator_map() |
| for waterfall in self.waterfalls: |
| for bot_name, tester in waterfall['machines'].iteritems(): |
| for suite_type, suite in tester.get('test_suites', {}).iteritems(): |
| if suite_type not in generator_map: |
| raise self.unknown_test_suite_type(suite_type, bot_name, |
| waterfall['name']) |
| if suite not in self.test_suites: |
| raise self.unknown_test_suite(suite, bot_name, waterfall['name']) |
| suites_seen.add(suite) |
| # Since we didn't resolve the configuration files, this set |
| # includes both composition test suites and regular ones. |
| resolved_suites = set() |
| for suite_name in suites_seen: |
| suite = self.test_suites[suite_name] |
| if isinstance(suite, list): |
| for sub_suite in suite: |
| resolved_suites.add(sub_suite) |
| resolved_suites.add(suite_name) |
| # At this point, every key in test_suites.pyl should be referenced. |
| missing_suites = set(self.test_suites.keys()) - resolved_suites |
| if missing_suites: |
| raise BBGenErr('The following test suites were unreferenced by bots on ' |
| 'the waterfalls: ' + str(missing_suites)) |
| |
| # All test suite exceptions must refer to bots on the waterfall. |
| all_bots = set() |
| missing_bots = set() |
| for waterfall in self.waterfalls: |
| for bot_name, tester in waterfall['machines'].iteritems(): |
| all_bots.add(bot_name) |
| # In order to disambiguate between bots with the same name on |
| # different waterfalls, support has been added to various |
| # exceptions for concatenating the waterfall name after the bot |
| # name. |
| all_bots.add(bot_name + ' ' + waterfall['name']) |
| for exception in self.exceptions.itervalues(): |
| removals = (exception.get('remove_from', []) + |
| exception.get('remove_gtest_from', []) + |
| exception.get('modifications', {}).keys()) |
| for removal in removals: |
| if removal not in all_bots: |
| missing_bots.add(removal) |
| |
| missing_bots = missing_bots - set(bots_that_dont_exist) |
| if missing_bots: |
| raise BBGenErr('The following nonexistent machines were referenced in ' |
| 'the test suite exceptions: ' + str(missing_bots)) |
| |
| # All mixins must be referenced |
| seen_mixins = set() |
| for waterfall in self.waterfalls: |
| seen_mixins = seen_mixins.union(waterfall.get('mixins', set())) |
| for bot_name, tester in waterfall['machines'].iteritems(): |
| seen_mixins = seen_mixins.union(tester.get('mixins', set())) |
| for suite in self.test_suites.values(): |
| if isinstance(suite, list): |
| # Don't care about this, it's a composition, which shouldn't include a |
| # swarming mixin. |
| continue |
| |
| for test in suite.values(): |
| if not isinstance(test, dict): |
| # Some test suites have top level keys, which currently can't be |
| # swarming mixin entries. Ignore them |
| continue |
| |
| seen_mixins = seen_mixins.union(test.get('mixins', set())) |
| |
| missing_mixins = set(self.mixins.keys()) - seen_mixins |
| if missing_mixins: |
| raise BBGenErr('The following mixins are unreferenced: %s. They must be' |
| ' referenced in a waterfall, machine, or test suite.' % ( |
| str(missing_mixins))) |
| |
| |
| def type_assert(self, node, typ, filename, verbose=False): |
| """Asserts that the Python AST node |node| is of type |typ|. |
| |
| If verbose is set, it prints out some helpful context lines, showing where |
| exactly the error occurred in the file. |
| """ |
| if not isinstance(node, typ): |
| if verbose: |
| lines = [""] + self.read_file(filename).splitlines() |
| |
| context = 2 |
| lines_start = max(node.lineno - context, 0) |
| # Add one to include the last line |
| lines_end = min(node.lineno + context, len(lines)) + 1 |
| lines = ( |
| ['== %s ==\n' % filename] + |
| ["<snip>\n"] + |
| ['%d %s' % (lines_start + i, line) for i, line in enumerate( |
| lines[lines_start:lines_start + context])] + |
| ['-' * 80 + '\n'] + |
| ['%d %s' % (node.lineno, lines[node.lineno])] + |
| ['-' * (node.col_offset + 3) + '^' + '-' * ( |
| 80 - node.col_offset - 4) + '\n'] + |
| ['%d %s' % (node.lineno + 1 + i, line) for i, line in enumerate( |
| lines[node.lineno + 1:lines_end])] + |
| ["<snip>\n"] |
| ) |
| # Print out a useful message when a type assertion fails. |
| for l in lines: |
| self.print_line(l.strip()) |
| |
| node_dumped = ast.dump(node, annotate_fields=False) |
| # If the node is huge, truncate it so everything fits in a terminal |
| # window. |
| if len(node_dumped) > 60: # pragma: no cover |
| node_dumped = node_dumped[:30] + ' <SNIP> ' + node_dumped[-30:] |
| raise BBGenErr( |
| 'Invalid .pyl file %r. Python AST node %r on line %s expected to' |
| ' be %s, is %s' % ( |
| filename, node_dumped, |
| node.lineno, typ, type(node))) |
| |
| def ensure_ast_dict_keys_sorted(self, node, filename, verbose): |
| is_valid = True |
| |
| keys = [] |
| # The keys of this dict are ordered as ordered in the file; normal python |
| # dictionary keys are given an arbitrary order, but since we parsed the |
| # file itself, the order as given in the file is preserved. |
| for key in node.keys: |
| self.type_assert(key, ast.Str, filename, verbose) |
| keys.append(key.s) |
| |
| keys_sorted = sorted(keys) |
| if keys_sorted != keys: |
| is_valid = False |
| if verbose: |
| for line in difflib.unified_diff( |
| keys, |
| keys_sorted, fromfile='current (%r)' % filename, tofile='sorted'): |
| self.print_line(line) |
| |
| if len(set(keys)) != len(keys): |
| for i in range(len(keys_sorted)-1): |
| if keys_sorted[i] == keys_sorted[i+1]: |
| self.print_line('Key %s is duplicated' % keys_sorted[i]) |
| is_valid = False |
| return is_valid |
| |
| def check_input_files_sorting(self, verbose=False): |
| # TODO(https://crbug.com/886993): Add the ability for this script to |
| # actually format the files, rather than just complain if they're |
| # incorrectly formatted. |
| bad_files = set() |
| |
| for filename in ( |
| 'mixins.pyl', |
| 'test_suites.pyl', |
| 'test_suite_exceptions.pyl', |
| ): |
| parsed = ast.parse(self.read_file(self.pyl_file_path(filename))) |
| |
| # Must be a module. |
| self.type_assert(parsed, ast.Module, filename, verbose) |
| module = parsed.body |
| |
| # Only one expression in the module. |
| self.type_assert(module, list, filename, verbose) |
| if len(module) != 1: # pragma: no cover |
| raise BBGenErr('Invalid .pyl file %s' % filename) |
| expr = module[0] |
| self.type_assert(expr, ast.Expr, filename, verbose) |
| |
| # Value should be a dictionary. |
| value = expr.value |
| self.type_assert(value, ast.Dict, filename, verbose) |
| |
| if filename == 'test_suites.pyl': |
| expected_keys = ['basic_suites', 'compound_suites'] |
| actual_keys = [node.s for node in value.keys] |
| assert all(key in expected_keys for key in actual_keys), ( |
| 'Invalid %r file; expected keys %r, got %r' % ( |
| filename, expected_keys, actual_keys)) |
| suite_dicts = [node for node in value.values] |
| # Only two keys should mean only 1 or 2 values |
| assert len(suite_dicts) <= 2 |
| for suite_group in suite_dicts: |
| if not self.ensure_ast_dict_keys_sorted( |
| suite_group, filename, verbose): |
| bad_files.add(filename) |
| |
| else: |
| if not self.ensure_ast_dict_keys_sorted( |
| value, filename, verbose): |
| bad_files.add(filename) |
| |
| # waterfalls.pyl is slightly different, just do it manually here |
| filename = 'waterfalls.pyl' |
| parsed = ast.parse(self.read_file(self.pyl_file_path(filename))) |
| |
| # Must be a module. |
| self.type_assert(parsed, ast.Module, filename, verbose) |
| module = parsed.body |
| |
| # Only one expression in the module. |
| self.type_assert(module, list, filename, verbose) |
| if len(module) != 1: # pragma: no cover |
| raise BBGenErr('Invalid .pyl file %s' % filename) |
| expr = module[0] |
| self.type_assert(expr, ast.Expr, filename, verbose) |
| |
| # Value should be a list. |
| value = expr.value |
| self.type_assert(value, ast.List, filename, verbose) |
| |
| keys = [] |
| for val in value.elts: |
| self.type_assert(val, ast.Dict, filename, verbose) |
| waterfall_name = None |
| for key, val in zip(val.keys, val.values): |
| self.type_assert(key, ast.Str, filename, verbose) |
| if key.s == 'machines': |
| if not self.ensure_ast_dict_keys_sorted(val, filename, verbose): |
| bad_files.add(filename) |
| |
| if key.s == "name": |
| self.type_assert(val, ast.Str, filename, verbose) |
| waterfall_name = val.s |
| assert waterfall_name |
| keys.append(waterfall_name) |
| |
| if sorted(keys) != keys: |
| bad_files.add(filename) |
| if verbose: # pragma: no cover |
| for line in difflib.unified_diff( |
| keys, |
| sorted(keys), fromfile='current', tofile='sorted'): |
| self.print_line(line) |
| |
| if bad_files: |
| raise BBGenErr( |
| 'The following files have invalid keys: %s\n. They are either ' |
| 'unsorted, or have duplicates.' % ', '.join(bad_files)) |
| |
| def check_output_file_consistency(self, verbose=False): |
| self.load_configuration_files() |
| # All waterfalls must have been written by this script already. |
| self.resolve_configuration_files() |
| ungenerated_waterfalls = set() |
| for waterfall in self.waterfalls: |
| expected = self.generate_waterfall_json(waterfall) |
| file_path = waterfall['name'] + '.json' |
| current = self.read_file(self.pyl_file_path(file_path)) |
| if expected != current: |
| ungenerated_waterfalls.add(waterfall['name']) |
| if verbose: # pragma: no cover |
| self.print_line('Waterfall ' + waterfall['name'] + |
| ' did not have the following expected ' |
| 'contents:') |
| for line in difflib.unified_diff( |
| expected.splitlines(), |
| current.splitlines(), |
| fromfile='expected', tofile='current'): |
| self.print_line(line) |
| if ungenerated_waterfalls: |
| raise BBGenErr('The following waterfalls have not been properly ' |
| 'autogenerated by generate_buildbot_json.py: ' + |
| str(ungenerated_waterfalls)) |
| |
| def check_consistency(self, verbose=False): |
| self.check_input_file_consistency(verbose) # pragma: no cover |
| self.check_output_file_consistency(verbose) # pragma: no cover |
| |
| def parse_args(self, argv): # pragma: no cover |
| parser = argparse.ArgumentParser() |
| parser.add_argument( |
| '-c', '--check', action='store_true', help= |
| 'Do consistency checks of configuration and generated files and then ' |
| 'exit. Used during presubmit. Causes the tool to not generate any files.') |
| parser.add_argument( |
| '-n', '--new-files', action='store_true', help= |
| 'Write output files as .new.json. Useful during development so old and ' |
| 'new files can be looked at side-by-side.') |
| parser.add_argument( |
| '-v', '--verbose', action='store_true', help= |
| 'Increases verbosity. Affects consistency checks.') |
| parser.add_argument( |
| 'waterfall_filters', metavar='waterfalls', type=str, nargs='*', |
| help='Optional list of waterfalls to generate.') |
| parser.add_argument( |
| '--pyl-files-dir', type=os.path.realpath, |
| help='Path to the directory containing the input .pyl files.') |
| self.args = parser.parse_args(argv) |
| |
| def main(self, argv): # pragma: no cover |
| self.parse_args(argv) |
| if self.args.check: |
| self.check_consistency(verbose=self.args.verbose) |
| else: |
| self.generate_waterfalls() |
| return 0 |
| |
| if __name__ == "__main__": # pragma: no cover |
| generator = BBJSONGenerator() |
| sys.exit(generator.main(sys.argv[1:])) |
