diff options
Diffstat (limited to 'buildscripts/validate_commit_message.py')
| -rwxr-xr-x | buildscripts/validate_commit_message.py | 421 |
1 files changed, 29 insertions, 392 deletions
diff --git a/buildscripts/validate_commit_message.py b/buildscripts/validate_commit_message.py index 37973faa584..03283b9d940 100755 --- a/buildscripts/validate_commit_message.py +++ b/buildscripts/validate_commit_message.py @@ -28,398 +28,36 @@ # """Validate that the commit message is ok.""" import argparse -import collections -import logging +import os import re import subprocess import sys -from contextlib import contextmanager -from enum import IntEnum -from http import HTTPStatus -from http.client import HTTPConnection # py3 -from typing import Dict, List, Type, Tuple, Optional, Match, Any - -import requests.exceptions -from jira import JIRA, Issue -from jira.exceptions import JIRAError VALID_PATTERNS = [ - # NOTE: re.VERBOSE is for visibility / debugging. As such significant white space must be - # escaped (e.g ' ' to \s). - re.compile( - r''' - ^ - ((?P<revert>Revert)\s*[\"\']?)? # Revert (optional) - ((?P<ticket>(?:EVG|SERVER|WT)-[0-9]+)\s*) # ticket identifier - #(?P<body>(?:(?!\(cherry\spicked\sfrom).)*)? # To also capture the body - ((?:(?!\(cherry\spicked\sfrom).)*)? # negative lookahead backport - (?P<backport>\(cherry\spicked\sfrom.*)? # back port (optional) - ''', re.MULTILINE | re.DOTALL | re.VERBOSE), - re.compile(r'(?P<lint>^Fix lint$)'), # Allow "Fix lint" as the sole commit summary - re.compile(r'(?P<imported>^Import (wiredtiger|tools): .*)'), # These are public tickets + re.compile(r"^Fix lint$"), # Allow "Fix lint" as the sole commit summary + re.compile(r'^(Revert ["\']?)?(EVG|SERVER|WT)-[0-9]+'), # These are public tickets + re.compile(r'^Import (wiredtiger|tools):'), # These are public tickets ] -"""valid public patterns.""" - -PRIVATE_PATTERNS = [re.compile(r'^(?P<ticket>[A-Z]+-[0-9]+)')] -"""private patterns.""" - -INVALID_JIRA_STATUS = ('closed', ) -"""List of lower cased invalid jira status strings.""" - -GIT_SHOW_COMMAND = ['git', 'show', '-1', '-s', '--format=%s'] -"""git command line to get the last commit message.""" - -DEFAULT_JIRA = 'https://jira.mongodb.org' -LOGGER = logging.getLogger(__name__) - - -class Status(IntEnum): - """Status enumeration values.""" - - OK = 0 - ERROR = 1 - WARNING = 2 - - -class Violation(collections.namedtuple('Fault', ['status', 'message'])): - """validation issue holder.""" - - def __str__(self): - return str(self.message) - - -def get_full_message(message: List[str]) -> str: - """ - Convert the message to a single string or get the last git commit message. - - If the input list is empty then the last git commit message is used. - - :param message: A list of the message components. - :return: The message. - """ - LOGGER.info('get commit message') - if not message: - LOGGER.info('Validating last git commit message') - result = subprocess.check_output(GIT_SHOW_COMMAND) - message = result.decode('utf-8') - else: - message = " ".join(message) - LOGGER.info('Validating commit message \'%s\'', message) - return message - - -def find_ticket(message: str) -> Dict: - """ - Find ticket data in message. - - :param message: The commit message. - :return: A dict of the commit message components (may be empty). - """ - ticket = find_matching_pattern(message, VALID_PATTERNS) - if ticket: - ticket['public'] = True - else: - ticket = find_matching_pattern(message, PRIVATE_PATTERNS) - if ticket: - ticket['public'] = False - return ticket - - -def find_matching_pattern(message: str, patterns: List[Match]) -> Dict: - """ - Find the first matching pattern. - - :param message: The commit message. - :param patterns: A list of regular expressions. - :return: A dict of the commit message components (may be empty). - """ - for valid_pattern in patterns: - matching_pattern = valid_pattern.match(message) - # pattern matches and there is a ticket - if matching_pattern: - return matching_pattern.groupdict() - return {} - - -def validate_message(message: str, author: str, - jira: Optional[Type[JIRA]]) -> Tuple[Dict, List[Violation]]: - """ - Validate the commit message. - - :param message: The commit message. - :param author: The author. - :param jira: The jira connection. - :return: The ticket dict and violations. - """ - LOGGER.info('validating message') - if not message.strip(): - ticket = {} - violations = [Violation(Status.ERROR, 'found empty commit message')] - else: - ticket = find_ticket(message) - violations = validate_ticket(ticket, author, jira) - return ticket, violations - - -def validate_ticket(ticket: Dict, author: str, jira: Optional[Type[JIRA]]) -> List[Violation]: - """ - Validate the ticket and commit message. - - :param ticket: The extract ticket information. - :param author: The author. - :param jira: The jira connection. - :return: The violations. - """ - violations = [] - - if not ticket: - violations.append(Violation(Status.WARNING, 'found a commit without a ticket')) - elif ticket['public']: - violations = validate_public_ticket(ticket, author, jira) - else: - tid = ticket['ticket'] - violations.append(Violation(Status.ERROR, f'private project: {tid}')) - - return violations - - -def validate_status(issue: Type[Issue]) -> Optional[Violation]: - """ - Validate that the issue status is allowed. - - :param issue: The jira issue. - :return: A violation (if applicable). - """ - status = str(issue.fields.status).lower() - if status in INVALID_JIRA_STATUS: - return Violation(Status.ERROR, f'status cannot be {status}') - return None - +PRIVATE_PATTERNS = [re.compile(r"^[A-Z]+-[0-9]+")] -def validate_author(issue: Type[Issue], ticket: Dict, author: str) -> Violation: - """ - Validate that the issue author is correct. +STATUS_OK = 0 +STATUS_ERROR = 1 - :param issue: The jira issue. - :param ticket: The ticket data. - :param author: The expected author. - :return: A violation (if applicable). - """ - assignee = issue.fields.assignee - if assignee.name != author: - details = (f'assignee is not author \'{assignee.name}\'(' - f' \'{assignee.displayName}\') != \'{author}\'') - if not ticket.get('backport', False): - return Violation(Status.WARNING, details) - else: - LOGGER.debug('%s but this is a backport', details) - return None +GIT_SHOW_COMMAND = ["git", "show", "-1", "-s", "--format=%s"] -def validate_public_ticket(ticket: Dict, author: str, jira: Type[JIRA], - verbose: bool = False) -> List[Violation]: - """ - Validate the status of a public ticket. - - :param ticket: The extract ticket information. - :param author: The author. - :param jira: The jira connection. - :param verbose: A flag to enable / disable verbose output. - :return: The violations. - """ - violations = [] - ticket_id = ticket['ticket'] - try: - if jira is not None: - with silence(verbose): - issue = jira.issue(ticket_id) - if issue: - violation = validate_status(issue) - if violation: - violations.append(violation) - - violation = validate_author(issue, ticket, author) - if violation: - violations.append(violation) - - else: - LOGGER.debug('unable to fully validate issue \'%s\'', ticket_id) - violations.append( - Violation(Status.WARNING, f'{ticket_id}: unable to validate with jira')) - except requests.exceptions.ConnectionError: - LOGGER.debug('%s: unexpected connection exception', exc_info=True) - violations.append( - Violation(Status.WARNING, f'{ticket_id}: unexpected connection exception')) - except JIRAError as ex: - LOGGER.debug('unexpected jira exception', exc_info=True) - if ex.status_code == HTTPStatus.NOT_FOUND: - violation = Violation(Status.ERROR, f'{ticket_id}: not found') - elif ex.status_code == HTTPStatus.UNAUTHORIZED: - violation = Violation(Status.ERROR, f'{ticket_id}: private (unauthorized)') - else: - violation = Violation(Status.WARNING, - f'{ticket_id}: unexpected jira error {ex.status_code}') - violations.append(violation) - except ValueError: - LOGGER.debug('unexpected exception', exc_info=True) - violations.append(Violation(Status.WARNING, f'{ticket_id}: unexpected exception')) - - return violations - - -def handle_violations(ticket: Dict, message: str, violations: List[Violation], - warning_as_errors: bool) -> Type[Status]: - """ - Handle any validation issues found. - - :param ticket: The extract ticket information. - :param message: The commit message. - :param violations: The validation violations. - :param warning_as_errors: If True then treat all violations as errors. - :return: The Status.ERROR if no errors or warning_as_errors. - """ - LOGGER.info('handle validation issues') - if warning_as_errors: - errors = violations - warnings = [] - else: - errors = [validation for validation in violations if validation.status == Status.ERROR] - warnings = [validation for validation in violations if validation.status == Status.WARNING] - - if errors: - LOGGER.error("%s\n\t%s", ticket['ticket'] if ticket and 'ticket' in ticket else message, - "\n\t".join(error.message for error in errors)) - - if warnings: - LOGGER.warning("%s\n\t%s", ticket['ticket'] if ticket and 'ticket' in ticket else message, - "\n\t".join(warning.message for warning in warnings)) - - return Status.ERROR if errors else Status.OK - - -def jira_client(jira_server: str, verbose: bool = False) -> Optional[Type[JIRA]]: - """ - Connect to jira. - - Create a connection to jira_server and validate that SERVER-1 is accessible. - :param jira_server: The jira server endpoint to connect and validate. - :param verbose: A flag controlling th verbosity of the checks. The requests and jira package - are very verbose by default. - :return: The jira client instance if all went well. - """ - try: - # requests and JIRA can be very verbose. - LOGGER.info('connecting to %s', jira_server) - with silence(verbose): - jira = JIRA(jira_server, logging=verbose) - # check the status of a known / existing ticket. A JIRAError with a status code of - # 404 Not Found maybe returned or a 401 unauthorized. - jira.issue("SERVER-1") - return jira - except (requests.exceptions.ConnectionError, JIRAError, ValueError) as ex: - # These are recoverable / ignorable exceptions. We print exception the full stack trace - # when debugging / verbose output is requested. - # ConnectionErrors relate to networking. - # JIRAError, ValueError refer to invalid / unexpected responses. - class_name = _get_class_name(ex) - if isinstance(ex, requests.exceptions.ConnectionError): - details = f'{class_name}: unable to connect to {jira_server}' - elif isinstance(ex, JIRAError): - details = f'{class_name}: unable to access {jira_server}, status: {ex.status_code}' - elif isinstance(ex, ValueError): - details = f'{class_name}: communication error with {jira_server}' - LOGGER.debug(details, exc_info=True) - except Exception as ex: # pylint: disable=broad-except - # recoverable / ignorable exceptions but unknown so print trace. - LOGGER.debug('%s unknown error: %s', _get_class_name(ex), jira_server, exc_info=True) - return None - - -def configure_logging(level: int, formatter: str = '%(levelname)s: %(message)s'): - """ - Configure logging. - - :param level: The log level verbosity. 0 logs warnings and above. 1 enabled info and above, all - other values are DEBUG and above. - :param formatter: The log formatter. - """ - # level 0 is the default level (warning or greater). - logging.basicConfig(format=formatter) - root_logger = logging.getLogger() - debuglevel = 0 - if level == 0: - level = logging.WARNING - elif level == 1: - level = logging.INFO - elif level >= 2: - debuglevel = 1 - level = logging.DEBUG - - root_logger.setLevel(level) - HTTPConnection.debuglevel = debuglevel - - -def _get_class_name(obj: Any) -> str: - """Get the class name without package.""" - return type(obj).__name__ - - -@contextmanager -def silence(disable: bool = False): - """ - Silence logging within this scope. - - :param disable: A flag to programmatically enable / disable the silence functionality. Useful - for debugging. - """ - logger = logging.getLogger() - old = logger.disabled - try: - if not disable: - logger.disabled = True - yield - finally: - logger.disabled = old - - -def parse_args(argv: List[str]) -> Type[argparse.ArgumentParser]: - """ - Parse the command line args. - - :param argv: The command line arguments. - :return: The parsed arguments. - """ +def main(argv=None): + """Execute Main function to validate commit messages.""" parser = argparse.ArgumentParser( usage="Validate the commit message. " "It validates the latest message when no arguments are provided.") parser.add_argument( - "-a", - '--author', - dest="author", - nargs='?', - const=1, - type=str, - help="Your jira username of the author. This value must match the JIRA assignee.", - required=True, - ) - parser.add_argument( - "-j", - dest="jira_server", - nargs='?', - const=1, - type=str, - help="The jira server location. Defaults to '" + DEFAULT_JIRA + "'", - default=DEFAULT_JIRA, - ) - parser.add_argument( - "-W", + "-i", action="store_true", - dest="warning_as_errors", - help="treat warnings as errors.", + dest="ignore_warnings", + help="Ignore all warnings.", default=False, ) - parser.add_argument("-v", "--verbosity", action="count", default=0, - help="increase output verbosity") parser.add_argument( "message", metavar="commit message", @@ -427,25 +65,24 @@ def parse_args(argv: List[str]) -> Type[argparse.ArgumentParser]: help="The commit message to validate", ) args = parser.parse_args(argv) - return args - - -def main(argv: List = None) -> Type[Status]: - """ - Execute main function to validate commit messages. - - :param argv: The command line arguments. - :return: Status.OK if the commit message validation passed other wise Status.ERROR. - """ - - args = parse_args(argv) - configure_logging(level=args.verbosity) - jira = jira_client(args.jira_server) - message = get_full_message(args.message) - ticket, violations = validate_message(message, args.author, jira) + if not args.message: + print('Validating last git commit message') + result = subprocess.check_output(GIT_SHOW_COMMAND) + message = result.decode('utf-8') + else: + message = " ".join(args.message) - return handle_violations(ticket, message, violations, args.warning_as_errors) + if any(valid_pattern.match(message) for valid_pattern in VALID_PATTERNS): + status = STATUS_OK + elif any(private_pattern.match(message) for private_pattern in PRIVATE_PATTERNS): + print("ERROR: found a reference to a private project\n{message}".format(message=message)) + status = STATUS_ERROR + else: + print("{message_type}: found a commit without a ticket\n{message}".format( + message_type="WARNING" if args.ignore_warnings else "ERROR", message=message)) + status = STATUS_OK if args.ignore_warnings else STATUS_ERROR + return status if __name__ == "__main__": |