def run()

in azdev/operations/linter/linter.py [0:0]

• 45 lines of code
• 26 McCabe index (conditional complexity)

    def run(self, run_params=None, run_commands=None, run_command_groups=None,
            run_help_files_entries=None, run_command_test_coverage=None):
        paths = import_module('{}.rules'.format(PACKAGE_NAME)).__path__

        if paths:
            ci_exclusions_path = os.path.join(paths[0], 'ci_exclusions.yml')
            with open(ci_exclusions_path) as f:
                self._ci_exclusions = yaml.safe_load(f) or {}

        # find all defined rules and check for name conflicts
        found_rules = set()
        for _, name, _ in iter_modules(paths):
            rule_module = import_module('{}.rules.{}'.format(PACKAGE_NAME, name))
            functions = inspect.getmembers(rule_module, inspect.isfunction)
            for rule_name, add_to_linter_func in functions:
                if hasattr(add_to_linter_func, 'linter_rule'):
                    if rule_name in found_rules:
                        raise LinterError('Multiple rules found with the same name: %s' % rule_name)
                    found_rules.add(rule_name)
                    add_to_linter_func(self)

        # run all rule-checks
        if run_help_files_entries and self._rules.get('help_file_entries'):
            # print('help_file_entries')
            self._run_rules('help_file_entries')

        if run_command_groups and self._rules.get('command_groups'):
            # print('command_groups')
            self._run_rules('command_groups')

        if run_commands and self._rules.get('commands'):
            self._run_rules('commands')

        if run_params and self._rules.get('params'):
            self._run_rules('params')

        if run_command_test_coverage and self._rules.get('command_test_coverage'):
            self._run_rules('command_test_coverage')

        if not self.exit_code:
            print(os.linesep + 'No violations found for linter rules.')

        if self._update_global_exclusion is not None:
            if self._update_global_exclusion == 'CLI':
                repo_paths = [get_cli_repo_path()]
            else:
                repo_paths = get_ext_repo_paths()
            exclusion_paths = [os.path.join(repo_path, 'linter_exclusions.yml') for repo_path in repo_paths]
            for exclusion_path in exclusion_paths:
                if not os.path.isfile(exclusion_path):
                    with open(exclusion_path, 'a'):
                        pass

                with open(exclusion_path) as f:
                    exclusions = yaml.safe_load(f) or {}
                exclusions.update(self._violiations)

                with open(exclusion_path, 'w') as f:
                    yaml.safe_dump(exclusions, f)

        return self.exit_code