Skip to content
Navigation Menu
{{ message }}
-
Notifications
You must be signed in to change notification settings - Fork 38
Expand file tree
/
Copy pathgithub.py
More file actions
308 lines (213 loc) · 9.54 KB
/
Copy pathgithub.py
File metadata and controls
308 lines (213 loc) · 9.54 KB
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
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
"""GitHub helper functions."""
import asyncio
import logging
import re
from github3 import GitHub
from github3.exceptions import GitHubException
from scriptworker.exceptions import ConfigError
from scriptworker.utils import get_parts_of_url_path, get_single_item_from_sequence, retry_async_decorator, retry_request, retry_sync
_GIT_FULL_HASH_PATTERN = re.compile(r"^[0-9a-f]{40}$")
# The github3 library already retries requests. It gives a round of waiting of usually 15 seconds.
# A delay factor of 7.5s means the second round of waiting will occur ~15s after the first one,
# the third one ~30s and so on.
_GITHUB_LIBRARY_SLEEP_TIME_KWARGS = {"delay_factor": 7.5}
log = logging.getLogger(__name__)
class GitHubRepository:
"""Wrapper around GitHub API. Used to access public data."""
def __init__(self, owner, repo_name, token=""):
"""Build the GitHub API URL which points to the definition of the repository.
Args:
owner (str): the owner's GitHub username
repo_name (str): the name of the repository
token (str): the GitHub API token
Returns:
dict: a representation of the repo definition
"""
github = retry_sync(GitHub, kwargs={"token": token}, sleeptime_kwargs=_GITHUB_LIBRARY_SLEEP_TIME_KWARGS)
self._github_repository = retry_sync(github.repository, args=(owner, repo_name), sleeptime_kwargs=_GITHUB_LIBRARY_SLEEP_TIME_KWARGS)
@property
def definition(self):
"""Fetch the definition of the repository, exposed by the GitHub API.
Returns:
dict: a representation of the repo definition
"""
return self._github_repository.as_dict()
@retry_async_decorator(retry_exceptions=GitHubException)
async def get_commit(self, commit_hash):
"""Fetch the definition of the commit, exposed by the GitHub API.
Args:
commit_hash (str): the hash of the git commit
Returns:
dict: a representation of the commit
"""
return self._github_repository.commit(commit_hash).as_dict()
@retry_async_decorator(retry_exceptions=GitHubException)
async def get_pull_request(self, pull_request_number):
"""Fetch the definition of the pull request, exposed by the GitHub API.
Args:
pull_request_number (int): the ID of the pull request
Returns:
dict: a representation of the pull request
"""
return self._github_repository.pull_request(pull_request_number).as_dict()
@retry_async_decorator(retry_exceptions=GitHubException)
async def get_release(self, tag_name):
"""Fetch the definition of the release matching the tag name.
Args:
tag_name (str): the tag linked to the release
Returns:
dict: a representation of the tag
"""
return self._github_repository.release_from_tag(tag_name).as_dict()
@retry_async_decorator(retry_exceptions=GitHubException)
async def get_tag_hash(self, tag_name):
"""Fetch the commit hash that was tagged with ``tag_name``.
Args:
tag_name (str): the name of the tag
Returns:
str: the commit hash linked by the tag
"""
tag_object = get_single_item_from_sequence(
sequence=self._github_repository.tags(),
condition=lambda tag: tag.name == tag_name,
no_item_error_message='No tag "{}" exist'.format(tag_name),
too_many_item_error_message='Too many tags "{}" found'.format(tag_name),
)
return tag_object.commit.sha
async def has_commit_landed_on_repository(self, context, revision):
"""Tell if a commit was landed on the repository or if it just comes from a pull request.
Args:
context (scriptworker.context.Context): the scriptworker context.
revision (str): the commit hash or the tag name.
Returns:
bool: True if the commit is present in one of the branches of the main repository
"""
if any(vcs_rule.get("require_secret") for vcs_rule in context.config["trusted_vcs_rules"]):
# This check uses unofficial API on github, which we can't easily
# check for private repos, assume its true in the private case.
log.info("has_commit_landed_on_repository() not implemented for private repositories, assume True")
return True
# Revision may be a tag name. `branch_commits` doesn't work on tags
if not _is_git_full_hash(revision):
revision = await self.get_tag_hash(tag_name=revision)
html_text = await _fetch_github_branch_commits_data(context, self._github_repository.html_url, revision)
# https://github.com/{repo_owner}/{repo_name}/branch_commits/{revision} just returns some \n
# when the commit hasn't landed on the origin repo. Otherwise, some HTML data is returned - it
# represents the branches on which the given revision is present.
return html_text != ""
_BRANCH_COMMITS_CACHE_TTL_IN_SECONDS = 10 * 60 # 10 minutes
_BRANCH_COMMITS_CACHE = {}
async def _fetch_github_branch_commits_data(context, repo_html_url, revision):
# Include context identity because different contexts carry
# different HTTP sessions that may be closed independently.
cache_key = (id(context), repo_html_url, revision)
if cache_key in _BRANCH_COMMITS_CACHE:
return await _BRANCH_COMMITS_CACHE[cache_key]
future = asyncio.get_running_loop().create_future()
_BRANCH_COMMITS_CACHE[cache_key] = future
try:
url = "/".join((repo_html_url.rstrip("/"), "branch_commits", revision))
html_text = await retry_request(context, url)
result = html_text.strip()
future.set_result(result)
asyncio.get_running_loop().call_later(
_BRANCH_COMMITS_CACHE_TTL_IN_SECONDS,
_BRANCH_COMMITS_CACHE.pop,
cache_key,
None,
)
except BaseException as e:
_BRANCH_COMMITS_CACHE.pop(cache_key, None)
future.set_exception(e)
raise
return result
def is_github_url(url):
"""Tell if a given URL matches a Github one.
Args:
url (str): The URL to test. It can be None.
Returns:
bool: False if the URL is not a string or if it doesn't match a Github URL
"""
if isinstance(url, str):
return url.startswith(("https://github.com/", "ssh://github.com/"))
else:
return False
def extract_github_repo_owner_and_name(url):
"""Given an URL, return the repo name and who owns it.
Args:
url (str): The URL to the GitHub repository
Raises:
ValueError: on url that aren't from github
Returns:
str, str: the owner of the repository, the repository name
"""
_check_github_url_is_supported(url)
parts = get_parts_of_url_path(url)
repo_owner = parts[0]
repo_name = parts[1]
return repo_owner, _strip_trailing_dot_git(repo_name)
def extract_github_repo_full_name(url):
"""Given an URL, return the full name of it.
The full name is ``RepoOwner/RepoName``.
Args:
url (str): The URL to the GitHub repository
Raises:
ValueError: on url that aren't from github
Returns:
str: the full name.
"""
return "/".join(extract_github_repo_owner_and_name(url))
def extract_github_repo_ssh_url(url):
"""Given an URL, return the ssh url.
Args:
url (str): The URL to the GitHub repository
Raises:
ValueError: on url that aren't from github
Returns:
str: the ssh url
"""
return "git@github.com:{}.git".format(extract_github_repo_full_name(url))
def extract_github_repo_and_revision_from_source_url(url):
"""Given an URL, return the repo name and who owns it.
Args:
url (str): The URL to the GitHub repository
Raises:
ValueError: on url that aren't from github or when the revision cannot be extracted
Returns:
str, str: the owner of the repository, the repository name
"""
_check_github_url_is_supported(url)
parts = get_parts_of_url_path(url)
repo_name = parts[1]
try:
revision = parts[3]
except IndexError:
raise ValueError("Revision cannot be extracted from url: {}".format(url))
end_index = url.index(repo_name) + len(repo_name)
repo_url = url[:end_index]
return _strip_trailing_dot_git(repo_url), revision
def _strip_trailing_dot_git(url):
if url.endswith(".git"):
url = url[: -len(".git")]
return url
def is_github_repo_owner_the_official_one(context, repo_owner):
"""Given a repo_owner, check if it matches the one configured to be the official one.
Args:
context (scriptworker.context.Context): the scriptworker context.
repo_owner (str): the repo_owner to verify
Raises:
scriptworker.exceptions.ConfigError: when no official owner was defined
Returns:
bool: True when ``repo_owner`` matches the one configured to be the official one
"""
official_repo_owner = context.config["official_github_repos_owner"]
if not official_repo_owner:
raise ConfigError(
'This worker does not have a defined owner for official GitHub repositories. Given "official_github_repos_owner": {}'.format(official_repo_owner)
)
return official_repo_owner == repo_owner
def _is_git_full_hash(revision):
return _GIT_FULL_HASH_PATTERN.match(revision) is not None
def _check_github_url_is_supported(url):
if not is_github_url(url):
raise ValueError('"{}" is not a supported GitHub URL!'.format(url))
You can’t perform that action at this time.