Skip to content

Instantly share code, notes, and snippets.

Show Gist options
  • Save nicwolff/931c517e04bce86fbb07b51f5740fa58 to your computer and use it in GitHub Desktop.
Save nicwolff/931c517e04bce86fbb07b51f5740fa58 to your computer and use it in GitHub Desktop.
import boto3
from base64 import b64decode
import json
from urllib2 import urlopen, Request
ENCRYPTED_GITHUB_TOKEN = (
# aws kms encrypt --profile <PROFILE_NAME> --region <REGION>
# --key-id <KMS_KEY_ID> --plaintext <GITHUB_ACCESS_TOKEN>
# --query CiphertextBlob --output text
)
kms = boto3.client('kms')
decrypted_token = kms.decrypt(CiphertextBlob=b64decode(ENCRYPTED_GITHUB_TOKEN))
GITHUB_TOKEN = decrypted_token['Plaintext']
def lambda_handler(event, context):
message = json.loads(event['Records'][0]['Sns']['Message'])
if 'pull_request' in message:
pulls = [message['pull_request']]
else:
pulls = get_pull_requests(message['repository'])
for pull in pulls:
if needs_rebase(pull):
set_status_to_failed(pull)
def get_pull_requests(repo):
return call_github_api(repo['pulls_url'].replace('{/number}', ''))
def needs_rebase(pull):
head_ref = pull['head']['ref']
base_ref = pull['base']['ref']
url = pull['base']['repo']['compare_url']
url = url.replace('{base}', base_ref).replace('{head}', head_ref)
compare = call_github_api(url)
return compare['status'] == 'diverged'
def set_status_to_failed(pull):
pull_branch_head = pull['head']['sha']
statuses_url = pull['head']['repo']['statuses_url']
base_ref = pull['base']['ref']
desc = 'git fetch; git rebase origin/' + base_ref + '; git push -f'
call_github_api(
statuses_url.replace('{sha}', pull_branch_head),
{'state': 'failure', 'description': desc, 'context': 'Needs rebase'}
)
def call_github_api(url, data=None):
request = Request(url, json.dumps(data) if data else None)
request.add_header('Authorization', 'token %s' % GITHUB_TOKEN)
response = urlopen(request)
return json.loads(response.read())
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment