"""Web module
This module holds web application logic.
"""
import click
import flask
import os
import sys
import time
from .constants import NO_GH_TOKEN_RETURN, NO_REPOS_SPEC_RETURN
from .constants import NO_WEBHOOK_SECRET_RETURN
from .helpers import create_config, extract_repos
from .github import GitHub, GitHubError
[docs]class LabelordChange:
"""Labelord Change
This class describes one change within web application and provides
method to check its validity. Initializing an instance will set its
action, name, color and old_name. It will also set the timestamp to
current time.
Example:
.. testcode::
import time
import labelord
change = labelord.web.LabelordChange('created', 'Bug', 'FF0000')
.. doctest::
>>> change.action
'created'
>>> change.name
'Bug'
>>> change.color
'FF0000'
>>> change.old_name
>>> type(change.timestamp)
<class 'int'>
Args:
action (str): Action performed (created, edited, deleted).
name (str): Label name.
color (str): Label color.
new_name (Optional[str]): New label name, default: None.
"""
CHANGE_TIMEOUT = 10
"""int: Timeout for a change."""
def __init__(self, action, name, color, new_name=None):
self.action = action
self.name = name
self.color = None if action == 'deleted' else color
self.old_name = new_name
self.timestamp = int(time.time())
@property
def tuple(self):
"""Formats change into a tuple
This method makes a representation of the change in the form of a tuple.
Example:
.. testcode::
import time
import labelord
change = labelord.web.LabelordChange('created', 'Bug', 'FF0000')
.. doctest::
>>> change.tuple
('created', 'Bug', 'FF0000', None)
Returns:
tuple: Labelord change (action, label name, label color, label new name).
"""
return self.action, self.name, self.color, self.old_name
def __eq__(self, other):
return self.tuple == other.tuple
[docs] def is_valid(self):
"""Tests change validity
This method checks whether given change is valid using timeout.
Example:
.. testcode::
import time
import labelord
change = labelord.web.LabelordChange('created', 'Bug', 'FF0000')
.. doctest::
>>> change.is_valid()
True
Returns:
bool: Whether the change is valid.
"""
return self.timestamp > (int(time.time()) - self.CHANGE_TIMEOUT)
[docs]class LabelordWeb(flask.Flask):
"""Labelord web application
This class holds web application logic. Initializing an instance will load
application's configuration, obtain GitHub class instance for communicating
with GitHub API and initialize a dictionary of ignores (valid changes for
given repository). It will also call the base class' constructor passing
any additional arguments.
Example:
.. testcode::
import labelord
app = labelord.web.LabelordWeb(None, None, import_name = '')
.. doctest::
>>> app.ignores
{}
Args:
labelord_config (configparser.ConfigParser): Parsed configuration object.
github (GitHub): GitHub class instance.
args (list): Additional arguments for base class.
kwargs (dict): Additional keyword arguments for base class.
"""
def __init__(self, labelord_config, github, *args, **kwargs):
super().__init__(*args, **kwargs)
self.labelord_config = labelord_config
self.github = github
self.ignores = {}
[docs] def inject_session(self, session):
"""Sets new session for communitation with GitHub API
This method passes new session to the GitHub class instance.
Args:
session (requests.Session): Session to inject.
"""
self.github.set_session(session)
[docs] def reload_config(self):
"""Reloads application's configuration
This method reloads application's configuration including GitHub
API token.
"""
config_filename = os.environ.get('LABELORD_CONFIG', None)
self.labelord_config = create_config(
token=os.getenv('GITHUB_TOKEN', None),
config_filename=config_filename
)
self._check_config()
self.github.token = self.labelord_config.get('github', 'token')
@property
def repos(self):
"""Gets GitHub repositories
This method extracts GitHub repositories from the application's
configuration.
Returns:
list: List of GitHub repositories.
"""
return extract_repos(flask.current_app.labelord_config)
def _check_config(self):
if not self.labelord_config.has_option('github', 'token'):
click.echo('No GitHub token has been provided', err=True)
sys.exit(NO_GH_TOKEN_RETURN)
if not self.labelord_config.has_section('repos'):
click.echo('No repositories specification has been found',
err=True)
sys.exit(NO_REPOS_SPEC_RETURN)
if not self.labelord_config.has_option('github', 'webhook_secret'):
click.echo('No webhook secret has been provided', err=True)
sys.exit(NO_WEBHOOK_SECRET_RETURN)
def _init_error_handlers(self):
from werkzeug.exceptions import default_exceptions
for code in default_exceptions:
self.errorhandler(code)(LabelordWeb._error_page)
[docs] def finish_setup(self):
"""Finalizes application's setup
This method checks application's configuration and performs
error handlers initialization.
"""
self._check_config()
self._init_error_handlers()
[docs] @staticmethod
def create_app(config=None, github=None):
"""Creates web application
This method performs configuration loading including GitHub API token and
creates labelord web application.
Example:
.. testcode::
import labelord
app = labelord.web.LabelordWeb.create_app()
.. doctest::
>>> type(app)
<class 'labelord.web.LabelordWeb'>
Args:
config (configparser.ConfigParser): Parsed configuration object.
github (GitHub): GitHub class instance.
Returns:
LabelordWeb: labelord web application instance
"""
cfg = config or create_config(
token=os.getenv('GITHUB_TOKEN', None),
config_filename=os.getenv('LABELORD_CONFIG', None)
)
gh = github or GitHub('') # dummy, but will be checked later
gh.token = cfg.get('github', 'token', fallback='')
return LabelordWeb(cfg, gh, import_name=__name__)
@staticmethod
def _error_page(error):
return flask.render_template('error.html', error=error), error.code
[docs] def cleanup_ignores(self):
"""Cleans up invalid changes
This method checks lists of changes for every GitHub repository and
removes those changes which are not valid.
Example:
.. testcode::
import labelord
valid_change = labelord.web.LabelordChange('created', 'Bug', 'FF0000')
invalid_change = labelord.web.LabelordChange('deleted', 'Bug', 'FF0000')
invalid_change.timestamp -= 20
app = labelord.web.LabelordWeb(None, None, import_name = '')
app.ignores['User/Repo'] = [valid_change, invalid_change]
app.cleanup_ignores()
.. doctest::
>>> len(app.ignores['User/Repo'])
1
>>> app.ignores['User/Repo'][0].action
'created'
"""
for repo in self.ignores:
self.ignores[repo] = [c for c in self.ignores[repo]
if c.is_valid()]
[docs] def process_label_webhook_create(self, label, repo):
"""Processes 'created' change
This method processes a create label action made by the user.
Args:
label (dict): Created label's name and color.
repo (str): Repository to create label in.
"""
self.github.create_label(repo, label['name'], label['color'])
[docs] def process_label_webhook_delete(self, label, repo):
"""Processes 'deleted' change
This method processes a remove label action made by the user.
Args:
label (dict): Deleted label's name and color.
repo (str): Repository to delete label from.
"""
self.github.delete_label(repo, label['name'])
[docs] def process_label_webhook_edit(self, label, repo, changes):
"""Processes 'edited' change
This method processes an update label action made by the user.
Args:
label (dict): Updated label's new name and color.
repo (str): Repository to update label in.
changes (dict): Contains updated label's old name.
"""
name = old_name = label['name']
color = label['color']
if 'name' in changes:
old_name = changes['name']['from']
self.github.update_label(repo, name, color, old_name)
[docs] def process_label_webhook(self, data):
"""Processes user action on label
This method processes webhook and extracts change made by the user.
Then it passes this change for further processing.
Args:
data (dict): Webhook data.
"""
self.cleanup_ignores()
action = data['action']
label = data['label']
repo = data['repository']['full_name']
flask.current_app.logger.info(
'Processing LABEL webhook event with action {} from {} '
'with label {}'.format(action, repo, label)
)
if repo not in self.repos:
return # This repo is not being allowed in this app
change = LabelordChange(action, label['name'], label['color'])
if action == 'edited' and 'name' in data['changes']:
change.new_name = label['name']
change.name = data['changes']['name']['from']
if repo in self.ignores and change in self.ignores[repo]:
self.ignores[repo].remove(change)
return # This change was initiated by this service
for r in self.repos:
if r == repo:
continue
if r not in self.ignores:
self.ignores[r] = []
self.ignores[r].append(change)
try:
if action == 'created':
self.process_label_webhook_create(label, r)
elif action == 'deleted':
self.process_label_webhook_delete(label, r)
elif action == 'edited':
self.process_label_webhook_edit(label, r, data['changes'])
except GitHubError:
pass # Ignore GitHub errors
app = LabelordWeb.create_app()
[docs]@app.before_first_request
def finalize_setup():
"""Finalizes application's setup
This function makes sure that application's setup is finalized before the
first request.
"""
flask.current_app.finish_setup()
[docs]@app.route('/', methods=['GET'])
def index():
"""Processes GET request
This function processes any GET requests incoming to '/' and renders
HTML template with a list of GitHub repositories.
"""
repos = flask.current_app.repos
return flask.render_template('index.html', repos=repos)
[docs]@app.route('/', methods=['POST'])
def hook_accept():
"""Processes POST request
This function processes any POST requests incoming to '/', verifies
webhook signature, extracts webhook events and passes them for further
processing.
Returns:
str: Empty string.
"""
headers = flask.request.headers
signature = headers.get('X-Hub-Signature', '')
event = headers.get('X-GitHub-Event', '')
data = flask.request.get_json()
if not flask.current_app.github.webhook_verify_signature(
flask.request.data, signature,
flask.current_app.labelord_config.get('github', 'webhook_secret')
):
flask.abort(401)
if event == 'label':
if data['repository']['full_name'] not in flask.current_app.repos:
flask.abort(400, 'Repository is not allowed in application')
flask.current_app.process_label_webhook(data)
return ''
if event == 'ping':
flask.current_app.logger.info('Accepting PING webhook event')
return ''
flask.abort(400, 'Event not supported')