scripts: release: list_backports.py

Created list_backports.py to examine prs applied to a backport
branch and extract associated issues. This is helpful for
adding to release notes.

The script may also be used to ensure that backported changes
also have one or more associated issues.

Signed-off-by: Christopher Friedt <[email protected]>
(cherry picked from commit 57762ca)

Author: Christopher Friedt

scripts/release/list_backports.py

@@ -0,0 +1,341 @@
+#!/usr/bin/env python3
+# Copyright (c) 2022, Meta
+#
+# SPDX-License-Identifier: Apache-2.0
+
+"""Query issues in a release branch
+
+This script searches for issues referenced via pull-requests in a release
+branch in order to simplify tracking changes such as automated backports,
+manual backports, security fixes, and stability fixes.
+
+A formatted report is printed to standard output either in JSON or
+reStructuredText.
+
+Since an issue is required for all changes to release branches, merged PRs
+must have at least one instance of the phrase "Fixes #1234" in the body. This
+script will throw an error if a PR has been made without an associated issue.
+
+Usage:
+    ./scripts/release/list_backports.py \
+        -t ~/.ghtoken \
+        -b v2.7-branch \
+        -s 2021-12-15 -e 2022-04-22 \
+        -P 45074 -P 45868 -P 44918 -P 41234 -P 41174 \
+        -j | jq . | tee /tmp/backports.json
+
+    GITHUB_TOKEN="<secret>" \
+    ./scripts/release/list_backports.py \
+        -b v3.0-branch \
+        -p 43381 \
+        -j | jq . | tee /tmp/backports.json
+"""
+
+import argparse
+from datetime import datetime, timedelta
+import io
+import json
+import logging
+import os
+import re
+import sys
+
+# Requires PyGithub
+from github import Github
+
+
+# https://gist.github.com/monkut/e60eea811ef085a6540f
+def valid_date_type(arg_date_str):
+    """custom argparse *date* type for user dates values given from the
+    command line"""
+    try:
+        return datetime.strptime(arg_date_str, "%Y-%m-%d")
+    except ValueError:
+        msg = "Given Date ({0}) not valid! Expected format, YYYY-MM-DD!".format(arg_date_str)
+        raise argparse.ArgumentTypeError(msg)
+
+
+def parse_args():
+    parser = argparse.ArgumentParser()
+    parser.add_argument('-t', '--token', dest='tokenfile',
+                        help='File containing GitHub token (alternatively, use GITHUB_TOKEN env variable)', metavar='FILE')
+    parser.add_argument('-b', '--base', dest='base',
+                        help='branch (base) for PRs (e.g. v2.7-branch)', metavar='BRANCH', required=True)
+    parser.add_argument('-j', '--json', dest='json', action='store_true',
+                        help='print output in JSON rather than RST')
+    parser.add_argument('-s', '--start', dest='start', help='start date (YYYY-mm-dd)',
+                        metavar='START_DATE', type=valid_date_type)
+    parser.add_argument('-e', '--end', dest='end', help='end date (YYYY-mm-dd)',
+                        metavar='END_DATE', type=valid_date_type)
+    parser.add_argument("-o", "--org", default="zephyrproject-rtos",
+                        help="Github organisation")
+    parser.add_argument('-p', '--include-pull', dest='includes',
+                        help='include pull request (can be specified multiple times)',
+                        metavar='PR', type=int, action='append', default=[])
+    parser.add_argument('-P', '--exclude-pull', dest='excludes',
+                        help='exlude pull request (can be specified multiple times, helpful for version bumps and release notes)',
+                        metavar='PR', type=int, action='append', default=[])
+    parser.add_argument("-r", "--repo", default="zephyr",
+                        help="Github repository")
+
+    args = parser.parse_args()
+
+    if args.includes:
+        if getattr(args, 'start'):
+            logging.error(
+                'the --start argument should not be used with --include-pull')
+            return None
+        if getattr(args, 'end'):
+            logging.error(
+                'the --end argument should not be used with --include-pull')
+            return None
+    else:
+        if not getattr(args, 'start'):
+            logging.error(
+                'if --include-pr PR is not used, --start START_DATE is required')
+            return None
+
+        if not getattr(args, 'end'):
+            setattr(args, 'end', datetime.now())
+
+        if args.end < args.start:
+            logging.error(
+                f'end date {args.end} is before start date {args.start}')
+            return None
+
+    if args.tokenfile:
+        with open(args.tokenfile, 'r') as file:
+            token = file.read()
+            token = token.strip()
+    else:
+        if 'GITHUB_TOKEN' not in os.environ:
+            raise ValueError('No credentials specified')
+        token = os.environ['GITHUB_TOKEN']
+
+    setattr(args, 'token', token)
+
+    return args
+
+
+class Backport(object):
+    def __init__(self, repo, base, pulls):
+        self._base = base
+        self._repo = repo
+        self._issues = []
+        self._pulls = pulls
+
+        self._pulls_without_an_issue = []
+        self._pulls_with_invalid_issues = {}
+
+    @staticmethod
+    def by_date_range(repo, base, start_date, end_date, excludes):
+        """Create a Backport object with the provided repo,
+        base, start datetime object, and end datetime objects, and
+        list of excluded PRs"""
+
+        pulls = []
+
+        unfiltered_pulls = repo.get_pulls(
+            base=base, state='closed')
+        for p in unfiltered_pulls:
+            if not p.merged:
+                # only consider merged backports
+                continue
+
+            if p.closed_at < start_date or p.closed_at >= end_date + timedelta(1):
+                # only concerned with PRs within time window
+                continue
+
+            if p.number in excludes:
+                # skip PRs that have been explicitly excluded
+                continue
+
+            pulls.append(p)
+
+        # paginated_list.sort() does not exist
+        pulls = sorted(pulls, key=lambda x: x.number)
+
+        return Backport(repo, base, pulls)
+
+    @staticmethod
+    def by_included_prs(repo, base, includes):
+        """Create a Backport object with the provided repo,
+        base, and list of included PRs"""
+
+        pulls = []
+
+        for i in includes:
+            try:
+                p = repo.get_pull(i)
+            except Exception:
+                p = None
+
+            if not p:
+                logging.error(f'{i} is not a valid pull request')
+                return None
+
+            if p.base.ref != base:
+                logging.error(
+                    f'{i} is not a valid pull request for base {base} ({p.base.label})')
+                return None
+
+            pulls.append(p)
+
+        # paginated_list.sort() does not exist
+        pulls = sorted(pulls, key=lambda x: x.number)
+
+        return Backport(repo, base, pulls)
+
+    @staticmethod
+    def sanitize_title(title):
+        # TODO: sanitize titles such that they are suitable for both JSON and ReStructured Text
+        # could also automatically fix titles like "Automated backport of PR #1234"
+        return title
+
+    def print(self):
+        for i in self.get_issues():
+            title = Backport.sanitize_title(i.title)
+            # * :github:`38972` - logging: Cleaning references to tracing in logging
+            print(f'* :github:`{i.number}` - {title}')
+
+    def print_json(self):
+        issue_objects = []
+        for i in self.get_issues():
+            obj = {}
+            obj['id'] = i.number
+            obj['title'] = Backport.sanitize_title(i.title)
+            obj['url'] = f'https://github.com/{self._repo.organization.login}/{self._repo.name}/pull/{i.number}'
+            issue_objects.append(obj)
+
+        print(json.dumps(issue_objects))
+
+    def get_pulls(self):
+        return self._pulls
+
+    def get_issues(self):
+        """Return GitHub issues fixed in the provided date window"""
+        if self._issues:
+            return self._issues
+
+        issue_map = {}
+        self._pulls_without_an_issue = []
+        self._pulls_with_invalid_issues = {}
+
+        for p in self._pulls:
+            # check for issues in this pr
+            issues_for_this_pr = {}
+            with io.StringIO(p.body) as buf:
+                for line in buf.readlines():
+                    line = line.strip()
+                    match = re.search(r"^Fixes[:]?\s*#([1-9][0-9]*).*", line)
+                    if not match:
+                        match = re.search(
+                            rf"^Fixes[:]?\s*https://github\.com/{self._repo.organization.login}/{self._repo.name}/issues/([1-9][0-9]*).*", line)
+                    if not match:
+                        continue
+                    issue_number = int(match[1])
+                    issue = self._repo.get_issue(issue_number)
+                    if not issue:
+                        if not self._pulls_with_invalid_issues[p.number]:
+                            self._pulls_with_invalid_issues[p.number] = [
+                                issue_number]
+                        else:
+                            self._pulls_with_invalid_issues[p.number].append(
+                                issue_number)
+                        logging.error(
+                            f'https://github.com/{self._repo.organization.login}/{self._repo.name}/pull/{p.number} references invalid issue number {issue_number}')
+                        continue
+                    issues_for_this_pr[issue_number] = issue
+
+            # report prs missing issues later
+            if len(issues_for_this_pr) == 0:
+                logging.error(
+                    f'https://github.com/{self._repo.organization.login}/{self._repo.name}/pull/{p.number} does not have an associated issue')
+                self._pulls_without_an_issue.append(p)
+                continue
+
+            # FIXME: when we have upgrade to python3.9+, use "issue_map | issues_for_this_pr"
+            issue_map = {**issue_map, **issues_for_this_pr}
+
+        issues = list(issue_map.values())
+
+        # paginated_list.sort() does not exist
+        issues = sorted(issues, key=lambda x: x.number)
+
+        self._issues = issues
+
+        return self._issues
+
+    def get_pulls_without_issues(self):
+        if self._pulls_without_an_issue:
+            return self._pulls_without_an_issue
+
+        self.get_issues()
+
+        return self._pulls_without_an_issue
+
+    def get_pulls_with_invalid_issues(self):
+        if self._pulls_with_invalid_issues:
+            return self._pulls_with_invalid_issues
+
+        self.get_issues()
+
+        return self._pulls_with_invalid_issues
+
+
+def main():
+    args = parse_args()
+
+    if not args:
+        return os.EX_DATAERR
+
+    try:
+        gh = Github(args.token)
+    except Exception:
+        logging.error('failed to authenticate with GitHub')
+        return os.EX_DATAERR
+
+    try:
+        repo = gh.get_repo(args.org + '/' + args.repo)
+    except Exception:
+        logging.error('failed to obtain Github repository')
+        return os.EX_DATAERR
+
+    bp = None
+    if args.includes:
+        bp = Backport.by_included_prs(repo, args.base, set(args.includes))
+    else:
+        bp = Backport.by_date_range(repo, args.base,
+                                    args.start, args.end, set(args.excludes))
+
+    if not bp:
+        return os.EX_DATAERR
+
+    pulls_with_invalid_issues = bp.get_pulls_with_invalid_issues()
+    if pulls_with_invalid_issues:
+        logging.error('The following PRs link to invalid issues:')
+        for (p, lst) in pulls_with_invalid_issues:
+            logging.error(
+                f'\nhttps://github.com/{repo.organization.login}/{repo.name}/pull/{p.number}: {lst}')
+        return os.EX_DATAERR
+
+    pulls_without_issues = bp.get_pulls_without_issues()
+    if pulls_without_issues:
+        logging.error(
+            'Please ensure the body of each PR to a release branch contains "Fixes #1234"')
+        logging.error('The following PRs are lacking associated issues:')
+        for p in pulls_without_issues:
+            logging.error(
+                f'https://github.com/{repo.organization.login}/{repo.name}/pull/{p.number}')
+        return os.EX_DATAERR
+
+    if args.json:
+        bp.print_json()
+    else:
+        bp.print()
+
+    return os.EX_OK
+
+
+if __name__ == '__main__':
+    sys.exit(main())