1#!/usr/bin/env python 2# Copyright (c) 2014 The Chromium Authors. All rights reserved. 3# Use of this source code is governed by a BSD-style license that can be 4# found in the LICENSE file. 5 6"""This module contains functions for using git.""" 7 8import os 9import re 10import shutil 11import subprocess 12import tempfile 13 14import utils 15 16 17class GitLocalConfig(object): 18 """Class to manage local git configs.""" 19 def __init__(self, config_dict): 20 self._config_dict = config_dict 21 self._previous_values = {} 22 23 def __enter__(self): 24 for k, v in self._config_dict.items(): 25 try: 26 prev = subprocess.check_output([ 27 'git', 'config', '--local', k]).decode('utf-8').rstrip() 28 if prev: 29 self._previous_values[k] = prev 30 except subprocess.CalledProcessError: 31 # We are probably here because the key did not exist in the config. 32 pass 33 subprocess.check_call(['git', 'config', '--local', k, v]) 34 35 def __exit__(self, exc_type, _value, _traceback): 36 for k in self._config_dict: 37 if self._previous_values.get(k): 38 subprocess.check_call( 39 ['git', 'config', '--local', k, self._previous_values[k]]) 40 else: 41 subprocess.check_call(['git', 'config', '--local', '--unset', k]) 42 43 44class GitBranch(object): 45 """Class to manage git branches. 46 47 This class allows one to create a new branch in a repository to make changes, 48 then it commits the changes, switches to main branch, and deletes the 49 created temporary branch upon exit. 50 """ 51 def __init__(self, branch_name, commit_msg, upload=True, commit_queue=False, 52 delete_when_finished=True, cc_list=None): 53 self._branch_name = branch_name 54 self._commit_msg = commit_msg 55 self._upload = upload 56 self._commit_queue = commit_queue 57 self._patch_set = 0 58 self._delete_when_finished = delete_when_finished 59 self._cc_list = cc_list 60 61 def __enter__(self): 62 subprocess.check_call(['git', 'reset', '--hard', 'HEAD']) 63 subprocess.check_call(['git', 'checkout', 'main']) 64 if self._branch_name in subprocess.check_output([ 65 'git', 'branch']).decode('utf-8').split(): 66 subprocess.check_call(['git', 'branch', '-D', self._branch_name]) 67 subprocess.check_call(['git', 'checkout', '-b', self._branch_name, 68 '-t', 'origin/main']) 69 return self 70 71 def commit_and_upload(self, use_commit_queue=False): 72 """Commit all changes and upload a CL, returning the issue URL.""" 73 subprocess.check_call(['git', 'commit', '-a', '-m', self._commit_msg]) 74 upload_cmd = ['git', 'cl', 'upload', '-f', '--bypass-hooks', 75 '--bypass-watchlists'] 76 self._patch_set += 1 77 if self._patch_set > 1: 78 upload_cmd.extend(['-t', 'Patch set %d' % self._patch_set]) 79 if use_commit_queue: 80 upload_cmd.append('--use-commit-queue') 81 # Need the --send-mail flag to publish the CL and remove WIP bit. 82 upload_cmd.append('--send-mail') 83 if self._cc_list: 84 upload_cmd.extend(['--cc=%s' % ','.join(self._cc_list)]) 85 subprocess.check_call(upload_cmd) 86 output = subprocess.check_output([ 87 'git', 'cl', 'issue']).decode('utf-8').rstrip() 88 return re.match('^Issue number: (?P<issue>\d+) \((?P<issue_url>.+)\)$', 89 output).group('issue_url') 90 91 def __exit__(self, exc_type, _value, _traceback): 92 if self._upload: 93 # Only upload if no error occurred. 94 try: 95 if exc_type is None: 96 self.commit_and_upload(use_commit_queue=self._commit_queue) 97 finally: 98 subprocess.check_call(['git', 'checkout', 'main']) 99 if self._delete_when_finished: 100 subprocess.check_call(['git', 'branch', '-D', self._branch_name]) 101 102 103class NewGitCheckout(utils.tmp_dir): 104 """Creates a new local checkout of a Git repository.""" 105 106 def __init__(self, repository, local=None): 107 """Set parameters for this local copy of a Git repository. 108 109 Because this is a new checkout, rather than a reference to an existing 110 checkout on disk, it is safe to assume that the calling thread is the 111 only thread manipulating the checkout. 112 113 You must use the 'with' statement to create this object: 114 115 with NewGitCheckout(*args) as checkout: 116 # use checkout instance 117 # the checkout is automatically cleaned up here 118 119 Args: 120 repository: URL of the remote repository (e.g., 121 'https://skia.googlesource.com/common') or path to a local repository 122 (e.g., '/path/to/repo/.git') to check out a copy of 123 local: optional path to an existing copy of the remote repo on local disk. 124 If provided, the initial clone is performed with the local copy as the 125 upstream, then the upstream is switched to the remote repo and the 126 new copy is updated from there. 127 """ 128 super(NewGitCheckout, self).__init__() 129 self._checkout_root = '' 130 self._repository = repository 131 self._local = local 132 133 @property 134 def name(self): 135 return self._checkout_root 136 137 @property 138 def root(self): 139 """Returns the root directory containing the checked-out files.""" 140 return self.name 141 142 def __enter__(self): 143 """Check out a new local copy of the repository. 144 145 Uses the parameters that were passed into the constructor. 146 """ 147 super(NewGitCheckout, self).__enter__() 148 remote = self._repository 149 if self._local: 150 remote = self._local 151 subprocess.check_call(['git', 'clone', remote]) 152 repo_name = remote.split('/')[-1] 153 if repo_name.endswith('.git'): 154 repo_name = repo_name[:-len('.git')] 155 self._checkout_root = os.path.join(os.getcwd(), repo_name) 156 os.chdir(repo_name) 157 if self._local: 158 subprocess.check_call([ 159 'git', 'remote', 'set-url', 'origin', self._repository]) 160 subprocess.check_call(['git', 'remote', 'update']) 161 subprocess.check_call(['git', 'checkout', 'main']) 162 subprocess.check_call(['git', 'reset', '--hard', 'origin/main']) 163 return self 164