tools/release/release-verify.py

#!/usr/bin/env python3
import subprocess
import argparse
import os
import yaml
import requests
import logging
from requests.auth import HTTPBasicAuth
import re

# Directories where the library submodules exist.
CSDK_LIBRARY_DIRS = ["libraries/aws", "libraries/standard"]

# CSDK organization and repo constants
CSDK_ORG = "aws"
CSDK_REPO = "aws-iot-device-sdk-embedded-c"

# Github API global. The Github API is used instead of pyGithub because some
# checks are not available yet in the package.
GITHUB_API_URL = "https://api.github.com"
GITHUB_ACCESS_TOKEN = ""
GITHUB_AUTH_HEADER = {"Authorization": "token {}", "Accept": "application/vnd.github.v3+json"}

# Jenkins API globals
JENKINS_API_URL = ""
JENKINS_USERNAME = ""
JENKINS_PASSWORD = ""
JENKINS_CSDK_DEMOS_PATH = "job/csdk/job/demo_pipeline/lastCompletedBuild"
JENKINS_CSDK_TESTS_PATH = "job/csdk/job/nightly/lastCompletedBuild"
JENKINS_API_PATH = "api/json"
JENKINS_SERVER_VERIFY = True
GITHUB_RELEASES_TAGS_VERIFY = True

# Errors found in this run.
errors = 0

def log_error(error_log):
    """
    Logs an error to error.log.
    Args:
        error_log (str): Error string to log.
    """
    global errors
    logging.error(error_log)
    errors = errors + 1


def validate_manifest(csdk_root, csdk_version, lib_versions):
    """
    Validates the manifest.yml file at the root of the CSDK.
    Args:
        csdk_root (str): The root path to the CSDK repo.
        csdk_versions (str): The new version of the CSDK repo.
        lib_versions (dict): A dictionary containing the new versions of each library.
        Please see tools/release/config.yml.
    """
    with open(os.path.join(csdk_root, "manifest.yml")) as manifest_file:
        manifest = yaml.safe_load(manifest_file, Loader=yaml.FullLoader)

    # Verify the CSDK version is correct.
    manifest_version = manifest["version"]
    if manifest_version != csdk_version:
        log_error(f"Invalid manifest.yml. CSDK version {manifest_version} should be {csdk_version}.")

    # Verify that all libraries in this branch are also in the manifest.yml.
    for library_dir in CSDK_LIBRARY_DIRS:
        libraries = os.listdir(os.path.join(csdk_root, library_dir))
        for library in libraries:
            library = library.lower()
            found = filter(lambda dep: dep["name"].casefold() == library, manifest["dependencies"])
            found = list(found)
            if len(found) != 1:
                log_error(f"Invalid manifest.yml. Found {len(found)} occurrences of required library {library}.")
            else:
                dep_version = found[0]["version"]
                dep_name = found[0]["name"]
                if dep_version.lower() != lib_versions[library]:
                    log_error(f"Invalid manifest.yml. Invalid version {dep_version} for {dep_name}.")


def validate_checks(configs) -> list:
    """
    Validates that all of the GHA and CBMC status checks passed on all repos.
    Returns a list of existing org/repo paths found.
    """
    docs_file = open("docs_to_review.txt", "w+")
    repo_paths = []
    for library_dir in CSDK_LIBRARY_DIRS:
        # Get the submodules in the library directory.
        git_resp = requests.get(
            f"{GITHUB_API_URL}/repos/{CSDK_ORG}/{CSDK_REPO}/contents/{library_dir}?ref=main",
            headers=GITHUB_AUTH_HEADER,
        )
        # A 404 status code means the branch doesn't exist.
        if git_resp.status_code == 404:
            log_error(
                "The main branch does not exist in the CSDK. Please recheck the branches in the repo."
            )
            break
        else:
            # For each library submodule in this directory, get the status checks results and docs to review.
            for library in git_resp.json():
                library_name = library["name"]
                # Get the commit SHA of the branch currently in main.
                commit_sha = library["sha"]
                # Get the organization of this repo
                html_url = library["html_url"]
                repo_path = re.search("(?<=\.com/)(.*)(?=/tree)", html_url).group(0)
                repo_paths.append(repo_path)

                # Verify CBMC checks for the libraries not excluded.
                if library_name.lower() not in configs["libraries-to-disable-cbmc-checks"]:
                    # Get the status of the CBMC checks
                    git_resp = requests.get(
                        f"{GITHUB_API_URL}/repos/{repo_path}/commits/{commit_sha}/status", headers=GITHUB_AUTH_HEADER
                    )
                    if git_resp.json()["state"] != "success":
                        log_error(f"The CBMC status checks failed for {html_url}.")

                # Get the status of the GHA checks
                git_resp = requests.get(
                    f"{GITHUB_API_URL}/repos/{repo_path}/commits/{commit_sha}/check-runs", headers=GITHUB_AUTH_HEADER
                )
                for check_run in git_resp.json()["check_runs"]:
                    if check_run["conclusion"] != "success":
                        check_run_name = check_run["name"]
                        log_error(f"The GHA {check_run_name} check failed for {html_url}.")

                # Collect the HTML URLS for reviewing the docs.
                html_url = html_url.split(commit_sha)[0] + "main"
                docs_file.write(f"{library_name}\n")
                docs_file.write(f"{html_url}/README.md\n")
                docs_file.write(f"{html_url}/CHANGELOG.md\n\n")

    docs_file.close()
    repo_paths.append(f"{CSDK_ORG}/{CSDK_REPO}")
    return repo_paths

