1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
|
# Copyright 2019 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Class to manage a git repository via python.
This class is to be used to implement git commands over
a python API and manage the current state of the git repo.
Typical usage example:
r_man = RepoManager('https://github.com/google/oss-fuzz.git')
r_man.checkout('5668cc422c2c92d38a370545d3591039fb5bb8d4')
"""
import os
import shutil
import utils
class RepoManagerError(Exception):
"""Class to describe the exceptions in RepoManager."""
class RepoManager:
"""Class to manage git repos from python.
Attributes:
repo_url: The location of the git repo
base_dir: The location of where the repo clone is stored locally
repo_name: The name of the github project
repo_dir: The location of the main repo
"""
def __init__(self, repo_url, base_dir, repo_name=None):
"""Constructs a repo manager class.
Args:
repo_url: The github url needed to clone
base_dir: The full filepath where the git repo is located
repo_name: The name of the directory the repo is cloned to
"""
self.repo_url = repo_url
self.base_dir = base_dir
if repo_name:
self.repo_name = repo_name
else:
self.repo_name = os.path.basename(self.repo_url).strip('.git')
self.repo_dir = os.path.join(self.base_dir, self.repo_name)
self._clone()
def _clone(self):
"""Creates a clone of the repo in the specified directory.
Raises:
RepoManagerError if the repo was not able to be cloned
"""
if not os.path.exists(self.base_dir):
os.makedirs(self.base_dir)
self.remove_repo()
out, err = utils.execute(['git', 'clone', self.repo_url, self.repo_name],
location=self.base_dir)
if not self._is_git_repo():
raise RepoManagerError('%s is not a git repo' % self.repo_url)
def _is_git_repo(self):
"""Test if the current repo dir is a git repo or not.
Returns:
True if the current repo_dir is a valid git repo
"""
git_path = os.path.join(self.repo_dir, '.git')
return os.path.isdir(git_path)
def commit_exists(self, commit):
"""Checks to see if a commit exists in the project repo.
Args:
commit: The commit SHA you are checking
Returns:
True if the commit exits in the project
Raises:
ValueException: an empty string was passed in as a commit
"""
# Handle the exception case, if empty string is passed execute will
# raise a ValueError
if not commit.rstrip():
raise RepoManagerError('An empty string is not a valid commit SHA')
_, err_code = utils.execute(['git', 'cat-file', '-e', commit],
self.repo_dir)
return not err_code
def get_current_commit(self):
"""Gets the current commit SHA of the repo.
Returns:
The current active commit SHA
"""
out, _ = utils.execute(['git', 'rev-parse', 'HEAD'],
self.repo_dir,
check_result=True)
return out.strip('\n')
def get_commit_list(self, old_commit, new_commit):
"""Gets the list of commits(inclusive) between the old and new commits.
Args:
old_commit: The oldest commit to be in the list
new_commit: The newest commit to be in the list
Returns:
The list of commit SHAs from newest to oldest
Raises:
RepoManagerError when commits dont exist
"""
if not self.commit_exists(old_commit):
raise RepoManagerError('The old commit %s does not exist' % old_commit)
if not self.commit_exists(new_commit):
raise RepoManagerError('The new commit %s does not exist' % new_commit)
if old_commit == new_commit:
return [old_commit]
out, err_code = utils.execute(
['git', 'rev-list', old_commit + '..' + new_commit], self.repo_dir)
commits = out.split('\n')
commits = [commit for commit in commits if commit]
if err_code or not commits:
raise RepoManagerError('Error getting commit list between %s and %s ' %
(old_commit, new_commit))
# Make sure result is inclusive
commits.append(old_commit)
return commits
def checkout_commit(self, commit):
"""Checks out a specific commit from the repo.
Args:
commit: The commit SHA to be checked out
Raises:
RepoManagerError when checkout is not successful
"""
if not self.commit_exists(commit):
raise RepoManagerError('Commit %s does not exist in current branch' %
commit)
git_path = os.path.join(self.repo_dir, '.git', 'shallow')
if os.path.exists(git_path):
utils.execute(['git', 'fetch', '--unshallow'],
self.repo_dir,
check_result=True)
utils.execute(['git', 'checkout', '-f', commit],
self.repo_dir,
check_result=True)
utils.execute(['git', 'clean', '-fxd'], self.repo_dir, check_result=True)
if self.get_current_commit() != commit:
raise RepoManagerError('Error checking out commit %s' % commit)
def remove_repo(self):
"""Attempts to remove the git repo. """
if os.path.isdir(self.repo_dir):
shutil.rmtree(self.repo_dir)
|
