Apply black

This commit is contained in:
Théophile Bastian 2021-09-01 00:00:42 +02:00
parent 8e77a9a676
commit f5b5bc8743

View file

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