tcp_tests/managers/jenkins/client.py

from __future__ import print_function
import datetime
import time

import jenkins
import json
import yaml
import requests
import re

from devops.helpers import helpers

from requests.exceptions import ConnectionError


def retry(max_count=6,
          sleep_before_retry=10):
    def _retry(func):
        def __retry(*args, **kwargs):
            """
            Waits some time and retries the requests if it fails with
            with any error
            Raises Exceptions after all unsuccessful tries

            :param func: callable
            :param args, kwargs: parameters of decorated functions
            :param max_count: times of retries
            :param sleep_before_retry: how many seconds needs to wait before
            the next retry
            :return: response
            :raise ConnectionError after several unsuccessful connections
            """
            err_msg = None
            for count in range(max_count):
                try:
                    return func(*args, **kwargs)
                except Exception as err:
                    err_msg = err
                    print("Try {count}/{max_count} caught "
                          "Exception in {fn}: {err}."
                          "\n... repeat after {secs} secs".
                          format(err=err,
                                 count=count+1,
                                 max_count=max_count,
                                 secs=sleep_before_retry,
                                 fn=func.__name__))
                time.sleep(sleep_before_retry)
            print("Function failed in {total_time} seconds".format(
                total_time=max_count*sleep_before_retry)
            )
            raise err_msg
        return __retry
    return _retry


@retry(max_count=5, sleep_before_retry=2)
def send_request(action, endpoint):
    """
    Makes request with described operation to endpoint
    :param action: string, type of operation GET, POST, DELETE and so on
    :param endpoint: string, url to send request
    :return:  response
    """
    response = requests.Request(action, endpoint)
    return response


