gitea-redmine/gitea_redmine.py

import json
import logging
import os
import re

import flask
import redminelib

logging.basicConfig(
    level=logging.INFO, format='%(asctime)s [%(levelname)s] %(message)s', handlers=[logging.StreamHandler()]
)
INCOMING_WEBHOOK_SECRET = os.environ['INCOMING_WEBHOOK_SECRET']
REDMINE_URL = os.environ['REDMINE_URL']
REDMINE_API_KEY = os.environ['REDMINE_API_KEY']

REDMINE_CLIENT = redminelib.Redmine(REDMINE_URL, key=REDMINE_API_KEY, raise_attr_exception=False)
REDMINE_ASSIGNABLE_GROUP = int(os.environ.get('REDMINE_ASSIGNABLE_GROUP', 21))
REDMINE_EXCLUDED_PROJECTS = [
    i.strip() for i in os.environ.get('REDMINE_EXCLUDED_PROJECTS', 'projets-clients').split(',')
]

REDMINE_STATUSES = {
    'En cours': 2,
    'Résolu': 3,
    'Solution proposée': 12,
    'Solution validée': 13,
    'Solution déployée': 4,
    'Fermé': 5,
    'Rejeté': 6,
}
CLOSED_STATUSES = [
    REDMINE_STATUSES['Résolu'],
    REDMINE_STATUSES['Solution déployée'],
    REDMINE_STATUSES['Fermé'],
    REDMINE_STATUSES['Rejeté'],
]

REDMINE_TRACKERS = {
    'Développement': 2,
    'Bug': 1,
    'Documentation': 6,
}


app = flask.Flask(__name__)


@app.route('/incoming-webhook/<token>', methods=['POST'])
def incoming_webhook(token):
    if token != INCOMING_WEBHOOK_SECRET:
        logging.warning('Received invalid token')
        return {'status': 'error', 'detail': 'Invalid token'}, 403

    payload = flask.request.json
    logging.info('Received payload %s', json.dumps(payload, indent=2))
    handler, event = get_handler(payload)
    if not handler:
        logging.info('Skipping, no handler found for this webhook')
        return {'status': 'success', 'detail': 'Skipped, unhandled webhook'}

    haystack = payload['pull_request']['title']
    try:
        pr_branch = payload['pull_request']['head']['ref']
    except KeyError:
        pr_branch = ''

    if pr_branch.startswith('hotfix/'):
        logging.info('Skipping, hotfix branch.')
        return {'status': 'success', 'detail': 'Skipped, hotfix branch'}

    issues_ids = get_issues_ids(haystack, pr_branch)

    issues = []
    for id in issues_ids:
        try:
            issues.append(get_redmine_issue(id))
        except redminelib.exceptions.ResourceNotFoundError:
            logging.warning(f'Unknown redmine issue {id}')

    if not issues:
        return {'status': 'success', 'detail': 'Skipped, no valid redmine issues linked'}

    for issue in issues:
        project = get_redmine_project(issue.project.id)
        logging.info(f'Calling handler {event} for issue {issue.id}…')
        handler(issue, payload, project)
        logging.info('Done')

    return {'status': 'success', 'detail': 'Event processed and forwarded to redmine'}


def get_handler(payload):
    if payload.get('action') == 'opened' and payload.get('pull_request'):
        if payload['pull_request']['title'].lower().startswith('wip:'):
            return handle_pull_request_opened_draft, 'pull_request.opened'
        else:
            return handle_pull_request_opened, 'pull_request.opened'

    if payload.get('action') == 'reviewed':
        if payload['review']['type'] == 'pull_request_review_rejected':
            return handle_pull_request_rejected, 'pull_request.rejected'
        if payload['review']['type'] == 'pull_request_review_approved':
            return handle_pull_request_approved, 'pull_request.approved'
        return None, None

    if payload.get('action') == 'review_requested':
        return handle_pull_request_review_requested, 'pull_request.review_requested'

    if payload.get('action') == 'edited' and payload.get('pull_request'):
        return handle_pull_request_edited, 'pull_request.edited'

    if (
        payload.get('action') == 'closed'
        and payload.get('pull_request')
        and payload['pull_request'].get('merged')
    ):
        return handle_pull_request_merged, 'pull_request.merged'

    if payload.get('action') == 'closed' and payload.get('pull_request'):
        return handle_pull_request_closed, 'pull_request.closed'

    return None, None


ISSUE_REGEX = re.compile(r'#(\d+)')
PR_BRANCH_REGEX = re.compile(r'wip/(\d+).*')


