esss_jenkins.py

import json
import random
from fnmatch import fnmatch
from pprint import pformat
from textwrap import dedent
from urllib.parse import urlencode

import requests
from errbot import BotPlugin, botcmd, webhook, arg_botcmd


class JenkinsBot(BotPlugin):
    """Jenkins commands tailored to ESSS workflow"""

    class ResponseError(Exception):
        """
        Use this as an inner-class so it can easily be accessed by tests from the plugin object.
        errbot loads plugins dynamically, so to importing this in a test will result in a different
        class object than the one loaded by errbot.
        """
        def __init__(self, context, response):
            super().__init__('{}: {}'.format(context, response))
            self.response = response

    def get_configuration_template(self):
        return {
            'JENKINS_TOKEN': '',
            'JENKINS_USERNAME': '',
            'JENKINS_URL': 'https://eden.esss.com.br/jenkins',
            'ROCKETCHAT_USER': '',
            'ROCKETCHAT_PASSWORD': '',
            'ROCKETCHAT_DOMAIN': '',
        }

    def load_user_settings(self, user):
        key = 'user:{}'.format(user)
        settings = {
            'token': '',
            'jobs': [],
            'last_job_listing': [],
        }
        loaded = self.get(key, settings)
        settings.update(loaded)
        self.log.debug('LOAD ({}) settings: {}'.format(user, settings))
        return settings

    def save_user_settings(self, user, settings):
        key = 'user:{}'.format(user)
        self[key] = settings
        self.log.debug('SAVE ({}) settings: {}'.format(user, settings))

    def _generate_job_listing(self, job_names, user):
        items = []
        for (i, job_name) in enumerate(job_names):
            status = self._fetch_job_status(job_name)
            emoji = get_emoji_for_job_status(status)
            if status is not None:
                fmt = '`{index:2d}`. {emoji} [{name}]({url})'
            else:
                fmt = '`{index:2d}`. {emoji} {name}'
            items.append(fmt.format(
                index=i,
                name=job_name,
                url=self._get_job_url(job_name),
                emoji=emoji,
            ))
        settings = self.load_user_settings(user)
        settings['last_job_listing'] = job_names
        self.save_user_settings(user, settings)
        return items

    @botcmd(split_args_with=None, admin_only=True)
    def debug_settings(self, msg, args):
        if not args:
            return 'Enter user name'
        settings = self.load_user_settings(args[0])
        return 'User settings:\n```python\n{}```'.format(pformat(settings))

    @botcmd(split_args_with=None)
    def bhist(self, msg, args):
        """Returns a list with your job history, including running and previous runs."""
        user = args[0] if args else msg.frm.nick
        settings = self.load_user_settings(user)
        infos = settings['jobs']
        if not infos:
            return "You never ran anything. Or at least I don't remember."

        item_texts = self._generate_job_listing([info['job_name'] for info in infos], user)

        s = dedent("""\
            Here you go:
            
            {}
            
            To trigger builds, use `!build <num1> <num2> <num3> ...`
        """)
        return s.format('\n'.join(item_texts))

    @arg_botcmd('user', type=str)
    @arg_botcmd('--confirm', action='store_true')
    def clear(self, msg, user, confirm):
        """Clears your job history."""
        if not confirm:
            return 'Need to pass `--confirm` to this command'
        settings = self.load_user_settings(user)
        settings['jobs'] = []
        self.save_user_settings(user, settings)
        return 'Job history on the trash.'

    @botcmd(split_args_with=None)
    def find(self, msg, args):
        """Finds jobs based on keywords, separated by spaces (`ETK 1456 sci20 win64,linux64`)."""
        if not args:
            yield dedent("""\
                Pass some search factors, for example:
                    `ASIM 507 win64,linux64`: matches jobs with `ASIM` *and* `507` *and* ( `win64` *or* `linux64` )                    
                    `"eden-master-win64-35"`: matches exactly the job named `eden-master-win64-35`                    
            """)
            return
        self.log.debug('find from {}: {}'.format(msg.frm.nick, args))

        user = msg.frm.nick
        yield "Hold on, lemme check..."

        all_job_names = self._fetch_all_job_names()
        self.log.debug('found {} jobs in total, filtering by {!r}'.format(len(all_job_names), args))

        job_names = sorted(filter_jobs_by_find_string(all_job_names, args))
        if len(job_names) > 50:
            yield "This resulted in **{}** jobs, which is too much.\n" \
                  "Try to narrow your research.".format(len(job_names))
            return

        if job_names:
            self.log.debug('filtered {} jobs'.format(len(job_names)))
            items = self._generate_job_listing(job_names, user)
            yield dedent("""
                Found these {} jobs:
    
                {}
                
                To trigger builds, use `!build <num1> <num2> <num3> ...`
            """.format(len(items), '\n'.join(items)))
        else:
            yield "No jobs found, sorry buddy."

        settings = self.load_user_settings(user)
        settings['last_job_listing'] = job_names
        self.save_user_settings(user, settings)

    @botcmd(split_args_with=None)
    def build(self, msg, args):
        """Triggers jobs by an alias or build number from last `!find` or `!history` commands"""
        user = msg.frm.nick
        settings = self.load_user_settings(user)
        if not settings['token']:
            return NO_TOKEN_MSG.format(user=user, jenkins_url=self.config['JENKINS_URL'])

        def trigger_jobs(job_names, parameters):
            for job_name in job_names:
                self._trigger_job(job_name, user, parameters)

            items = ['[{job_name}]({url})'.format(job_name=x, url=self._get_job_url(x)) for x in job_names]
            return dedent("""
                Triggered **{}** jobs:
                
                {}
            """.format(len(items), '\n'.join(items)))

        aliases = settings.get('aliases')
        if aliases is not None and args and len(args) > 0:
            alias = args[0]
            if alias in aliases:
                search_pattern, parameters = aliases[alias]

                job_names = self._find_all_job_names_filtered(search_pattern + args[1:])
                if len(job_names) == 0:
                    return 'No job found with pattern: `{}`'.format(search_pattern + args[1:])

                if len(job_names) > 1:
                    msg = 'Multiple jobs found with pattern: `{}`'.format(search_pattern + args[1:])
                    for name in job_names[:5]:
                        msg += '\n - {}'.format(name)
                    return msg

                return trigger_jobs(job_names, parameters)

        if not settings['last_job_listing']:
            return dedent("""\
                No job listing yet, list jobs first with:
                    `!history`
                    `!find <word1> <word2> ...`                    
            """)

        try:
            indexes = [int(x) for x in args]
        except ValueError:
            return "Expected a list of indexes from the previous find command: `!build <word1> <word2> ...`"

        job_names = [x for (i, x) in enumerate(settings['last_job_listing']) if i in indexes]
        if not job_names:
            return "No jobs selected with those indexes."

        return trigger_jobs(job_names, None)

    @arg_botcmd('search_pattern', nargs='*', help='Job search pattern')
    @arg_botcmd('alias', nargs='?', help='Alias name')
    @arg_botcmd('--parameters', dest='parameters', help='Job parameters')
    def buildalias(self, msg, alias, search_pattern, parameters):
        """Adds a build alias based on keywords and parameters e.g: `!buildalias r30l rocky30 linux64 --parameters=BM='source'`)."""
        user = msg.frm.nick
        settings = self.load_user_settings(user)

        aliases = settings.get('aliases')
        if alias is None:
            if aliases:
                msg = 'Existing aliases:\n'
                for alias, (search_pattern, parameters) in aliases.items():
                    msg += ' - `{}: {} - {}`'.format(alias, search_pattern, parameters)
                return msg

            return dedent("""\
                Pass alias name, some search keywords, and an optional set of parameters, for example:
                    `rr30l rocky30 linux64 --parameters=BUILD_MODE=source&EDEN_SKIP_DEPS_TESTS=etk`
            """)

        if aliases is None:
            aliases = {}

        aliases[alias] = search_pattern, parameters
        settings['aliases'] = aliases
        self.save_user_settings(user, settings)

        return str('Alias registered: `{}: {} : {}`'.format(alias, search_pattern, parameters))


    @botcmd(split_args_with=None)
    def jenkins_token(self, msg, args):
        """Set or get your Jenkins token"""
        user = msg.frm.nick
        settings = self.load_user_settings(user)
        if not args:
            if settings['token']:
                return "You API Token is: `{}` (user: {})".format(settings['token'], user)
            else:
                return NO_TOKEN_MSG.format(user=user, jenkins_url=self.config['JENKINS_URL'])
        else:
            settings['token'] = args[0]
            self.save_user_settings(user, settings)
            return "Token saved."

    @webhook(raw=True)
    def jenkins(self, request):
        """
         {
         'number': '2',
         'job_name': 'fett-master-none-dev-ubuntu16.04-linux-sv01-ci01-execute_cmd',
         'timestamp': '1508516240981',
         'builtOn': 'dev-ubuntu16.04-linux-sv01-ci01',
         'event': 'jenkins.job.started',
         'userId': 'prusse',
         'url': 'job/fett-master-none-dev-ubuntu16.04-linux-sv01-ci01-execute_cmd/2/',
         }
        :param request:
        :return:
        """
        from rocketchat.api import RocketChatAPI

        rocket_api = RocketChatAPI(
            settings={
                'username': self.config['ROCKETCHAT_USER'],
                'password': self.config['ROCKETCHAT_PASSWORD'],
                'domain': self.config['ROCKETCHAT_DOMAIN'],
            }
        )
        self.log.debug('Jenkins: received request: {}'.format(pformat(dict(request.params))))
        info = dict(request.params)

        settings = self.load_user_settings(info['userId'])

        # move this job info to the first position
        infos = settings['jobs']
        for index, existing_info in enumerate(infos):
            if existing_info['job_name'] == info['job_name']:
                del infos[index]
        infos.insert(0, info)

        # limit number of jobs
        infos = infos[:15]
        settings['jobs'] = infos

        if info['event'] == 'jenkins.job.started':
            template = JOB_STARTED_MSG
            status = ':pray:'
            fmt_kwargs = {'comment': get_job_state_comment('STARTED')}
        else:
            template = JOB_COMPLETED_MSG
            status = ":white_check_mark:" if info['result'] == 'SUCCESS' else ":x:"
            comment = get_job_state_comment(info['result'])
            info['test_failures'] = self._get_build_test_errors(info['job_name'], info['number'])
            if info['test_failures']:
                max_show = 10
                failures_items = ['**{} failed tests**'.format(len(info['test_failures']))]
                failures_items += ['`{}`'.format(x['name']) for x in info['test_failures'][:max_show]]
                if len(info['test_failures']) > max_show:
                    failures_items.append('... and {} more'.format(len(info['test_failures']) - max_show))
                failures_msg = '\n'.join(failures_items)
            else:
                failures_msg = ''

            fmt_kwargs = {
                'test_failures_msg': failures_msg,
                'comment': comment,
            }

        info['status'] = status
        self.save_user_settings(info['userId'], settings)

        fmt_kwargs['jenkins_url'] = self.config['JENKINS_URL']
        fmt_kwargs.update(info)
        rocket_api.send_message(template.format(**fmt_kwargs).strip(), '@{}'.format(info['userId']))
        return 'OK'

    def _get_job_url(self, job_name):
        return '{}/job/{}'.format(self.config['JENKINS_URL'], job_name)

    def _get_build_test_errors(self, job_name, build_number):
        if build_number is None:
            build_number = 'lastBuild'

        url = 'job/{job_name}/{build_number}/testReport/api/json'.format(job_name=job_name, build_number=build_number)
        try:
            result = self._get_jenkins_json_request(url, params={'tree': 'suites[cases[name,status]]'})
        except self.ResponseError as e:
            if e.response.status_code == 404:
                return []
            raise
        try:
            cases = result['suites'][0]['cases']
        except:
            self.log.exception('Failed to get cases: {}'.format(url))
            return []

        errors = []
        for case in cases:
            if case['status'] not in ['PASSED', 'SKIPPED', 'FIXED']:
                errors.append(case)

        return errors

    def _fetch_all_job_names(self):
        result = self._get_jenkins_json_request('api/json', params={'tree': 'jobs[fullName]'})
        return [job['fullName'] for job in result['jobs']]

    def _find_all_job_names_filtered(self, args):
        all_job_names = self._fetch_all_job_names()
        self.log.debug('found {} jobs in total, filtering by {!r}'.format(len(all_job_names), args))

        return sorted(filter_jobs_by_find_string(all_job_names, args))

    def _get_jenkins_json_request(self, query_url, params=None):
        """
        returns None if fails to request
        """
        user = self.config['JENKINS_USERNAME']
        token = self.config['JENKINS_TOKEN']
        url = self.config['JENKINS_URL']
        if not url.endswith('/'):
            url += '/'
        url += query_url

        r = requests.get(url, auth=(user, token), params=params)
        if r.status_code not in [200, 201]:
            self.log.debug('_get_jenkins_json_request invalid response: {}'.format(r))
            raise self.ResponseError('json request to {url}'.format(url=url), r)
        return json.loads(r.text)

    def _fetch_job_status(self, job_name, build=None):
        """
        Fetch the status of the given job name:
        * 'SUCCESS'
        * 'FAILURE'
        * 'ABORTED'
        * 'UNSTABLE'
        * 'NOT_STARTED'
        * 'RUNNING'
        * None: does not exist
        """
        if build is None:
            build = 'lastBuild'
        try:
            response = self._get_jenkins_json_request(
                'job/{job_name}/{build}/api/json'.format(job_name=job_name, build=build),
                params={'tree': 'result'},
            )
        except self.ResponseError as e:
            # no result yet, check if the job exists then
            if e.response.status_code == 404:
                try:
                    self._get_jenkins_json_request('job/{job_name}/api/json'.format(job_name=job_name))
                    return 'NOT_STARTED'
                except self.ResponseError as e:
                    if e.response.status_code == 404:
                        return None
                    raise
            raise
        else:
            result = response.get('result')
            if result is None:
                # response like this: {"_class":"hudson.model.FreeStyleBuild","result":null}
                # means that there's a last build and it is currently running
                result = 'RUNNING'
            return result

    def _get_job_status_emoji(self, job_name):
        return get_emoji_for_job_status(self._fetch_job_status(job_name))

    def _get_job_parameters(self, job_name):
        query_url = 'job/{job_name}/api/json'.format(job_name=job_name)
        result = self._get_jenkins_json_request(query_url, params={'tree': 'actions[parameterDefinitions[name]]'})
        actions = result.get('actions')
        if actions:
            for action in actions:
                parameter_definitions = action.get('parameterDefinitions', [])
                return [p['name'] for p in parameter_definitions]
        return []

    def _get_job_builds(self, job_name):
        result = self._get_jenkins_json_request('job/{}/api/json'.format(job_name), params={'tree': 'builds[number]'})
        if result is None or 'builds' not in result:
            return None

        return [x['number'] for x in result['builds']]

    def _get_job_build_parameters_values(self, job_name, build_number='lastBuild'):
        query = 'job/{}/{}/api/json'.format(job_name, build_number)
        result = self._get_jenkins_json_request(query, params={'tree': 'actions[parameters[name,value]]'})
        actions = result.get('actions', [])
        for action in actions:
            parameters = action.get('parameters')
            if parameters:
                return [(p['name'], p['value']) for p in parameters]
        return []

    def _post_jenkins_json_request(self, post_url, user):
        jenkins_url = self.config['JENKINS_URL']
        token = self.load_user_settings(user)['token']

        if not token:
            raise RuntimeError('Token for user {} not configured'.format(user))

        if not jenkins_url.endswith('/'):
            jenkins_url += '/'

        post_url = jenkins_url + post_url
        r = requests.post(post_url, auth=(user, token))
        self.log.debug('post_jenkins_json_request: url {} = {}'.format(post_url, r.status_code))
        if r.status_code not in (200, 201):
            raise self.ResponseError('Error posting to {url}: {r}\n{text}'.format(url=post_url, r=r, text=r.text), r)

    def _trigger_job(self, job_name, user, parameters=None):
        if parameters is not None:
            post_url = 'job/{}/buildWithParameters?{}'.format(job_name, parameters)
            self._post_jenkins_json_request(post_url, user)

        else:
            builds = self._get_job_builds(job_name)
            parameters = self._get_job_parameters(job_name)

            never_built = not builds
            takes_parameters = bool(parameters)
            if never_built or not takes_parameters:
                post_url = 'job/{}/build' if not parameters else 'job/{}/buildWithParameters'
                post_url = post_url.format(job_name)
                self._post_jenkins_json_request(post_url, user)
            elif takes_parameters:
                parameters_values = self._get_job_build_parameters_values(job_name)
                post_url = 'job/{}/buildWithParameters?{}'.format(job_name, urlencode(parameters_values))
                self._post_jenkins_json_request(post_url, user)