class JenkinsClient(object):

    def __init__(self, host=None, username='admin', password='r00tme'):
        host = host or 'http://172.16.44.33:8081'
        self.__client = jenkins.Jenkins(
            host,
            username=username,
            password=password)
        self.__client._session.verify = False

    @retry()
    def jobs(self):
        return self.__client.get_jobs()

    def find_jobs(self, name):
        return filter(lambda x: name in x['fullname'], self.jobs())

    @retry()
    def job_info(self, name):
        return self.__client.get_job_info(name)

    def list_builds(self, name):
        return self.job_info(name).get('builds')

    @retry()
    def build_info(self, name, build_id):
        return self.__client.get_build_info(name, build_id)

    def job_params(self, name):
        job = self.job_info(name)
        job_params = next(
            p for p in job['property'] if
            'hudson.model.ParametersDefinitionProperty' == p['_class'])
        job_params = job_params['parameterDefinitions']
        return job_params

    def make_defaults_params(self, name):
        job_params = self.job_params(name)
        def_params = dict(
            [(j['name'], j['defaultParameterValue']['value'])
             for j in job_params])
        return def_params

    def _correct_yaml_params(self, job_name, params):
        """
        Params can be defined as a nested dict.
        In that case 2nd-layer dict will be translated to YAML text and
        added to default parameter value

        :param job_name: Job name
        :param params: dict of JenkinsJobs parameters
        :return: nothing
        """
        for param_name, param_value in params.items():
            if not isinstance(param_value, dict):
                continue
            default_param = self.make_defaults_params(job_name).get(param_name)
            if default_param is None:
                print("{param} param of {job} job doesn't exist. "
                      "Ignoring enriching it with {value}".format(
                        param=param_name,
                        job=job_name,
                        value=param_value
                        ))
                continue
            yaml_param = yaml.load(default_param)
            yaml_param.update(param_value)
            params[param_name] = yaml.dump(yaml_param,
                                           default_flow_style=False)
        return params

    @retry()
    def run_build(self, name, params=None, timeout=600, verbose=False):
        params = params or self.make_defaults_params(name)
        params = self._correct_yaml_params(job_name=name,
                                           params=params)

        num = self.__client.build_job(name, params)
        time.sleep(2)  # wait while job is started

        def is_build_queued():
            try:
                item = self.__client.get_queue_item(num)
                ts = item['inQueueSince'] / 1000
                since_time = datetime.datetime.fromtimestamp(ts)
                print("Build in the queue since {}".format(since_time))
                return True
            except jenkins.JenkinsException:
                if verbose:
                    print("Build have not been queued {} yet".format(num))

        helpers.wait(
            is_build_queued,
            timeout=timeout,
            interval=30,
            timeout_msg='Timeout waiting to queue the build '
                        'for {} job'.format(name))

        def is_blocked():
            queued = self.__client.get_queue_item(num)
            status = not queued['blocked']
            if not status and verbose:
                print("pending the job [{}] : {}".format(name, queued['why']))
            return (status and
                    'executable' in (queued or {}) and
                    'number' in (queued['executable'] or {}))

        helpers.wait(
            is_blocked,
            timeout=timeout,
            interval=30,
            timeout_msg='Timeout waiting to run the job [{}]'.format(name))
        build_id = self.__client.get_queue_item(num)['executable']['number']

        def is_build_started():
            try:
                build = self.__client.get_build_info(name, build_id)
                ts = float(build['timestamp']) / 1000
                start_time = datetime.datetime.fromtimestamp(ts)
                print("the build {} in {} have started at {} UTC".format(
                    build_id, name, start_time))
                return True
            except jenkins.JenkinsException:
                if verbose:
                    print("the build {} in {} have not strated yet".format(
                        build_id, name))
        helpers.wait(
            is_build_started,
            timeout=timeout,
            interval=30,
            timeout_msg='Timeout waiting to run build of '
                        'the job [{}]'.format(name))

        return name, build_id

    def wait_end_of_build(self, name, build_id, timeout=600, interval=5,
                          verbose=False, job_output_prefix=''):
        '''Wait until the specified build is finished

        :param name: ``str``, job name
        :param build_id: ``int``, build id
        :param timeout: ``int``, timeout waiting the job, sec
        :param interval: ``int``, interval of polling the job result, sec
        :param verbose: ``bool``, print the job console updates during waiting
        :param job_output_prefix: ``str``, print the prefix for each console
                                  output line, with the pre-defined
                                  substitution keys:
                                  - '{name}' : the current job name
                                  - '{build_id}' : the current build-id
                                  - '{time}' : the current time
        :returns: requests object with headers and console output,  ``obj``
        '''
        start = [0]
        time_str = time.strftime("%H:%M:%S")
        prefix = "\n" + job_output_prefix.format(job_name=name,
                                                 build_number=build_id,
                                                 time=time_str)
        if verbose:
            print(prefix, end='')

        def building():
            try:
                # Nested retry decorator. Need to wait >30 min
                # During mcp-upgrade job the Jenkins can being upgrading
                # and can be inaccessible for >20 min
                status = not retry(max_count=30)(
                    self.build_info)(name, build_id)['building']
            except ConnectionError:
                status = False

            if verbose:
                time_str = time.strftime("%H:%M:%S")
                prefix = "\n" + job_output_prefix.format(
                    job_name=name, build_number=build_id, time=time_str)
                res = self.get_progressive_build_output(name,
                                                        build_id,
                                                        start=start[0])
                if 'X-Text-Size' in res.headers:
                    text_size = int(res.headers['X-Text-Size'])
                    if start[0] < text_size:
                        text = res.content.decode('utf-8',
                                                  errors='backslashreplace')
                        print(text.replace("\n", prefix), end='')
                        start[0] = text_size
            return status

        helpers.wait(
            building,
            timeout=timeout,
            interval=interval,
            timeout_msg=('Timeout waiting the job {0}:{1} in {2} sec.'
                         .format(name, build_id, timeout)))

    @retry()
    def get_build_output(self, name, build_id):
        output = self.__client.get_build_console_output(name, build_id)
        # Clean output from any info added by addons
        result = ''
        for line in output.splitlines():
            result += re.sub(r'<span.+</span>', '', line)
        return result

    @retry(max_count=20, sleep_before_retry=30)
    def get_progressive_build_output(self, name, build_id, start=0):
        '''Get build console text.

        :param name: Job name, ``str``
        :param build_id: Build id, ``int``
        :param start: Start offset, ``int``
        :returns: requests object with headers and console output,  ``obj``
        '''
        folder_url, short_name = self.__client._get_job_folder(name)

        PROGRESSIVE_CONSOLE_OUTPUT = (
            '%(folder_url)sjob/%(short_name)s/%(build_id)d/'
            'logText/progressiveText?start=%(start)d')
        req = send_request(
                'GET',
                self.__client._build_url(PROGRESSIVE_CONSOLE_OUTPUT, locals()))
        return (self.__client.jenkins_request(req))

    @retry(max_count=12)
    def get_workflow(self, name, build_id, enode=None, mode='describe'):
        '''Get workflow results from pipeline job

        :param name: job name
        :param build_id: str, build number or 'lastBuild'
        :param enode: int, execution node in the workflow
        :param mode: the stage or execution node description if 'describe',
                     the execution node log if 'log'
        '''
        folder_url, short_name = self.__client._get_job_folder(name)

        if enode:
            WORKFLOW_DESCRIPTION = (
                '%(folder_url)sjob/%(short_name)s/%(build_id)s/'
                'execution/node/%(enode)d/wfapi/%(mode)s')
        else:
            WORKFLOW_DESCRIPTION = (
                '%(folder_url)sjob/%(short_name)s/%(build_id)s/wfapi/%(mode)s')
        req = send_request(
                'GET',
                self.__client._build_url(WORKFLOW_DESCRIPTION, locals()))
        response = self.__client.jenkins_open(req)
        return json.loads(response)

    @retry(max_count=12)
    def get_artifact(self, name, build_id, artifact_path, destination_name):
        '''Wait until the specified build is finished

        :param name: ``str``, job name
        :param build_id: ``str``, build id or "lastBuild"
        :param artifact_path: ``str``, path and filename of the artifact
                              relative to the job URL
        :param artifact_path: ``str``, destination path and filename
                              on the local filesystem where to save
                              the artifact content
        :returns: requests object with headers and console output,  ``obj``
        '''
        folder_url, short_name = self.__client._get_job_folder(name)

        DOWNLOAD_URL = ('%(folder_url)sjob/%(short_name)s/%(build_id)s/'
                        'artifact/%(artifact_path)s')
        req = send_request(
                'GET',
                self.__client._build_url(DOWNLOAD_URL, locals()))

        response = self.__client.jenkins_request(req)
        return response.content