# 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)