SDLC helper for Jira and TFS users.

sdlc_helper.py

import sys
import subprocess
import getpass
from optparse import OptionParser
import requests
import json

'''
This script is useful for maintaining an easy SDLC process.
Using this script, the build master and quickly find all the Jira tickets
in Dev Complete state and with pull requests against master (non conflicting)
Other uses:
  Find all the open PRs.
  Find all the dev complete tickets.
  Find all ticket with any PR (not against master or conflicting included)
  
Assumption: Branch name is the same as ticket id.
Author: Mahesh Gattani ([email protected])
'''

def find_dev_complete_tickets(username, password, jira_base_url):
   dev_complete_query = jira_base_url + '/rest/api/2/search?jql=project="SPM" AND status="Dev Complete"'
   resp = requests.get(dev_complete_query, auth=(username, password))
   data = json.loads(resp.text)
   dev_completed_issues = []
   for issue in data['issues']:
      url = 'https://my-jet.atlassian.net/browse/' + issue['key']
      assignee = None
      if 'fields' in issue and issue['fields'] is not None:
        if 'assignee' in issue['fields'] and issue['fields']['assignee'] is not None:
          if 'emailAddress' in issue['fields']['assignee']:
            assignee = issue['fields']['assignee']['emailAddress']
      dev_completed_issues.append((url, issue['key'], assignee))
   return dev_completed_issues

def find_pull_requests(tfs_key, tfs_base_url):
   pull_requests_query = tfs_base_url + '/pullRequests?api-version=1.0'
   resp = requests.get(pull_requests_query, auth=("", tfs_key))
   data = json.loads(resp.text)
   pull_requests = data['value']
   processed_pull_requests = []
   for pull_request in pull_requests:
     source_branch = pull_request['sourceRefName']
     target_branch = pull_request['targetRefName']
     merge_status = pull_request['mergeStatus']
     status = pull_request['status']
     created_by = pull_request['createdBy']['uniqueName']
     processed_pull_requests.append((source_branch, target_branch, merge_status, status, created_by))

   return processed_pull_requests

def check_issue_merable(issue_id, processed_pull_requests):
   issue_branch = 'refs/heads/' + issue_id
   for source_branch, target_branch, merge_status, status, created_by in processed_pull_requests:
     if source_branch == issue_branch and merge_status == 'succeeded' and target_branch == 'refs/heads/master' and status == 'active':
       return True
   return False

def handle_mergable_tickets(dev_completed_issues, processed_pull_requests):
   mergable_tickets = []
   for url, issue_id, assignee in dev_completed_issues:
     if check_issue_merable(issue_id, processed_pull_requests):
       mergable_tickets.append((url, issue_id, assignee))
   return mergable_tickets

def check_issue_has_pull_request(issue_id, processed_pull_requests):
   for source_branch, target_branch, merge_status, status, created_by in processed_pull_requests:
     if issue_id in source_branch:
       return True
   return False

def handle_tickets_with_pull_requests(dev_completed_issues, processed_pull_requests):
   tickets = []
   for url, issue_id, assignee in dev_completed_issues:
     if check_issue_has_pull_request(issue_id, processed_pull_requests):
       tickets.append((url, issue_id, assignee))
   return tickets

def print_jira_tickets(tickets):
  for url, issue_id, assignee in tickets:
    print ("Ticket: %s. Assignee: %s" % (url, assignee))

def print_tfs_pull_requests(pull_requests):
  pull_requests.sort(key = lambda (source_branch, target_branch, merge_status, status, created_by): merge_status)
  for source_branch, target_branch, merge_status, status, created_by in pull_requests:
    print ("Source Branch: %s. Target Branch: %s. Merge Status: %s. PR Status: %s. Created B: %s" % (source_branch, target_branch, merge_status, status, created_by))