def get_issues_ids(text, branch_name):
    """Extract issues number from their #1234 notation or the wip/xxx branch name"""
    ids = [int(i) for i in ISSUE_REGEX.findall(text)]
    ids += [int(i) for i in PR_BRANCH_REGEX.findall(branch_name or '')]

    return list(sorted(set(ids)))


def noop(payload):
    pass


def get_redmine_user(username):
    for user in REDMINE_CLIENT.user.filter(group_id=REDMINE_ASSIGNABLE_GROUP):
        # 21 is the Entrouvert group
        # We cannot easily find a redmine user through the API using
        # only its username. We need to loop on possible candida
        if user.login == username:
            return user
    raise Exception(f'No redmine user found for username {username} in group {REDMINE_ASSIGNABLE_GROUP}')


def get_redmine_issue(id):
    return REDMINE_CLIENT.issue.get(id, includes=['journals'])


def get_redmine_project(id):
    return REDMINE_CLIENT.project.get(id)


class Abort(Exception):
    pass


def is_excluded_project(project):
    # easiest case, the project itself is excluded
    if project.identifier in REDMINE_EXCLUDED_PROJECTS:
        return True
    # now, if the project has a parent, we must recursively check
    # if the parent is excluded as well
    if hasattr(project, 'parent') and project.parent and project.parent.id:
        # since some project may have grandparents, we must still check for this case
        parent = get_redmine_project(project.parent.id)
        return is_excluded_project(parent)
    return False


def make_handler(*actions):
    def inner(issue, payload, project):
        if is_excluded_project(project):
            logging.info('Issue belongs to excluded project %s', project.id)
            return False
        flat_payload = flatten(payload)
        for action in actions:
            try:
                action(issue, flat_payload)
            except Abort:
                return False
        return True

    return inner


def flatten(d, parent_key='', sep='_'):
    items = []
    for k, v in d.items():
        new_key = parent_key + sep + k if parent_key else k
        if isinstance(v, dict):
            items.extend(flatten(v, new_key, sep=sep).items())
        else:
            items.append((new_key, v))
    return dict(items)


def assign_to(user_field):
    def inner(issue, payload):
        redmine_user = get_redmine_user(payload[user_field])
        issue.assigned_to_id = redmine_user.id

    return inner


def set_status(status_id, unless=[]):
    def inner(issue, payload):
        current_status = issue.status.id
        if current_status not in unless:
            if issue.tracker.id not in list(REDMINE_TRACKERS.values()):
                issue.tracker_id = REDMINE_TRACKERS['Développement']
            else:
                # always assigne value, to get tracker_id attribute on the object
                # for the unit tests.
                issue.tracker_id = issue.tracker.id
            issue.status_id = status_id

    return inner


def add_note(template):
    def inner(issue, payload):
        issue.notes = template.format(**payload)

    return inner


def save():
    def inner(issue, payload):
        issue.save()

    return inner


def skip_if_not_draft(issue, payload):
    if not payload['pull_request_title'].lower().startswith('wip:'):
        raise Abort('Not a WIP: PR')


def skip_if_notes_include(field, reason):
    def inner(issue, payload):
        for entry in issue.journals:
            if payload[field] in entry.notes:
                raise Abort(reason)

    return inner


def skip_unless_notes_include(text, reason):
    def inner(issue, payload):
        for entry in issue.journals:
            if text in entry.notes:
                break
        else:
            raise Abort(reason)

    return inner


def skip_if_status(status):
    def inner(issue, payload):
        if issue.status.id == status:
            raise Abort(f'Issue already in status {status}')

    return inner


def skip_if_not_status(status):
    def inner(issue, payload):
        if issue.status.id != status:
            raise Abort(f'Issue not in status {status}')

    return inner


def set_value_if_empty(field, value):
    def inner(issue, payload):
        if not payload.get(field):
            payload[field] = value

    return inner


PULL_REQUEST_OPENED_NOTE = '{pull_request_user_full_name} ({pull_request_user_username}) a ouvert une pull request sur Gitea concernant cette demande :\n\n* URL : {pull_request_url}\n* Titre : {pull_request_title}\n* Modifications : {pull_request_url}/files'

