# ----------------------------------------------------------------------------- # Copyright (c) Microsoft Corporation. All rights reserved. # Licensed under the MIT License. See License.txt in the project root for # license information. # ----------------------------------------------------------------------------- # pylint: disable=line-too-long from difflib import context_diff from enum import Enum from importlib import import_module import inspect import os import re from pkgutil import iter_modules from typing import List, Tuple import yaml from knack.log import get_logger from azdev.operations.regex import ( get_all_tested_commands_from_regex, search_aaz_raw_command, search_aaz_custom_command, search_argument, search_argument_context, search_command, search_deleted_command, search_command_group) from azdev.utilities import diff_branches_detail, diff_branch_file_patch from azdev.utilities.path import get_cli_repo_path, get_ext_repo_paths from .util import (share_element, exclude_commands, LinterError, get_cmd_example_configurations, get_cmd_example_threshold) PACKAGE_NAME = 'azdev.operations.linter' _logger = get_logger(__name__) class LinterSeverity(Enum): HIGH = 2 MEDIUM = 1 LOW = 0 @staticmethod def get_linter_severity(severity_name): for severity in LinterSeverity: if severity_name.lower() == severity.name.lower(): return severity raise ValueError("Severity must be a valid linter severity name or value.") @staticmethod def get_ordered_members(): return sorted(LinterSeverity, key=lambda sev: sev.value) class Linter: # pylint: disable=too-many-public-methods, too-many-instance-attributes def __init__(self, command_loader=None, help_file_entries=None, loaded_help=None, git_source=None, git_target=None, git_repo=None, exclusions=None): self._all_yaml_help = help_file_entries self._loaded_help = loaded_help self._command_loader = command_loader self._parameters = {} self._help_file_entries = set(help_file_entries.keys()) self._command_parser = command_loader.cli_ctx.invocation.parser self._command_groups = [] for command_name, command in self._command_loader.command_table.items(): self._parameters[command_name] = set() for name in command.arguments: self._parameters[command_name].add(name) self.git_source = git_source self.git_target = git_target self.git_repo = git_repo self.exclusions = exclusions self.diffed_lines = set() self._get_diffed_patches() @property def commands(self): return self._command_loader.command_table.keys() @property def command_groups(self): if not self._command_groups: added_command_groups = set() for command_group in self._command_loader.command_group_table.keys(): prefix_name = "" for word in command_group.split(): prefix_name = "{} {}".format(prefix_name, word).strip() if prefix_name in added_command_groups: # if the parent command group is added continue continue added_command_groups.add(prefix_name) self._command_groups.append(prefix_name) return self._command_groups @property def help_file_entries(self): return self._help_file_entries @property def command_parser(self): return self._command_parser @property def command_loader_map(self): return self._command_loader.cmd_to_loader_map def get_command_metadata(self, command_name): try: return self._command_loader.command_table[command_name] except KeyError: return None def get_command_parameters(self, command_name): return self._parameters.get(command_name) def get_command_group_metadata(self, command_group_name): try: return self._command_loader.command_group_table[command_group_name] except KeyError: return None def get_help_entry_type(self, entry_name): return self._all_yaml_help.get(entry_name).get('type') def get_help_entry_examples(self, entry_name): return self._all_yaml_help.get(entry_name).get('examples', []) def get_help_entry_parameter_names(self, entry_name): return [param_help.get('name', None) for param_help in self._all_yaml_help.get(entry_name).get('parameters', [])] def is_valid_parameter_help_name(self, entry_name, param_name): return param_name in [param.name for param in getattr(self._loaded_help.get(entry_name), 'parameters', [])] def get_command_help(self, command_name): return self._get_loaded_help_description(command_name) def get_command_group_help(self, command_group_name): return self._get_loaded_help_description(command_group_name) def get_parameter_options(self, command_name, parameter_name): return self.get_command_metadata(command_name).arguments.get(parameter_name).type.settings.get('options_list') def get_parameter_help(self, command_name, parameter_name): options = self.get_parameter_options(command_name, parameter_name) command_help = self._loaded_help.get(command_name, None) if not command_help: return None parameter_helps = command_help.parameters param_help = next((param for param in parameter_helps if share_element(options, param.name.split())), None) # workaround for --ids which is not does not generate doc help (BUG) if not param_help: command_args = self._command_loader.command_table.get(command_name).arguments return command_args.get(parameter_name).type.settings.get('help') return param_help.short_summary or param_help.long_summary def get_parameter_settings(self, command_name, parameter_name): return self.get_command_metadata(command_name).arguments.get(parameter_name).type.settings def get_parameter_help_info(self, command_name, parameter_name): options = self.get_parameter_options(command_name, parameter_name) command_help = self._loaded_help.get(command_name, None) if not command_help: return None parameter_helps = command_help.parameters param_help = next((param for param in parameter_helps if share_element(options, param.name.split())), None) return param_help def command_expired(self, command_name): deprecate_info = self._command_loader.command_table[command_name].deprecate_info if deprecate_info: return deprecate_info.expired() return False def command_group_expired(self, command_group_name): try: group_kwargs = self._command_loader.command_group_table[command_group_name].group_kwargs deprecate_info = group_kwargs.get('deprecate_info', None) if deprecate_info: return deprecate_info.expired() except KeyError: # ignore command_group_name which is not in command_group_table. pass except AttributeError: # Items with only token presence in the command table will not have any data. They can't be expired. pass return False def parameter_expired(self, command_name, parameter_name): parameter = self._command_loader.command_table[command_name].arguments[parameter_name].type.settings deprecate_info = parameter.get('deprecate_info', None) if deprecate_info: return deprecate_info.expired() return False def option_expired(self, command_name, parameter_name): from knack.deprecation import Deprecated parameter = self._command_loader.command_table[command_name].arguments[parameter_name].type.settings options_list = parameter.get('options_list', []) expired_options_list = [] for opt in options_list: if isinstance(opt, Deprecated) and opt.expired(): expired_options_list.append(opt.target) return expired_options_list def _get_loaded_help_description(self, entry): help_entry = self._loaded_help.get(entry, None) if help_entry: return help_entry.short_summary or help_entry.long_summary return help_entry def get_loaded_help_entry(self, entry): help_entry = self._loaded_help.get(entry, None) return help_entry def get_command_test_coverage(self): diff_index = diff_branches_detail(repo=self.git_repo, target=self.git_target, source=self.git_source) commands, _ = self._detect_new_command(diff_index) all_tested_command = self._detect_tested_command(diff_index) return self._run_command_test_coverage(commands, all_tested_command) def get_parameter_test_coverage(self): diff_index = diff_branches_detail(repo=self.git_repo, target=self.git_target, source=self.git_source) _, parameters = self._detect_new_command(diff_index) all_tested_command = self._detect_tested_command(diff_index) return self._run_parameter_test_coverage(parameters, all_tested_command) def check_missing_command_example(self): _exclude_commands = self._get_cmd_exclusions(rule_name="missing_command_example") cmd_example_config = get_cmd_example_configurations() commands = self._detect_modified_command() violations = [] for cmd in commands: if cmd in _exclude_commands: continue cmd_help = self._loaded_help.get(cmd, None) if not cmd_help: continue cmd_suffix = cmd.split()[-1] cmd_example_threshold = get_cmd_example_threshold(cmd_suffix, cmd_example_config) if cmd_example_threshold == 0: continue if not hasattr(cmd_help, "parameters") or len(cmd_help.parameters) == 0: # skip cmd without parameters continue if not hasattr(cmd_help, "examples") or len(cmd_help.examples) < cmd_example_threshold: violations.append(f'Command `{cmd}` should have at least {cmd_example_threshold} example(s)') if violations: violations.insert(0, 'Check command example failed.') violations.extend(['Please add examples for the modified command or add the command in rule_exclusions: missing_command_example in linter_exclusions.yml']) return violations def _get_exclusions(self): _exclude_commands = set() _exclude_parameters = set() for command, details in self.exclusions.items(): if 'parameters' in details: for param, rules in details['parameters'].items(): if 'missing_parameter_test_coverage' in rules['rule_exclusions']: _exclude_parameters.add((command, param)) if 'rule_exclusions' in details and 'missing_command_test_coverage' in details['rule_exclusions']: _exclude_commands.add(command) _logger.debug('exclude_parameters: %s', _exclude_parameters) _logger.debug('exclude_comands: %s', _exclude_commands) return _exclude_commands, _exclude_parameters def _get_cmd_exclusions(self, rule_name=None): _exclude_commands = set() if not rule_name: return _exclude_commands for command, details in self.exclusions.items(): if 'rule_exclusions' in details and rule_name in details['rule_exclusions']: _exclude_commands.add(command) _logger.debug('exclude_commands: %s', _exclude_commands) return _exclude_commands def _split_path(self, path: str): parts = path.rsplit('/', maxsplit=1) return parts if len(parts) == 2 else ('', parts[0]) def _read_blob_lines(self, blob): return blob.data_stream.read().decode("utf-8").splitlines(True) if blob else [] def _get_line_number(self, lines: List[str], row_num: int, pattern: str): offset = -1 while row_num > 0: row_num -= 1 match = re.findall(pattern, lines[row_num]) offset += 1 if match: return int(match[0]) + offset return -1 def _extract_parameters(self, lines, current_lines, _exclude_commands, _exclude_parameters, parameters): for row_num, line in enumerate(lines): params, param_name = search_argument(line) if params: idx = self._get_line_number(lines, row_num, r'--- (\d+),(?:\d+) ----') commands = search_argument_context(idx, current_lines) for cmd in commands: if cmd not in _exclude_commands and (cmd, param_name) not in _exclude_parameters: parameters.append((cmd, params)) _logger.debug('Detected parameter: [%s, %s]', cmd, params) return parameters def _extract_commands(self, lines, original_lines, current_lines, added_commands, deleted_commands, _exclude_commands, yellow_color): for row_num, line in enumerate(lines): added_command = search_command(line) deleted_command = search_deleted_command(line) if added_command: idx = self._get_line_number(lines, row_num, r'--- (\d+),(?:\d+) ----') cmd = search_command_group(idx, current_lines, added_command) if cmd: if cmd in _exclude_commands: _logger.warning('%sCommand %s not tested and excluded in linter_exclusions.yml', yellow_color, cmd) else: added_commands.add(cmd) elif deleted_command: idx = self._get_line_number(lines, row_num, r'\*\*\* (\d+),(?:\d+) \*\*\*\*') cmd = search_command_group(idx, original_lines, deleted_command) if cmd: deleted_commands.add(cmd) return added_commands, deleted_commands def _detect_new_command(self, diff_index: List) -> Tuple[List[str], List[Tuple[str, str]]]: YELLOW = '\x1b[33m' _exclude_commands, _exclude_parameters = self._get_exclusions() added_commands, deleted_commands, parameters = set(), set(), [] for diff in diff_index: _, filename = self._split_path(diff.a_path) if not any(key in filename for key in ('params', 'commands')): continue original_lines = self._read_blob_lines(diff.a_blob) current_lines = self._read_blob_lines(diff.b_blob) lines = list(context_diff(original_lines, current_lines, 'Original', 'Current')) if 'params.py' in filename: parameters = self._extract_parameters(lines, current_lines, _exclude_commands, _exclude_parameters, parameters) if 'commands.py' in filename: added_commands, deleted_commands = self._extract_commands(lines, original_lines, current_lines, added_commands, deleted_commands, _exclude_commands, YELLOW) commands = list(added_commands - deleted_commands) _logger.debug('New parameters: %s', parameters) _logger.debug('Added commands: %s', added_commands) _logger.debug('Deleted commands: %s', deleted_commands) _logger.debug('Final commands: %s', commands) return commands, parameters def _detect_tested_command(self, diff_index): all_tested_command = [] # get tested command by regex for diff in diff_index: filename = diff.a_path.split('/')[-1] if re.findall(r'^test_.*\.py$', filename) and \ os.path.exists(os.path.join(self.git_repo, diff.a_path)): with open(os.path.join(self.git_repo, diff.a_path), encoding='utf-8') as f: lines = f.readlines() ref = get_all_tested_commands_from_regex(lines) all_tested_command += ref # get tested command by recording file if re.findall(r'^test_.*\.yaml$', filename) and \ os.path.exists(os.path.join(self.git_repo, diff.a_path)): with open(os.path.join(self.git_repo, diff.a_path)) as f: records = yaml.load(f, Loader=yaml.Loader) or {} for record in records['interactions']: # parse command ['acr agentpool create'] command = record['request']['headers'].get('CommandName', [''])[0] # parse argument ['-n -r'] argument = record['request']['headers'].get('ParameterSetName', [''])[0] if command or argument: all_tested_command.append(command + ' ' + argument) _logger.debug('All tested command: %s', all_tested_command) return all_tested_command @staticmethod def _run_command_test_coverage(commands, all_tested_command): exec_state = True violations = [] for command in commands: for code in all_tested_command: if command in code: break else: violations.append(f'Missing command test coverage: `{command}`') exec_state = False if violations: violations.insert(0, 'Failed.') violations.extend([ 'Please add some scenario tests for the new command', 'Or add the command with missing_command_test_coverage rule in linter_exclusions.yml']) return exec_state, violations @staticmethod def _run_parameter_test_coverage(parameters, all_tested_command): flag = False exec_state = True violations = [] for command, opt_list in parameters: for opt in opt_list: for code in all_tested_command: if command in code and opt in code: flag = True break if flag: break else: violations.append(f'Missing parameter test coverage: `{command} {opt}`') exec_state = False if violations: violations.insert(0, 'Failed.') violations.extend([ 'Please add some scenario tests for the new parameter', 'Or add the parameter with missing_parameter_test_coverage rule in linter_exclusions.yml']) return exec_state, violations def _detect_modified_command(self): modified_commands = set() diff_patches = diff_branch_file_patch(repo=self.git_repo, target=self.git_target, source=self.git_source) for change in diff_patches: if not change.b_path or not change.diff: continue file_path, filename = self._split_path(change.b_path) if "commands.py" not in filename and "aaz" not in file_path: continue current_lines = self._read_blob_lines(change.b_blob) patch = change.diff.decode("utf-8") patch_lines = patch.splitlines() if 'commands.py' in filename: added_lines = [line for line in patch_lines if line.startswith('+') and not line.startswith('+++')] for line in added_lines: if aaz_custom_command := search_aaz_custom_command(line): modified_commands.add(aaz_custom_command) for row_num, line in enumerate(patch_lines): if not line.startswith("+") or line.startswith('+++'): continue manual_command_suffix = search_command(line) if manual_command_suffix: idx = self._get_line_number(patch_lines, row_num, r'@@ -(\d+),(?:\d+) \+(?:\d+),(?:\d+) @@') manual_command = search_command_group(idx, current_lines, manual_command_suffix) if manual_command: modified_commands.add(manual_command) if "aaz" in file_path: if aaz_raw_command := search_aaz_raw_command(patch): modified_commands.add(aaz_raw_command) commands = list(modified_commands) _logger.debug('Modified commands: %s', modified_commands) return commands def _get_diffed_patches(self): if not self.git_source or not self.git_target or not self.git_repo: return diff_patches = diff_branch_file_patch(repo=self.git_repo, target=self.git_target, source=self.git_source) for change in diff_patches: if not change.diff: continue patch = change.diff.decode("utf-8") added_lines = [line for line in patch.splitlines() if line.startswith('+') and not line.startswith('+++')] self.diffed_lines |= set(added_lines) if added_lines: _logger.info("Changes in file '%s':", change.a_path) for line in added_lines: _logger.info(line) # pylint: disable=too-many-instance-attributes class LinterManager: _RULE_TYPES = {'help_file_entries', 'command_groups', 'commands', 'params', 'command_test_coverage'} def __init__(self, command_loader=None, help_file_entries=None, loaded_help=None, exclusions=None, rule_inclusions=None, use_ci_exclusions=None, min_severity=None, update_global_exclusion=None, git_source=None, git_target=None, git_repo=None): # default to running only rules of the highest severity self.min_severity = min_severity or LinterSeverity.get_ordered_members()[-1] self._exclusions = exclusions or {} self.linter = Linter(command_loader=command_loader, help_file_entries=help_file_entries, loaded_help=loaded_help, git_source=git_source, git_target=git_target, git_repo=git_repo, exclusions=self._exclusions) self._rules = {rule_type: {} for rule_type in LinterManager._RULE_TYPES} # initialize empty rules self._ci_exclusions = {} self._rule_inclusions = rule_inclusions self._loaded_help = loaded_help self._command_loader = command_loader self._help_file_entries = help_file_entries self._exit_code = 0 self._ci = use_ci_exclusions if use_ci_exclusions is not None else os.environ.get('CI', False) self._violiations = {} self._update_global_exclusion = update_global_exclusion def add_rule(self, rule_type, rule_name, rule_callable, rule_severity): include_rule = not self._rule_inclusions or rule_name in self._rule_inclusions if rule_type in self._rules and include_rule: def get_linter(): # if a rule has exclusions return a linter that factors in those exclusions # otherwise return the main linter. if rule_name in self._ci_exclusions and self._ci: mod_exclusions = self._ci_exclusions[rule_name] command_loader, help_file_entries = exclude_commands( self._command_loader, self._help_file_entries, mod_exclusions) return Linter(command_loader=command_loader, help_file_entries=help_file_entries, loaded_help=self._loaded_help) return self.linter self._rules[rule_type][rule_name] = rule_callable, get_linter, rule_severity def mark_rule_failure(self, rule_severity): if rule_severity is LinterSeverity.HIGH: self._exit_code = 1 @property def exclusions(self): return self._exclusions @property def exit_code(self): return self._exit_code 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 def _run_rules(self, rule_group): # https://docs.microsoft.com/en-us/windows/console/console-virtual-terminal-sequences#text-formatting RED = '\x1b[31m' GREEN = '\x1b[32m' YELLOW = '\x1b[33m' CYAN = '\x1b[36m' RESET = '\x1b[39m' # print('enter _run_rules') for rule_name, (rule_func, linter_callable, rule_severity) in self._rules.get(rule_group).items(): # print('enter_items') severity_str = rule_severity.name # use new linter if needed with LinterScope(self, linter_callable): # print('enter_with') # if the rule's severity is lower than the linter's severity skip it. if self._linter_severity_is_applicable(rule_severity, rule_name): # print('enter violations', rule_func) violations = sorted(rule_func()) or [] # print('enter to find') if violations: if rule_severity == LinterSeverity.HIGH: sev_color = RED elif rule_severity == LinterSeverity.MEDIUM: sev_color = YELLOW else: sev_color = CYAN # pylint: disable=duplicate-string-formatting-argument print('- {} FAIL{} - {}{}{} severity: {}'.format(RED, RESET, sev_color, severity_str, RESET, rule_name, )) for violation_msg, entity_name, name in violations: print(violation_msg) self._save_violations(entity_name, name) print() else: print('- {} pass{}: {} '.format(GREEN, RESET, rule_name)) # print('enter_end') def _linter_severity_is_applicable(self, rule_severity, rule_name): if self.min_severity.value > rule_severity.value: _logger.info("Skipping rule %s, because its severity '%s' is lower than the linter's min severity of '%s'.", rule_name, rule_severity.name, self.min_severity.value) return False return True # pylint: disable=line-too-long def _save_violations(self, entity_name, rule_name): if isinstance(entity_name, str): command_name = entity_name self._violiations.setdefault(command_name, {}).setdefault('rule_exclusions', []).append(rule_name) else: command_name, param_name = entity_name self._violiations.setdefault(command_name, {}).setdefault('parameters', {}).setdefault(param_name, {}).setdefault( 'rule_exclusions', []).append(rule_name) class RuleError(Exception): """ Exception thrown by rule violation """ pass # pylint: disable=unnecessary-pass class LinterScope: """ Linter Context manager. used when calling a rule function. Allows substitution of main linter for a linter that takes into account any applicable exclusions, if applicable. """ def __init__(self, linter_manager, linter_callable): self.linter_manager = linter_manager self.linter = linter_callable() self.main_linter = linter_manager.linter def __enter__(self): self.linter_manager.linter = self.linter def __exit__(self, exc_type, value, traceback): self.linter_manager.linter = self.main_linter