"""Cleans up PRs that correspond to abandoned CLs in Gerrit."""
import logging
from datetime import datetime, timedelta, timezone
from typing import List, Optional
from blinkpy.w3c.gerrit import GerritAPI, GerritError, GerritNotFoundError
from blinkpy.w3c.wpt_github import PullRequest
_log = logging.getLogger(__name__)
class PrCleanupTool(object):
def __init__(self, host):
self.host = host
def run(self, wpt_github, gerrit):
"""Closes all PRs that are abandoned two weeks ago in Gerrit."""
_log.info(
"Close exported PRs where the corresponding CLs have been abandoned."
)
pull_requests = self.retrieve_provisioned_prs(wpt_github)
for pull_request in pull_requests:
if pull_request.state != 'open':
continue
change_id = wpt_github.extract_metadata('Change-Id: ',
pull_request.body)
maybe_cleanup_reason = self._cleanup_reason(gerrit, change_id)
if maybe_cleanup_reason:
self.log_affected_pr_details(wpt_github, pull_request,
maybe_cleanup_reason)
self.close_pr_and_delete_branch(wpt_github,
pull_request.number,
maybe_cleanup_reason)
return True
def _cleanup_reason(self, gerrit: GerritAPI,
change_id: Optional[str]) -> Optional[str]:
"""Get a human-readable comment describing why a PR will be closed.
Returns `None` if the PR should not be closed.
"""
if not change_id:
return None
try:
cl = gerrit.query_cl(change_id)
except GerritNotFoundError:
return 'Close this PR because the corresponding CL has been deleted.'
except GerritError as e:
_log.error('Could not query change_id %s: %s', change_id, str(e))
return None
cl_status = cl.status
expiration = datetime.now(timezone.utc) - timedelta(days=14)
if cl_status == 'ABANDONED' and cl.updated < expiration:
return 'Close this PR because the Chromium CL has been abandoned.'
elif cl_status == 'MERGED' and not cl.is_exportable():
return 'Close this PR because the Chromium CL does not have exportable changes.'
return None
def retrieve_provisioned_prs(self, wpt_github) -> List[PullRequest]:
"""Retrieves last 1000 PRs with 'do not merge' label."""
return wpt_github.all_provisional_pull_requests()
def close_pr_and_delete_branch(self, wpt_github, pull_request_number,
comment):
"""Closes a PR with a comment and delete the corresponding branch."""
wpt_github.add_comment(pull_request_number, comment)
wpt_github.update_pr(pull_request_number, state='closed')
branch = wpt_github.get_pr_branch(pull_request_number)
wpt_github.delete_remote_branch(branch)
def log_affected_pr_details(self, wpt_github, pull_request, comment):
"""Logs details of an affected PR."""
_log.info(comment)
_log.info('https://github.com/web-platform-tests/wpt/pull/%s',
pull_request.number)
_log.info(
wpt_github.extract_metadata('Reviewed-on: ', pull_request.body))
Codebase Browser
chromium