def handle_merge(mergable_tickets, jira_username, tfs_key):
   should_merge = raw_input('Do you want to try and merge these tickets? (Y/n)')
   if should_merge == 'Y':
     git_url = raw_input('TFS Repo url (Example: https://<instance>.visualstudio.com/DefaultCollection/<project>/_git/<repository>"): ')
     branch = raw_input('Branch name (Branch will be created if it does not exist.): ')
     if branch == None or branch == '':
       print 'No branch provided. Exiting.'
       sys.exit(2)
     if branch == 'master':
       print 'Merge branch cannot be master'
       sys.exit(2)
     git_url = git_url.replace('https://', '').replace('http://', '')
     tfs_username = (jira_username.split('@'))[0]
     tfs_checkout_url = ('git clone "https://%s:%s@%s"' % (tfs_username, tfs_key, git_url))
     repo = (git_url.split('/'))[-1]
     command = ('%s;\ncd %s;\ngit checkout -b %s;\n' % (tfs_checkout_url, repo, branch))
     for url, issue_id, assignee in mergable_tickets:
       command = command + 'git merge origin/' + issue_id + ';\n'
     command = command + 'git push origin ' + branch + ';\n'
     
     print '\n---------------EXECUTING MERGE---------\n'
     print subprocess.Popen(command, stdout=subprocess.PIPE, shell=True).stdout.read()

     clean_up_command = 'rm -rf ' + repo + ';'
     print subprocess.Popen(clean_up_command, stdout=subprocess.PIPE, shell=True).stdout.read()
     print '\n---------------CLEANING UP------------\n'

     print 'Branch ' + branch + ' is pushed and ready to be built'

def main(argv):
   parser = OptionParser()
   parser.add_option('-u', '--jira_username', dest = 'jira_username')
   parser.add_option('-m', '--mode', dest = 'mode', help = 'Options: "md" for mergable dev complete tickets. "d" for dev complete tickets. "m" for mergable PRs. "dp" for dev complete tikets with any PR')
   parser.add_option('-j', '--jira_base_url', dest = 'jira_base_url', help = 'Example: "https://<instance>.atlassian.net"')
   parser.add_option('-t', '--tfs_base_url', dest = 'tfs_base_url', help = 'Example: "https://<instance>.visualstudio.com/DefaultCollection/<project>/_apis/git/repositories/<repository>"')
   (options, args) = parser.parse_args()
   jira_username = options.jira_username
   mode = options.mode
   jira_base_url = options.jira_base_url
   tfs_base_url = options.tfs_base_url

   if not mode or not jira_username or not jira_base_url or not tfs_base_url:
     print 'Insufficient arguments provided. Try: "<script>.py -h" for help'
     sys.exit(2)
   
   if mode != 'md' and mode != 'd' and mode != 'm' and mode != 'dp':
     print 'Please enter a correct mode. Options: "md" for mergable dev complete tickets. "d" for dev complete tickets. "m" for mergable PRs. "dp" for dev complete tikets with any PR'
     sys.exit(2)

   jira_password = getpass.getpass('Enter JIRA password: ')
   tfs_key = getpass.getpass('Enter TFS key (You can create one here https://<instance>.visualstudio.com/_details/security/tokens): ')

   dev_completed_issues = find_dev_complete_tickets(jira_username, jira_password, jira_base_url)
   processed_pull_requests = find_pull_requests(tfs_key, tfs_base_url)

   if mode == 'md':
     mergable_tickets = handle_mergable_tickets(dev_completed_issues, processed_pull_requests)
     if len(mergable_tickets) == 0:
       print 'No tickets in Dev Complete state and an out standing non-conflicting PR against master'
     else:
       print '\n--------MERGABLE TICKETS--------\n'
       print_jira_tickets(mergable_tickets)
       print '\n'
       handle_merge(mergable_tickets, jira_username, tfs_key)
   elif mode == 'd':
     print 'Dev complete issues in Jira are:'
     print_jira_tickets(dev_completed_issues)
   elif mode == 'm':
     print 'Outstanding mergeable PRs are:'
     print_tfs_pull_requests(processed_pull_requests)
   elif mode == 'dp':
     tickets = handle_tickets_with_pull_requests(dev_completed_issues, processed_pull_requests)
     if len(tickets) == 0:
       print 'No tickets in Dev Complete state and an out standing PR'
     else:
       print_jira_tickets(tickets)

if __name__ == "__main__":
   main(sys.argv[1:])