104 lines
2.9 KiB
Python
104 lines
2.9 KiB
Python
""" GogsMaker
|
|
|
|
A webhook-handler for Gogs running `make` when needed. """
|
|
|
|
import os
|
|
import subprocess
|
|
from functools import wraps
|
|
from flask import Flask, request
|
|
from . import settings
|
|
|
|
app = Flask(__name__)
|
|
|
|
|
|
class UnmonitoredRepository(Exception):
|
|
pass
|
|
|
|
|
|
class GitError(Exception):
|
|
def __init__(self, what):
|
|
super().__init__()
|
|
self.what = what
|
|
|
|
def __str__(self):
|
|
return self.what
|
|
|
|
|
|
def get_hook(url):
|
|
''' Get the hook matching an URL, or raise UnmonitoredRepository '''
|
|
for hook in settings.HOOKS:
|
|
if hook['url'] == url:
|
|
return hook
|
|
raise UnmonitoredRepository
|
|
|
|
|
|
def repo_path(hook):
|
|
''' Get the path at which the hook's repo is cloned '''
|
|
return os.path.join(settings.CLONE_ROOT, hook['name'])
|
|
|
|
|
|
def update_repo(hook, clone_url):
|
|
''' Update (or clone) the given repository. May raise GitError. '''
|
|
path = repo_path(hook)
|
|
if os.path.isdir(os.path.join(path, '.git')): # Repo is already cloned
|
|
try:
|
|
subprocess.run(['git', '-C', path, 'reset', '--hard'],
|
|
check=True) # Just in case.
|
|
subprocess.run(['git', '-C', path, 'pull'], check=True)
|
|
except subprocess.CalledProcessError:
|
|
raise GitError("Cannot pull {}".format(hook['name']))
|
|
|
|
else: # Simply update
|
|
try:
|
|
subprocess.run(['mkdir', '-p', path])
|
|
subprocess.run(['git', 'clone', clone_url, path], check=True)
|
|
except subprocess.CalledProcessError:
|
|
raise GitError("Cannot clone {}".format(clone_url))
|
|
|
|
|
|
def gogs_payload(required):
|
|
def wrapper(fct):
|
|
@wraps(fct)
|
|
def wrapped(*args, **kwargs):
|
|
# TODO: check signature
|
|
# payload_raw = request.data
|
|
|
|
payload = request.json
|
|
if payload is None:
|
|
return 'Expected json\n', 415
|
|
|
|
for field in required + ['repository/html_url']:
|
|
path = field.split('/')
|
|
explore = payload
|
|
for section in path:
|
|
if section not in explore:
|
|
return (
|
|
'Invalid json: missing {}\n'.format(
|
|
'/'.join(path)),
|
|
400)
|
|
explore = explore[section]
|
|
|
|
try:
|
|
hook = get_hook(payload['repository']['html_url'])
|
|
except UnmonitoredRepository:
|
|
return 'Unmonitored repository\n', 403
|
|
|
|
return fct(payload, hook, *args, **kwargs)
|
|
return wrapped
|
|
return wrapper
|
|
|
|
|
|
@app.route('/', methods=['POST'])
|
|
@gogs_payload(['repository/clone_url'])
|
|
def view_root(payload, hook):
|
|
clone_url = payload['repository']['clone_url']
|
|
|
|
try:
|
|
update_repo(hook, clone_url)
|
|
except GitError as error:
|
|
return 'Git error: {}\n'.format(error), 500
|
|
|
|
# TODO: make
|
|
|
|
return 'OK\n', 200
|