def validate_ci():
    """
    Validates that all CSDK jobs in the Jenkins CI passed.
    """
    jenkins_resp = requests.get(
        f"{JENKINS_API_URL}/{JENKINS_CSDK_DEMOS_PATH}/{JENKINS_API_PATH}",
        auth=HTTPBasicAuth(JENKINS_USERNAME, JENKINS_PASSWORD),
        verify=JENKINS_SERVER_VERIFY,
    )
    if jenkins_resp.json()["result"] != "SUCCESS":
        log_error(f"Jenkins job failed: {JENKINS_API_URL}/{JENKINS_CSDK_DEMOS_PATH}.")
    jenkins_resp = requests.get(
        f"{JENKINS_API_URL}/{JENKINS_CSDK_TESTS_PATH}/{JENKINS_API_PATH}",
        auth=HTTPBasicAuth(JENKINS_USERNAME, JENKINS_PASSWORD),
        verify=JENKINS_SERVER_VERIFY,
    )
    if jenkins_resp.json()["result"] != "SUCCESS":
        log_error(f"Jenkins job failed: {JENKINS_API_URL}/{JENKINS_CSDK_TESTS_PATH}.")


def validate_branches(repo_paths):
    """
    Validates that only the main branch exists on each library repo.
    Args:
        repo_paths (dict): Paths to all library repos in the CSDK, including their org.
    """
    for repo_path in repo_paths:
        git_resp = requests.get(f"{GITHUB_API_URL}/repos/{repo_path}/branches", headers=GITHUB_AUTH_HEADER)
        valid_branches = ["main", "gh-pages", "dev"]
        if repo_path == f"{CSDK_ORG}/{CSDK_REPO}":
            valid_branches += ["v4_beta_deprecated"]
        for branch in git_resp.json():
            branch_name = branch["name"]
            if branch_name not in valid_branches:
                log_error(f"Invalid branch {branch_name} found in {repo_path}.")

def validate_tags_and_releases(repo_paths, lib_versions):
    """
    Validates that each library repo has a release and tag for the specified version number.
    Args:
        repo_paths (dict): Paths to all library repos in the CSDK, including their org.
        lib_versions (dict): A dictionary containing the new versions of each library.
    """
    if not GITHUB_RELEASES_TAGS_VERIFY:
        return
    # Verify that all repos have a release and tag for the version specified in the config.
    for repo_path in repo_paths:
        library = repo_path.split("/")[-1].casefold()
        # Only need to verify for spoke repos because CSDK is the library being released.
        if library == CSDK_REPO:
            continue
        # Check that a release exists with the same name as the version.
        git_releases_resp = requests.get(f"{GITHUB_API_URL}/repos/{repo_path}/releases", headers=GITHUB_AUTH_HEADER)
        found_release_for_version = False
        for release in git_releases_resp.json():
            if release["name"] == lib_versions[library] and release["tag_name"] == lib_versions[library]:
                found_release_for_version = True
                break
        if not found_release_for_version:
            log_error(f"Could not find release {lib_versions[library]} for {repo_path}.")
        # Check that a tag exists with the same name as the version.
        git_tags_resp = requests.get(f"{GITHUB_API_URL}/repos/{repo_path}/tags", headers=GITHUB_AUTH_HEADER)
        found_tag_for_version = False
        for tag in git_tags_resp.json():
            if tag["name"] == lib_versions[library]:
                found_tag_for_version = True
                break
        if not found_tag_for_version:
            log_error(f"Could not find tag {lib_versions[library]} for {repo_path}.")

def validate_main_branch():
    """
    Verifies there are no pending PRs to the main branch.
    """
    git_resp = requests.get(
        f"{GITHUB_API_URL}/repos/{CSDK_ORG}/{CSDK_REPO}/pulls?base=main", headers=GITHUB_AUTH_HEADER
    )
    if len(git_resp.json()) == 0:
        logging.warning("main branch does not exist in CSDK.")
    for pr in git_resp.json():
        if not pr["draft"]:
            pr_url = pr["url"]
            log_error(f"Pull request to main {pr_url}.")

def set_globals(configs):
    """
    Set global variables used in this script.
    Args:
        configs (dict): User configurations for this script.
    """
    global GITHUB_ACCESS_TOKEN
    global GITHUB_AUTH_HEADER
    global GITHUB_RELEASES_TAGS_VERIFY
    global JENKINS_API_URL
    global JENKINS_USERNAME
    global JENKINS_PASSWORD
    global JENKINS_SERVER_VERIFY

    access_token = os.environ.get("GITHUB_ACCESS_TOKEN")
    if access_token == None:
        access_token = configs["github_access_token"]
    if access_token == None:
        raise Exception(
            "Please define GITHUB_ACCESS_TOKEN in your system's environment variables or pass argument --github-access-token to this script."
        )
    jenkins_username = os.environ.get("JENKINS_USERNAME")
    if jenkins_username == None:
        jenkins_username = configs["jenkins_username"]
    if jenkins_username == None:
        raise Exception(
            "Please define JENKINS_USERNAME in your system's environment variables or pass argument --jenkins-username to this script."
        )
    jenkins_password = os.environ.get("JENKINS_PASSWORD")
    if jenkins_password == None:
        jenkins_password = configs["jenkins_password"]
    if jenkins_password == None:
        raise Exception(
            "Please define JENKINS_PASSWORD in your system's environment variables or pass argument --jenkins-password to this script."
        )
    jenkins_api_url = os.environ.get("JENKINS_API_URL")
    if jenkins_api_url == None:
        jenkins_password = configs["jenkins_api_url"]
    if jenkins_password == None:
        raise Exception(
            "Please define JENKINS_API_URL in your system's environment variables or pass argument --jenkins_api_url to this script."
        )
    GITHUB_ACCESS_TOKEN = access_token
    GITHUB_AUTH_HEADER["Authorization"] = GITHUB_AUTH_HEADER["Authorization"].format(GITHUB_ACCESS_TOKEN)
    JENKINS_USERNAME = jenkins_username
    JENKINS_PASSWORD = jenkins_password
    JENKINS_API_URL = jenkins_api_url
    JENKINS_SERVER_VERIFY = False if configs["disable_jenkins_server_verify"] else True
    GITHUB_RELEASES_TAGS_VERIFY = False if configs["disable_github_release_tag_verify"] else True


def get_configs() -> dict:
    """
    Parse the input user arguments and return a dictionary of arguments.
    """
    # Parse the input arguments to this script.
    parser = argparse.ArgumentParser(description="Perform CSDK Release activities.")
    parser.add_argument("-r", "--root", action="store", required=True, dest="root", help="CSDK repo root path.")
    parser.add_argument(
        "--github-access-token",
        action="store",
        required=False,
        dest="github_access_token",
        help="Github API developer access token.",
    )
    parser.add_argument(
        "--jenkins-api-url", action="store", required=False, dest="jenkins_api_url", help="Jenkins CI API url."
    )
    parser.add_argument(
        "--jenkins-username", action="store", required=False, dest="jenkins_username", help="Jenkins CI username."
    )
    parser.add_argument(
        "--jenkins-password", action="store", required=False, dest="jenkins_password", help="Jenkins CI password."
    )
    parser.add_argument(
        "--csdk-version", action="store", required=True, dest="csdk_version", help="CSDK version to be released."
    )
    parser.add_argument(
        "--disable-jenkins-server-verify",
        action="store_true",
        required=False,
        default=False,
        dest="disable_jenkins_server_verify",
        help="Disable server verification for the Jenkins API calls if your system doesn't have the certificate in its store.",
    )
    parser.add_argument(
        "--disable-github-release-tag-verify",
        action="store_true",
        required=False,
        default=False,
        dest="disable_github_release_tag_verify",
        help="Disable verifying that there are Github releases and tags named after the versions of the spoke repos.",
    )
    parser.add_argument(
        "--disable-cbmc-checks-for",
        action="append",
        required=False,
        default=[],
        dest="libraries-to-disable-cbmc-checks",
        type=str.lower,
        help="Disable CBMC checks for libraries, in case it is not available.",
    )

    args, unknown = parser.parse_known_args()
    csdk_root = os.path.abspath(args.root)

    # For each of the available libraries in the current branch require a version argument.
    for library_dir in CSDK_LIBRARY_DIRS:
        libraries = os.listdir(os.path.join(csdk_root, library_dir))
        for library in libraries:
            parser.add_argument(
                f"--{library.casefold()}-version",
                action="store",
                required=True,
                dest=f"{library.casefold()}",
                type=str.lower,
                help=f"{library} library version.",
            )
    args = parser.parse_args()
    configs = vars(args)
    return (configs, csdk_root)


def main():
    """
    Performs pre-release validation of the CSDK and the library spoke repos.
    """
    # Parse the input arguments and args for configurations and the  CSDK root.
    (configs, csdk_root) = get_configs()

    # Set the authentication variables.
    set_globals(configs)

    # Create error.log to write errors to.
    logging.basicConfig(filename="errors.log", filemode="w", level=logging.ERROR)

    # Verify that Manifest.yml has all libraries and their versions.
    validate_manifest(csdk_root, configs["csdk_version"], configs)

    # Verify status checks in all repos.
    repo_paths = validate_checks(configs)

    # Validate that the Jenkins CI passed.
    validate_ci()

    # Check that only qualified branches exist in each library repo.
    validate_branches(repo_paths)

    # Check that library repos have a tag and release for each specified version.
    validate_tags_and_releases(repo_paths, configs)

    # Verify there are no pending PRs to the CSDK main branch.
    validate_main_branch()

    if errors > 0:
        print("Release verification failed, please see errors.log")
    else:
        print("Release verification passed.")


if __name__ == "__main__":
    main()