handle_pull_request_opened = make_handler(
    assign_to('pull_request_user_username'),
    set_status(REDMINE_STATUSES['Solution proposée'], unless=CLOSED_STATUSES),
    add_note(PULL_REQUEST_OPENED_NOTE),
    save(),
)
handle_pull_request_opened_draft = make_handler(
    assign_to('pull_request_user_username'),
    set_status(REDMINE_STATUSES['En cours'], unless=CLOSED_STATUSES),
    add_note(PULL_REQUEST_OPENED_NOTE),
    save(),
)


def handle_pull_request_edited(issue, payload, project):
    if payload['pull_request']['title'].lower().startswith('wip:'):
        func = make_handler(
            skip_if_status(REDMINE_STATUSES['En cours']),
            skip_if_not_draft,
            set_status(REDMINE_STATUSES['En cours'], unless=CLOSED_STATUSES),
            add_note(
                '{pull_request_user_full_name} ({pull_request_user_username}) a commencé à travailler sur une pull request sur Gitea concernant cette demande :\n\n* URL : {pull_request_url}\n* Titre : {pull_request_title}\n* Modifications : {pull_request_url}/files'
            ),
            save(),
        )
        return func(issue, payload, project)

    # maybe it was wip and is no more
    func = make_handler(
        skip_if_not_status(REDMINE_STATUSES['En cours']),
        skip_unless_notes_include('Titre : WIP', 'Pull request do not mention a draft title'),
        set_status(REDMINE_STATUSES['Solution proposée']),
        save(),
    )
    if func(issue, payload, project) is True:
        # the issue was changed, stop now.
        return

    # default behaviour
    func = make_handler(
        skip_if_notes_include('pull_request_url', 'Pull request already linked in issue'),
        assign_to('pull_request_user_username'),
        set_status(REDMINE_STATUSES['En cours'], unless=CLOSED_STATUSES),
        add_note(
            '{pull_request_user_full_name} ({pull_request_user_username}) a lié une pull request sur Gitea concernant cette demande :\n\n* URL : {pull_request_url}\n* Titre : {pull_request_title}\n* Modifications : {pull_request_url}/files'
        ),
        save(),
    )
    return func(issue, payload, project)


handle_pull_request_merged = make_handler(
    set_status(REDMINE_STATUSES['Résolu'], unless=CLOSED_STATUSES),
    add_note(
        '{sender_full_name} ({sender_username}) a mergé une pull request sur Gitea concernant cette demande :\n\n* URL : {pull_request_url}\n* Titre : {pull_request_title}\n* Modifications : {pull_request_url}/files'
    ),
    save(),
)

handle_pull_request_approved = make_handler(
    skip_if_status(REDMINE_STATUSES['Solution validée']),
    set_status(REDMINE_STATUSES['Solution proposée'], unless=CLOSED_STATUSES),
    save(),
    set_status(REDMINE_STATUSES['Solution validée'], unless=CLOSED_STATUSES),
    add_note(
        '{sender_full_name} ({sender_username}) a approuvé une pull request sur Gitea concernant cette demande :\n\n* URL : {pull_request_url}'
    ),
    save(),
)

handle_pull_request_rejected = make_handler(
    set_status(REDMINE_STATUSES['En cours'], unless=CLOSED_STATUSES),
    add_note(
        '{sender_full_name} ({sender_username}) a relu et demandé des modifications sur une pull request sur Gitea concernant cette demande :\n\n* URL : {pull_request_url}'
    ),
    save(),
)

handle_pull_request_closed = make_handler(
    set_status(REDMINE_STATUSES['En cours'], unless=CLOSED_STATUSES),
    add_note(
        '{sender_full_name} ({sender_username}) a fermé une pull request sur Gitea concernant cette demande.'
    ),
    save(),
)


handle_pull_request_review_requested = make_handler(
    set_status(REDMINE_STATUSES['Solution proposée'], unless=CLOSED_STATUSES),
    add_note(
        '{sender_full_name} ({sender_username}) a demandé une relecture de {requested_reviewer_full_name} ({requested_reviewer_username}) sur une pull request sur Gitea concernant cette demande :\n\n* URL : {pull_request_url}'
    ),
    save(),
)


@app.route('/redmine-issue/<issue_id>/', methods=['GET'])
def redmine_issue(issue_id):
    try:
        issue = get_redmine_issue(issue_id)
    except redminelib.exceptions.ResourceNotFoundError:
        logging.warning('Unknown redmine issue %s', issue_id)
        return {'status': 'error', 'detail': 'Not found'}, 404
    if issue.is_private:
        logging.warning('Private redmine issue %s', issue_id)
        return {'status': 'error', 'detail': 'Private issue'}, 403
    return {'status': 'success', 'subject': issue.subject}