| # Copyright 2016 The Chromium Authors. All rights reserved. |
| # Use of this source code is governed by a BSD-style license that can be |
| # found in the LICENSE file. |
| |
| import base64 |
| import json |
| import logging |
| import urllib2 |
| |
| |
| _log = logging.getLogger(__name__) |
| API_BASE = 'https://api.github.com' |
| EXPORT_LABEL = 'chromium-export' |
| |
| |
| class WPTGitHub(object): |
| |
| def __init__(self, host, user, token): |
| self.host = host |
| self.user = user |
| self.token = token |
| assert self.user and self.token |
| |
| def auth_token(self): |
| return base64.b64encode('{}:{}'.format(self.user, self.token)) |
| |
| def request(self, path, method, body=None): |
| assert path.startswith('/') |
| if body: |
| body = json.dumps(body) |
| opener = urllib2.build_opener(urllib2.HTTPHandler) |
| request = urllib2.Request(url=API_BASE + path, data=body) |
| request.add_header('Accept', 'application/vnd.github.v3+json') |
| request.add_header('Authorization', 'Basic {}'.format(self.auth_token())) |
| request.get_method = lambda: method |
| response = opener.open(request) |
| status_code = response.getcode() |
| try: |
| return json.load(response), status_code |
| except ValueError: |
| return None, status_code |
| |
| def create_pr(self, remote_branch_name, desc_title, body): |
| """Creates a PR on GitHub. |
| |
| API doc: https://developer.github.com/v3/pulls/#create-a-pull-request |
| |
| Returns: |
| A raw response object if successful, None if not. |
| """ |
| assert remote_branch_name |
| assert desc_title |
| assert body |
| |
| path = '/repos/w3c/web-platform-tests/pulls' |
| body = { |
| 'title': desc_title, |
| 'body': body, |
| 'head': remote_branch_name, |
| 'base': 'master', |
| } |
| data, status_code = self.request(path, method='POST', body=body) |
| |
| if status_code != 201: |
| return None |
| |
| return data |
| |
| def add_label(self, number): |
| path = '/repos/w3c/web-platform-tests/issues/%d/labels' % number |
| body = [EXPORT_LABEL] |
| return self.request(path, method='POST', body=body) |
| |
| def in_flight_pull_requests(self): |
| path = '/search/issues?q=repo:w3c/web-platform-tests%20is:open%20type:pr%20label:{}'.format(EXPORT_LABEL) |
| data, status_code = self.request(path, method='GET') |
| if status_code == 200: |
| return data['items'] |
| else: |
| raise Exception('Non-200 status code (%s): %s' % (status_code, data)) |
| |
| def merge_pull_request(self, pull_request_number): |
| path = '/repos/w3c/web-platform-tests/pulls/%d/merge' % pull_request_number |
| body = { |
| 'merge_method': 'rebase', |
| } |
| data, status_code = self.request(path, method='PUT', body=body) |
| |
| if status_code == 405: |
| raise Exception('PR did not passed necessary checks to merge: %d' % pull_request_number) |
| elif status_code == 200: |
| return data |
| else: |
| raise Exception('PR could not be merged: %d' % pull_request_number) |
| |
| def delete_remote_branch(self, remote_branch_name): |
| # TODO(jeffcarp): Unit test this method |
| path = '/repos/w3c/web-platform-tests/git/refs/heads/%s' % remote_branch_name |
| data, status_code = self.request(path, method='DELETE') |
| |
| if status_code != 204: |
| # TODO(jeffcarp): Raise more specific exception (create MergeError class?) |
| raise Exception('Received non-204 status code attempting to delete remote branch: {}'.format(status_code)) |
| |
| return data |