#!/usr/bin/env python3 # # Copyright 2019 Red Hat, Inc # # 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. import datetime import time import requests import urllib.parse from ansible.module_utils.basic import AnsibleModule SB_REPO = 'https://storyboard.openstack.org/#!/project/{org}/{repo}' SB_FORMAT = 'https://storyboard.openstack.org/#!/story/{{index}}' LP_REPO = 'https://bugs.launchpad.net/{repo}' LP_FORMAT = 'https://bugs.launchpad.net/{repo}/+bug/{{index}}' class Gitea(object): def __init__(self, url, password, always_update, projects): self.url = url self.password = password self.always_update = always_update self.projects = projects self.orgs = { f['project'].split('/')[0] for f in self.projects } self._log = [] def log(self, *args): now = datetime.datetime.utcnow().isoformat() self._log.append(" ".join((now,) + args)) def get_log(self): return "\n".join(self._log) def request(self, method, endpoint, *args, **kwargs): resp = requests.request( method, urllib.parse.urljoin(self.url, endpoint), auth=('root', self.password), verify=False, *args, **kwargs) resp.raise_for_status() return resp def get(self, endpoint, *args, **kwargs): return self.request('GET', endpoint, *args, **kwargs) def post(self, endpoint, *args, **kwargs): return self.request('POST', endpoint, *args, **kwargs) def put(self, endpoint, *args, **kwargs): return self.request('PUT', endpoint, *args, **kwargs) def get_gitea_orgs(self): orgs = self.get("/api/v1/user/orgs").json() return [f['username'] for f in orgs] def make_gitea_org(self, org): self.post( '/api/v1/admin/users/root/orgs', json=dict(username=org)) self.log("Created org:", org) def ensure_gitea_teams(self, org): team_list = self.get('/api/v1/orgs/{org}/teams'.format(org=org)).json() owner_id = [f['id'] for f in team_list if f['name'] == 'Owners'][0] org_owners = self.get( '/api/v1/teams/{owner_id}/members'.format(owner_id=owner_id)) if 'gerrit' not in [f['username'] for f in org_owners.json()]: self.put('/api/v1/teams/{owner_id}/members/gerrit'.format( owner_id=owner_id)) self.log("Added gerrit to team:", org) def get_org_repo_list(self, org): return self.get('/api/v1/orgs/{org}/repos'.format(org=org)).json() def get_csrf_token(self): resp = self.get('/') return urllib.parse.unquote(resp.cookies.get('_csrf')) def make_gitea_project(self, project, csrf_token): org, repo = project['project'].split('/', 1) resp = self.post( '/api/v1/org/{org}/repos'.format(org=org), json=dict( auto_init=True, description=project.get('description', '')[:255], name=repo, private=False, readme='Default')) self.log("Created repo:", project['project']) if project.get('use-storyboard'): external_tracker_url = SB_REPO.format(org=org, repo=repo) tracker_url_format = SB_FORMAT elif project.get('groups'): external_tracker_url = LP_REPO.format(repo=project['groups'][0]) tracker_url_format = LP_FORMAT.format(repo=project['groups'][0]) else: external_tracker_url = LP_REPO.format(repo=repo) tracker_url_format = LP_FORMAT.format(repo=repo) self.post( '/{org}/{repo}/settings'.format(org=org, repo=repo), data=dict( _csrf=csrf_token, action='advanced', # enable_pulls is not provided, which disables it # enable_wiki is not provided, which disables it enable_external_wiki=False, external_wiki_url='', # enable_issues is on so that issue links work enable_issues='on', enable_external_tracker=True, external_tracker_url=external_tracker_url, tracker_url_format=tracker_url_format, tracker_issue_style='numeric', )) self.log("Updated tracker url:", external_tracker_url) for count in range(0, 5): try: self.post( '/{org}/{repo}/settings/branches'.format( org=org, repo=repo), data=dict( _csrf=csrf_token, action='default_branch', branch='master', )) self.log("Set master branch:", project['project']) return except requests.exceptions.HTTPError as e: time.sleep(3) raise Exception("Could not update branch settings") def run(self): gitea_orgs = self.get_gitea_orgs() gitea_repos = [] for org in self.orgs: if org not in gitea_orgs: self.make_gitea_org(org) self.ensure_gitea_teams(org) gitea_repos.extend(self.get_org_repo_list(org)) csrf_token = self.get_csrf_token() for project in self.projects: if project['project'] not in gitea_repos or self.always_update: self.make_gitea_project(project, csrf_token) def ansible_main(): module = AnsibleModule( argument_spec=dict( url=dict(required=True), password=dict(required=True), projects=dict(required=True, type='list'), always_update=dict(type='bool', default=True), ) ) p = module.params gitea = Gitea( url=p.get('url'), password=p.get('password'), always_update=p.get('always_update'), projects=p.get('projects'), ) try: gitea.run() except Exception as e: module.fail_json(msg=str(e), changed=True) module.exit_json(changed=True, log=gitea.get_log()) if __name__ == '__main__': ansible_main()