def get_job_state_comment(key):
    comments = COMMENTS.get(key, [key])
    return random.choice(comments)


def get_emoji_for_job_status(result):
    return {
        'SUCCESS': ":white_check_mark:",
        'FAILURE': ":x:",
        'ABORTED': ":white_circle:",
        'NOT_STARTED': ":white_circle:",
        'UNSTABLE': ":warning:",
        'RUNNING': ":arrows_counterclockwise:",
        None: ":grey_question:",
    }.get(result, result)


def filter_jobs_by_find_string(job_names, input_factors):
    factors = []
    for factor in input_factors:
        if factor.startswith('"') and factor.endswith('"'):
            word = factor[1:-1].lower()
            job_names = [x for x in job_names if fnmatch(x.lower(), word)]
        else:
            factor = factor.lower()
            factors.extend(factor.split('-'))

    and_factors = []
    or_factors = []
    for factor in factors:
        if ',' in factor:
            or_factors.extend(factor.split(','))
        else:
            and_factors.append(factor)

    def matches(fields):
        fields = {x.lower() for x in fields}
        for test_field in and_factors:
            if not any(fnmatch(x, test_field) for x in fields):
                return False
        if not or_factors:
            return True
        for or_factor in or_factors:
            if any(fnmatch(x, or_factor) for x in fields):
                return True

        return False

    return [x for x in job_names if matches(set(x.split('-')))]


COMMENTS = {
    'STARTED': [
        'Now we wait... :popcorn:',
        "Hope it won't take long... ",
        "Coffee maybe? :coffee:",
        "Time to take a break? :smoking:",
    ],
    'SUCCESS': [
        "Way to go! :beer:",
        "Nice! :champagne_glass:",
        "Hooray! :wine_glass:",
    ],
    'FAILURE': [
        "No good...",
        "Tsk tsk tsk...",
        "I feel bad for you... well not really.",
        "Flaky, perhaps?",
    ],
}

NO_TOKEN_MSG = """
**Jenkins API Token not configured**. 
Find your API Token [here]({jenkins_url}/user/{user}/configure) (make sure you are logged in) and execute:

    `!jenkins token <TOKEN>` 

This only needs to be done once.
"""

JOB_STARTED_MSG = '''
**Job Started**!
{status} [{job_name}]({jenkins_url}/{url}) build **{number}**
Building on: **{builtOn}**
'''

JOB_COMPLETED_MSG = '''
**Job Completed**!
{status} [{job_name}]({jenkins_url}/{url}) build **{number}**
{test_failures_msg}
'''