HGKeeper

# vi:et:ts=4 sw=4 sts=4

import os

import logging

import subprocess

from flask import Flask, request, make_response

logging.getLogger().setLevel(logging.NOTSET)

app = Flask(__name__)

class HG(object):

def __init__(self, path):

self.path = path

def pull(self):

return f'hg pull --cwd {self.path} incoming'

def push(self):

return f'hg push --cwd {self.path} --new-branch outgoing'

VCS_TOOLS = {

'hg': HG,

}

@app.route('/hooked', methods=['POST'])

def hooked():

data = request.get_json()

if 'repository' in data:

repo = data.get('repository')

path = os.path.join('repos', repo.get('full_name'))

if not os.path.isdir(path):

msg = f'unknown repository {repo.get("full_name")}'

logging.error(msg)

return make_response((msg, 400))

vcs_cls = VCS_TOOLS.get(repo.get('scm'), None)

if vcs_cls is None:

msg = f'unknown vcs {repo.get("scm")}'

logging.error(msg)

return make_response((msg, 400))

vcs = vcs_cls(path)

try:

out = subprocess.check_output(vcs.pull().split())

logging.info(out)

out = subprocess.check_output(vcs.push().split())

logging.info(out)

except subprocess.CalledProcessError as exp:

if exp.returncode == 1:

logging.info(exp.output)

return make_response((exp.output, 500))

return make_response((exp, 500))

except Exception as exp:

logging.exception(exp)

return make_response((exp, 500))

else:

return make_response(('bitshifted', 200))

if __name__ == '__main__':

app.run(host='0.0.0.0', port=11101, debug=False)

grim/bitshifter