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
|
# Copyright 2020 Codethink Limited
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; version 2 of the License.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License along
# with this program; if not, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
import json
import urllib.error
import urllib.parse
import urllib.request
from . import hosts
class _Gitea:
'''Run commands on a Gitea instance.'''
def __init__(self, url, token):
self._api_url = urllib.parse.urljoin(url, 'api/v1')
self._api_token = token
def request(self, method, *path, headers={}, data=None):
url = self._api_url
for part in path:
url = url + '/' + urllib.parse.quote(part, safe='')
headers = headers.copy()
headers['Authorization'] = 'token ' + self._api_token
if data is not None:
if method == 'GET':
data = urllib.parse.urlencode(data).encode('ascii')
elif method in ['POST', 'PATCH']:
data = json.dumps(data).encode('utf-8')
headers['Content-Type'] = 'application/json; charset=utf-8'
else:
assert isinstance(data, bytes)
request = urllib.request.Request(url, data=data, headers=headers,
method=method)
with urllib.request.urlopen(request) as response:
if response.getcode() == 204:
return None
return json.loads(response.read().decode('utf-8'))
class GiteaDownstream(hosts.DownstreamHost):
@staticmethod
def add_app_settings(app_settings):
app_settings.string(
['gitea-access-token'],
'access token for Gitea API access')
@staticmethod
def check_app_settings(app_settings):
app_settings.require('downstream-http-url')
app_settings.require('gitea-access-token')
def __init__(self, app_settings):
self._gitea = _Gitea(app_settings['downstream-http-url'],
app_settings['gitea-access-token'])
# Gitea supports all visibilities for organisations, but
# repositories can only be private or inherit the visibility
# of the organisation. If repository visibility is supposed
# to be 'internal', assume that the group will be 'internal'
# rather than making the repository less visible.
visibility = app_settings['downstream-visibility']
self._group_vis = 'limited' if visibility == 'internal' else visibility
self._repo_private = visibility == 'private'
def prepare_repo(self, repo_path, metadata):
path_comps = repo_path.split('/')
# As of Gitea 1.11.5, a 2-level hierarchy must be used:
# user/organisation and repository. Deeper hiearchies must be
# collapsed using a prefixmap.
if len(path_comps) < 2:
raise ValueError(
'cannot create Gitea repository outside an organisation')
if len(path_comps) > 2:
raise ValueError('cannot create nested Gitea organisations')
org_name, repo_name = path_comps
repo_edit = {}
try:
repo = self._gitea.request('GET', 'repos', org_name, repo_name)
except urllib.error.HTTPError as e:
if e.code != 404:
raise
# Get or create organisation
try:
self._gitea.request('GET', 'orgs', org_name)
except urllib.error.HTTPError as e:
if e.code != 404:
raise
org_data = {
'username': org_name,
'visibility': self._group_vis,
}
self._gitea.request('POST', 'orgs', data=org_data)
# Create repository. This was under /org, not /orgs,
# before Gitea 1.11.5 (which supports both).
repo_create = {
'auto_init': False,
'name': repo_name,
'private': self._repo_private,
}
repo = self._gitea.request('POST', 'org', org_name, 'repos',
data=repo_create)
# These can only be turned off after creating the repo
for feature in ['has_issues', 'has_wiki', 'has_pull_requests']:
if repo[feature] != False:
repo_edit[feature] = False
# Update repository metadata
if 'head' in metadata and repo['default_branch'] != metadata['head']:
repo_edit['default_branch'] = metadata['head']
if 'description' in metadata \
and repo['description'] != metadata['description']:
repo_edit['description'] = metadata['description']
if repo_edit:
self._gitea.request('PATCH', 'repos', org_name, repo_name,
data=repo_edit)
|
