Created
August 6, 2020 10:47
-
-
Save ov7a/a70e821c01ad491e33ca3c273c5d926c to your computer and use it in GitHub Desktop.
Print related jira issue status for each branch in git repo to find branches which can be closed
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
import argparse | |
import subprocess | |
from jira import JIRA | |
from getpass import getpass | |
import requests | |
EXCLUDE_BRANCHES = {"master", "develop"} | |
COMMON_BRANCH_PREFIX = "feature/" | |
COMMON_BRANCH_SUFFIX = "-epic" | |
HEADS_PREFIX = "refs/heads/" | |
def read_args(): | |
parser = argparse.ArgumentParser() | |
parser.add_argument('--repoUrl', required=True) | |
parser.add_argument('--jiraHost', required=True) | |
return parser.parse_args() | |
def strip_prefix(str, prefix): | |
if str.startswith(prefix): | |
return str[len(prefix):] | |
else: | |
return str | |
def strip_suffix(str, suffix): | |
if suffix and str.endswith(suffix): | |
return str[:-len(suffix)] | |
else: | |
return str | |
def get_active_branches(repo_url): | |
result = subprocess.run(['git', 'ls-remote', '--heads', repo_url], stdout=subprocess.PIPE) | |
branches = map( | |
lambda x: strip_prefix(x.partition('\t')[-1], HEADS_PREFIX), | |
filter(None, result.stdout.decode('utf-8').split('\n')) | |
) | |
return list(filter(lambda x: x not in EXCLUDE_BRANCHES, branches)) | |
def branch_to_issue_id(branch): | |
branch = strip_prefix(branch, COMMON_BRANCH_PREFIX) | |
branch = strip_suffix(branch, COMMON_BRANCH_SUFFIX) | |
return branch | |
def get_issue_statuses(host, issues_ids): | |
username = input("Input JIRA login: ") | |
password = getpass() | |
jira = JIRA(host, auth=(username, password)) | |
jira_query = 'issuekey in ("{}")'.format('","'.join(issues_ids)) | |
issues = jira.search_issues(jira_query, validate_query=False, maxResults=len(issues_ids)) | |
result = dict() | |
for issue in issues: | |
result[issue.key] = (issue.fields.status.name, issue.fields.status.statusCategory.key) | |
return result | |
def process(repo_url, jira_host): | |
branches = get_active_branches(repo_url) | |
issue_ids = list(map(branch_to_issue_id, branches)) | |
issue_statuses = get_issue_statuses(jira_host, issue_ids) | |
for branch in sorted(branches): | |
issue_status = issue_statuses.get(branch_to_issue_id(branch)) | |
if issue_status is None: | |
print(branch, "unknown", sep="\t") | |
else: | |
status_name, status_category = issue_status | |
print(branch, status_name, status_category, sep="\t") | |
if __name__ == "__main__": | |
args = read_args() | |
requests.packages.urllib3.disable_warnings() | |
process(args.repoUrl, args.jiraHost) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment