Skip to content

Instantly share code, notes, and snippets.

@rgpower
Created March 1, 2021 01:23
Show Gist options
  • Save rgpower/21481ce28f2f66858dee439d3e2ce4cf to your computer and use it in GitHub Desktop.
Save rgpower/21481ce28f2f66858dee439d3e2ce4cf to your computer and use it in GitHub Desktop.
You can use this tool to generate the list of repos, branches, and commits associated with a given Jira Project Release.
#! /usr/bin/env python3
import sys
import argparse
import json
import netrc
import textwrap
import urllib.parse
import urllib.request
import urllib.error
from base64 import b64encode
netrz = netrc.netrc()
jira_creds = None
jira_host = None
# bb_host = None
# bb_creds = None
def main():
global jira_creds, jira_host
args = parse_args()
jira_host = args.jira
jira_creds = netrz.authenticators(args.jira)
if jira_creds is None:
raise LookupError(f'could not locate {args.jira} credentials from your $HOME/.netrc file for {args.jira}')
# bb_host = args.b
# bb_creds = netrz.authenticators(args.b)
# if bb_creds is None:
# raise LookupError(f'could not locate {args.b} credentials from your $HOME/.netrc file for {args.b}')
try:
repo_info = fetch_issues_for_release(args.p, args.jira_release[0])
if 'json' == args.output:
print(json.dumps(repo_info, indent=True))
else:
for repo_name, issues in repo_info.items():
print(f'# {repo_name}')
print(f'git fetch')
for issue_key, v in issues.items():
short_summary = textwrap.shorten(v["summary"], width=72)
print(f'# {issue_key} {short_summary}')
for branch_name in v['branches']:
print(f'git merge origin/{branch_name} # {issue_key}')
for commit_id in v['commits']:
print(
f"git merge-base --is-ancestor {commit_id} HEAD || echo 'error for {commit_id}@{repo_name} of {issue_key}'")
except urllib.error.HTTPError as error:
if 400 == error.code:
print("Bad Request, check Release Version, or Project Key")
else:
raise error
def fetch_issues_for_release(project, release):
repos = {}
def issue_cb(results, response):
issue_count = len(results["issues"])
for issue in results["issues"]:
issue_id = issue["id"]
issue_key = issue["key"]
deets_paload = get_dev_details_payload(issue_id)
jh = {'Content-Type': 'application/json'}
dev_details = jira_request("/jsw/graphql?operation=DevDetailsDialog", data=deets_paload, headers=jh).json()
for it in dev_details['data']['developmentInformation']['details']['instanceTypes']:
for repo in it['repository']:
repo_name = repo['name']
if repo_name not in repos:
repos[repo_name] = {}
if issue_key not in repos[repo_name]:
repos[repo_name][issue_key] = {
'summary': issue['fields']['summary'],
'branches': [], 'commits': []
}
for branch in repo['branches']:
branch_name = branch['name']
repos[repo_name][issue_key]['branches'].append(branch_name)
for branch in repo['branches']:
branch_name = branch['name']
for commit in repo['commits']:
commit_id = commit['id']
repos[repo_name][issue_key]['commits'].append(commit_id)
return issue_count
jql = f'project = "{project}" AND development[commits].all > 0 and fixVersion = "{release}"'
paged_jira_request(issue_cb, '/rest/api/3/search', json={'maxResults': 10, 'jql': jql})
return repos
def parse_args():
help_text = 'Confirm a set of repo branches against a Jira release'
parser = argparse.ArgumentParser(description=help_text)
parser.add_argument('jira_release', nargs=1, help='jira release')
parser.add_argument('-p', help='Jira Project Key', required=True)
parser.add_argument('--jira', nargs='?', help='Jira endpoint, e.g. example.atlassian.net',
default='example.atlassian.net')
parser.add_argument('-b', nargs='?', help='Bitbucket endpoint, e.g. api.bitbucket.org',
default='api.bitbucket.org')
parser.add_argument('--output', help='output', choices=['json', 'shell'], default='shell')
return parser.parse_args()
def basic_auth(creds):
return 'Basic ' + b64encode((creds[0] + ':' + creds[2]).encode('utf-8')).decode('utf-8')
def request(method, uri, data=None, json=None, params=None, headers=None, auth=None):
bindata = None
url = f'{uri}'
hdrs = {**headers} if headers else {}
if 'get' == method or 'delete' == method:
if params:
url += '?' + urllib.parse.urlencode(params)
else:
bindata = None
if json:
if 'Content-Type' not in hdrs:
hdrs['Content-Type'] = 'application/json'
bindata = globals()['json'].dumps(json).encode('utf-8')
elif data:
bindata = data if type(data) == bytes else data.encode('utf-8')
else:
raise Exception("data or json param required")
if auth:
creds = (auth[0], None, auth[1])
hdrs['Authorization'] = basic_auth(creds)
req = urllib.request.Request(url, data=bindata, headers=hdrs, method=method.upper())
with urllib.request.urlopen(req) as resp:
return JsonResponse(resp)
class JsonResponse:
def __init__(self, response):
self.data = response.read().decode('utf-8')
self.status_code = response.status
self.headers = response.headers
self.url = response.url
def json(self):
return json.loads(self.data)
def jira_request(path, data=None, json=None, params=None, method='post', **kwargs):
global jira_creds, jira_host
jira_up = (jira_creds[0], jira_creds[2])
response = request(
method, f"https://{jira_host}{path}", data=data, json=json, **kwargs, params=params, auth=jira_up)
if response.status_code >= 400:
raise Exception(f"Jira responded with status code: {response.status_code}")
return response
def paged_jira_request(cb, path, json=None, params=None, method='post', **kwargs):
global jira_creds, jira_host
jira_up = (jira_creds[0], jira_creds[2])
start_at = 0
done = False
while not done:
if json:
json["startAt"] = start_at
else:
if params:
params['startAt'] = start_at
if 'maxResults' not in params:
params['maxResults'] = 100
else:
params = {'startAt': start_at, 'maxResults': 100}
response = request(
method, f"https://{jira_host}{path}", json=json, **kwargs, params=params, auth=jira_up)
if response.status_code >= 400:
raise Exception(f"Jira responded with status code: {response.status_code}")
results = response.json()
start_at += cb(results, response)
if start_at >= results['total']:
done = True
# HACK
# got this from browser Dev Tools, Bitbucket uses GraphQL to load bitbucket info related to jira issue
# seems like it might eventually become a public api since it is not yet marked with internal path
# I copied at pasted the GraphQL payload as is, and mutate it just enough, to change the issue_id
def get_dev_details_payload(issue_id):
data = '{"operationName":"DevDetailsDialog","query":"\\n query DevDetailsDialog ($issueId: ID\u0021) {\\n developmentInformation(issueId: $issueId){\\n \\n details {\\n instanceTypes {\\n id\\n name\\n type\\n typeName\\n isSingleInstance\\n baseUrl\\n devStatusErrorMessages\\n repository {\\n name\\n avatarUrl\\n description\\n url\\n parent {\\n name\\n url\\n }\\n branches {\\n name\\n url\\n createReviewUrl\\n createPullRequestUrl\\n lastCommit {\\n url\\n displayId\\n timestamp\\n }\\n pullRequests {\\n name\\n url\\n status\\n lastUpdate\\n }\\n reviews {\\n state\\n url\\n id\\n }\\n }\\n commits{\\n id\\n displayId\\n url\\n createReviewUrl\\n timestamp\\n isMerge\\n message\\n author {\\n name\\n avatarUrl\\n }\\n files{\\n linesAdded\\n linesRemoved\\n changeType\\n url\\n path\\n }\\n reviews{\\n id\\n url\\n state\\n }\\n }\\n pullRequests {\\n id\\n url\\n name\\n branchName\\n branchUrl\\n lastUpdate\\n status\\n author {\\n name\\n avatarUrl\\n }\\n reviewers{\\n name\\n avatarUrl\\n isApproved\\n }\\n }\\n }\\n danglingPullRequests {\\n id\\n url\\n name\\n branchName\\n branchUrl\\n lastUpdate\\n status\\n author {\\n name\\n avatarUrl\\n }\\n reviewers{\\n name\\n avatarUrl\\n isApproved\\n }\\n }\\n buildProviders {\\n id\\n name\\n url\\n description\\n avatarUrl\\n builds {\\n id\\n buildNumber\\n name\\n description\\n url\\n state\\n testSummary {\\n totalNumber\\n numberPassed\\n numberFailed\\n numberSkipped\\n }\\n lastUpdated\\n references {\\n name\\n uri\\n }\\n }\\n }\\n }\\n deploymentProviders {\\n id\\n name\\n homeUrl\\n logoUrl\\n deployments {\\n displayName\\n url\\n state\\n lastUpdated\\n pipelineId\\n pipelineDisplayName\\n pipelineUrl\\n environment {\\n id\\n type\\n displayName\\n }\\n }\\n }\\n featureFlagProviders {\\n id\\n createFlagTemplateUrl\\n linkFlagTemplateUrl\\n featureFlags {\\n id\\n key\\n displayName\\n providerId\\n details{\\n url\\n lastUpdated\\n environment{\\n name\\n type\\n }\\n status{\\n enabled\\n defaultValue\\n rollout{\\n percentage\\n text\\n rules\\n }\\n }\\n }\\n }\\n}\\n remoteLinksByType {\\n providers {\\n id\\n name\\n homeUrl\\n logoUrl\\n documentationUrl\\n actions {\\n id\\n label {\\n value\\n }\\n templateUrl\\n }\\n }\\n types {\\n type\\n remoteLinks {\\n id\\n providerId\\n displayName\\n url\\n type\\n description\\n status {\\n appearance\\n label\\n }\\n actionIds\\n attributeMap {\\n key\\n value\\n }\\n }\\n }\\n }\\n \\n embeddedMarketplace {\\n shouldDisplayForBuilds,\\n shouldDisplayForDeployments,\\n shouldDisplayForFeatureFlags\\n }\\n\\n }\\n\\n }\\n }","variables":{"issueId":"' + issue_id + '"}}'
return data
# def bb_request(path, data=None, json=None, params=None, method='get', **kwargs):
# global bb_creds, bb_host
# bb_up = (bb_creds[0], bb_creds[2])
# response = request(
# method, f"https://{bb_host}{path}", data=data, json=json, **kwargs, params=params, auth=bb_up)
# if response.status_code >= 400:
# raise Exception(f"BitBucket responded with status code: {response.status_code}")
# return response
if __name__ == '__main__':
main()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment