def merge_pr()

in zk-merge-pr.py [0:0]

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

def merge_pr(pr_num, title, pr_repo_desc):

    merge_message = []
    result = input("Would you like to squash the commit messages? (y/n): ")
    if result.lower().strip() == "y":
        # Retrieve the commits separately.
        json_commits = get_json(f"https://api.github.com/repos/{PUSH_REMOTE_NAME}/{PROJECT_NAME}/pulls/{pr_num}/commits")
        if json_commits and isinstance(json_commits, list):
            for commit in json_commits:
                commit_message = commit['commit']['message']
                # Remove empty lines and lines containing "Change-Id:"
                filtered_lines = [line for line in commit_message.split('\n') if 'Change-Id:' not in line and line.strip()]
                modified_commit_message = '\n'.join(filtered_lines)
                if modified_commit_message.strip() != title.strip():
                    merge_message += [modified_commit_message]

    # Check for disapproval reviews.
    json_reviewers = get_json(f"https://api.github.com/repos/{PUSH_REMOTE_NAME}/{PROJECT_NAME}/pulls/{pr_num}/reviews")
    disapproval_reviews = [review['user']['login'] for review in json_reviewers if review['state'] == 'CHANGES_REQUESTED']
    if disapproval_reviews:
        continue_maybe("Warning: There are requested changes. Proceed with merging pull request #%s?" % pr_num)
    # Verify if there are no approved reviews.
    approved_reviewers = [review['user']['login'] for review in json_reviewers if review['state'] == 'APPROVED']
    if not approved_reviewers:
        continue_maybe("Warning: Pull Request does not have an approved review. Proceed with merging pull request #%s?" % pr_num)
    else:
        reviewers_string = ', '.join(approved_reviewers)
        merge_message += [f"Reviewers: {reviewers_string}"]
    # Check the author and the closing line.
    json_pr = get_json(f"https://api.github.com/repos/{PUSH_REMOTE_NAME}/{PROJECT_NAME}/pulls/{pr_num}")
    primary_author = json_pr["user"]["login"]
    if primary_author != "":
        merge_message += [f"Author: {primary_author}"]
    close_line = "Closes #%s from %s" % (pr_num, pr_repo_desc)
    merge_message += [close_line]
    merged_string = '\n'.join(merge_message)

    # Get the latest commit SHA.
    latest_commit_sha = json_pr["head"]["sha"]
    json_status = get_json(f"https://api.github.com/repos/{PUSH_REMOTE_NAME}/{PROJECT_NAME}/commits/{latest_commit_sha}/check-runs")
    # Check if all checks have passed on GitHub.
    all_checks_passed = all(status["conclusion"] == "success" for status in json_status["check_runs"])
    if all_checks_passed:
        print("All checks have passed on the github.")
    else:
        any_in_progress = any(run["status"] == "in_progress" for run in json_status["check_runs"])
        if any_in_progress:
            continue_maybe("Warning: There are pending checks. Would you like to continue the merge?")
        else:
            continue_maybe("Warning: Not all checks have passed on GitHub. Would you like to continue the merge?")

    headers = {
        "Authorization": f"token {GITHUB_OAUTH_KEY}",
        "Accept": "application/vnd.github.v3+json"
    }
    data = {
        "commit_title": title,
        "commit_message": merged_string,
        "merge_method": "squash"
    }

    response = requests.put(f"https://api.github.com/repos/{PUSH_REMOTE_NAME}/{PROJECT_NAME}/pulls/{pr_num}/merge", headers=headers, json=data)

    if response.status_code == 200:
        merge_response_json = response.json()
        merge_commit_sha = merge_response_json.get("sha")
        print(f"Pull request #{pr_num} merged. Sha: #{merge_commit_sha}")
        return merge_commit_sha
    else:
        print(f"Failed to merge pull request #{pr_num}. Status code: {response.status_code}")
        print(response.text)
        exit()