diff --git a/gogsmaker.py b/gogsmaker.py index cddf985..c35097a 100644 --- a/gogsmaker.py +++ b/gogsmaker.py @@ -33,24 +33,24 @@ class GitError(Exception): def get_hook(url): - ''' Get the hook matching an URL, or raise UnmonitoredRepository ''' + """ Get the hook matching an URL, or raise UnmonitoredRepository """ for hook in settings.HOOKS: - if hook['url'] == url: + 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']) + """ Get the path at which the hook's repo is cloned """ + return os.path.join(settings.CLONE_ROOT, hook["name"]) def subprocess_run(command, **kwargs): - ''' Run subprocess with default arguments ''' + """ Run subprocess with default arguments """ args = { - 'check': True, - 'stdout': subprocess.DEVNULL, - 'stderr': subprocess.PIPE, + "check": True, + "stdout": subprocess.DEVNULL, + "stderr": subprocess.PIPE, } args.update(kwargs) @@ -58,64 +58,62 @@ def subprocess_run(command, **kwargs): class MakeWorker(Thread): - ''' A make job ''' + """ A make job """ def __init__(self, hook): super().__init__() self.hook = hook - self.name = 'makeworker-{}'.format(hook['name']) + self.name = "makeworker-{}".format(hook["name"]) self.path = repo_path(hook) def run(self): - ''' Run the make job ''' + """ Run the make job """ try: - subprocess_run(['make', '-C', self.path, '--'] - + self.hook['targets']) + subprocess_run(["make", "-C", self.path, "--"] + self.hook["targets"]) except subprocess.CalledProcessError as error: logging.error( - ("Hook %s: make failed with status %s. " - "Error output:\n%s\n"), - self.hook['name'], + ("Hook %s: make failed with status %s. " "Error output:\n%s\n"), + self.hook["name"], error.returncode, - error.stderr.decode('utf-8')) + error.stderr.decode("utf-8"), + ) def update_repo(hook, clone_url): - ''' Update (or clone) the given repository. May raise GitError. ''' + """ 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 + if os.path.isdir(os.path.join(path, ".git")): # Repo is already cloned try: - subprocess_run(['git', '-C', path, 'reset', '--hard'] - ) # Just in case. - subprocess_run(['git', '-C', path, 'pull']) + subprocess_run(["git", "-C", path, "reset", "--hard"]) # Just in case. + subprocess_run(["git", "-C", path, "pull"]) except subprocess.CalledProcessError as error: logging.error( - ("Hook %s: git failed with status %s. " - "Error output:\n%s\n"), - hook['name'], + ("Hook %s: git failed with status %s. " "Error output:\n%s\n"), + hook["name"], error.returncode, - error.stderr.decode('utf-8')) - raise GitError("Cannot pull {}".format(hook['name'])) + error.stderr.decode("utf-8"), + ) + raise GitError("Cannot pull {}".format(hook["name"])) else: # Repo is to be cloned try: - subprocess_run(['mkdir', '-p', path]) - subprocess_run(['git', 'clone', clone_url, path], check=True) + subprocess_run(["mkdir", "-p", path]) + subprocess_run(["git", "clone", clone_url, path], check=True) except subprocess.CalledProcessError as error: logging.error( - ("Hook %s: git failed cloning with status %s. " - "Error output:\n%s"), - hook['name'], + ("Hook %s: git failed cloning with status %s. " "Error output:\n%s"), + hook["name"], error.returncode, - error.stderr.decode('utf-8')) + error.stderr.decode("utf-8"), + ) raise GitError("Cannot clone {}".format(clone_url)) def check_signature(received_sig, hook, payload): - ''' Check Gogs signature ''' - digest = hmac.new(hook['secret'].encode('utf-8'), - msg=payload, - digestmod=sha256).hexdigest() + """ Check Gogs signature """ + digest = hmac.new( + hook["secret"].encode("utf-8"), msg=payload, digestmod=sha256 + ).hexdigest() return hmac.compare_digest(digest, received_sig) @@ -125,51 +123,53 @@ def gogs_payload(required): def wrapped(*args, **kwargs): payload = request.json if payload is None: - return 'Expected json\n', 415 + return "Expected json\n", 415 - for field in required + ['repository/html_url']: - path = field.split('/') + 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) + "Invalid json: missing {}\n".format("/".join(path)), + 400, + ) explore = explore[section] try: - hook = get_hook(payload['repository']['html_url']) + hook = get_hook(payload["repository"]["html_url"]) except UnmonitoredRepository: - return 'Unmonitored repository\n', 403 + return "Unmonitored repository\n", 403 if not settings.DEBUG: - received_sig = request.headers['X-Gogs-Signature'] + received_sig = request.headers["X-Gogs-Signature"] payload_raw = request.data if not check_signature(received_sig, hook, payload_raw): - return 'Invaild signature\n', 403 + return "Invaild signature\n", 403 return fct(payload, hook, *args, **kwargs) + return wrapped + return wrapper -@app.route('/', methods=['POST']) -@gogs_payload(['repository/clone_url']) +@app.route("/", methods=["POST"]) +@gogs_payload(["repository/clone_url"]) def view_root(payload, hook): - clone_url = payload['repository']['clone_url'] - if 'clone_url' in hook: - clone_url = hook['clone_url'] + clone_url = payload["repository"]["clone_url"] + if "clone_url" in hook: + clone_url = hook["clone_url"] try: update_repo(hook, clone_url) except GitError as error: - return 'Git error: {}\n'.format(error), 500 + return "Git error: {}\n".format(error), 500 worker = MakeWorker(hook) worker.start() - return 'OK\n', 200 + return "OK\n", 200 @app.before_first_request # FIXME this should be run on startup... @@ -179,28 +179,30 @@ def startup_actions(): def setup_logger(): - ''' Setup the default logger ''' + """ Setup the default logger """ coloredlogs.install( fmt="%(asctime)s [%(levelname)s] %(message)s", ) def check_settings(): - ''' Check the supplied settings ''' + """ Check the supplied settings """ if settings.DEBUG: - logging.warning('GogsMaker is running in DEBUG MODE, this is ' - 'unsuitable for production environments!') + logging.warning( + "GogsMaker is running in DEBUG MODE, this is " + "unsuitable for production environments!" + ) - required_keys = ['name', 'url', 'targets', 'secret'] + required_keys = ["name", "url", "targets", "secret"] for hook_id, hook in enumerate(settings.HOOKS): for key in required_keys: if key not in hook: - if key == 'name': - descr = '#{}'.format(hook_id) + if key == "name": + descr = "#{}".format(hook_id) else: - descr = '{} (#{})'.format(hook['name'], hook_id) + descr = "{} (#{})".format(hook["name"], hook_id) - logging.critical(('Configuration error: hook %s lacks ' - 'attribute %s.'), - descr, key) + logging.critical( + ("Configuration error: hook %s lacks " "attribute %s."), descr, key + ) sys.exit(1)