path: root/utils/buildminer.py

import re
from jenkinsapi.jenkins import Jenkins
from jenkinsapi.build import Build as JenkinsApiBuild
from jenkinsapi.utils.requester import Requester

#import logging
#logger = logging.getlogger(__name__)

class Build(object):
    def get_number(self):
        raise NotImplementedError("Should have implemented this")

    def get_timestamp(self):
        raise NotImplementedError("Should have implemented this")

    def get_status(self):
        raise NotImplementedError("Should have implemented this")

    def is_running(self):
        raise NotImplementedError("Should have implemented this")

    def get_description(self):
        raise NotImplementedError("Should have implemented this")

    def get_revision(self):
        raise NotImplementedError("Should have implemented this")

    def get_console(self):
        raise NotImplementedError("Should have implemented this")

    def get_url(self):
        raise NotImplementedError("Should have implemented this")


class JenkinsBuild(Build):
    def __init__(self, jenkins_build):
        self.jenkins_build = jenkins_build
        self.job = self.jenkins_build.job

    def get_number(self):
        return self.jenkins_build.get_number()

    def get_timestamp(self):
        return self.jenkins_build.get_timestamp()

    def get_status(self):
        if self.jenkins_build.is_running():
            return "RUNNING"
        return self.jenkins_build.get_status()

    def get_description(self):
        if 'description' in self.jenkins_build._data and self.jenkins_build._data['description']:
            return self.jenkins_build._data['description']
        return None

    def get_revision(self):
        try:
            return self.jenkins_build.get_revision()
        except:
            return None

    def is_running(self):
        return self.jenkins_build.get_status()

    def get_console(self):
        params = {"stream": True}
        console_stream = self.job.jenkins.requester.get_url(
            "%s/consoleText" % self.jenkins_build.baseurl,
            params=params)
        return console_stream

    def get_url(self):
        return self.jenkins_build.baseurl

class BuildSystem(object):
    def get_build_status(self, build):
        raise NotImplementedError("Should have implemented this")

    def get_build_revision(self, build):
        """
        Returns revision ID from SCM system if present.
        Otherwise returns None
        """
        raise NotImplementedError("Should have implemented this")

    def get_build_scm_url(self, build):
        raise NotImplementedError("Should have implemented this")

    def get_build_scm_branch(self, build):
        raise NotImplementedError("Should have implemented this")

    def get_last_build(self, project_id, configuration=None):
        raise NotImplementedError("Should have implemented this")

    def get_last_successful_build(self, project_id, configuration=None):
        raise NotImplementedError("Should have implemented this")

    def get_last_completed_build(self, project_id, configuration=None):
        raise NotImplementedError("Should have implemented this")

    def get_build(self, project_id, build_id, configuration=None):
        raise NotImplementedError("Should have implemented this")

    def get_test_job_ids(self, build):
        raise NotImplementedError("Should have implemented this")

    def get_available_configurations(self, project_id):
        raise NotImplementedError("Should have implemented this")

    @staticmethod
    def reduce_build_results(result_list):
        return None


class JenkinsBuildSystem(BuildSystem):
    def __init__(self, base_url, username=None, password=None):
        self.url = base_url
        self.username = username
        self.password = password
        self.jenkinsRequester = Requester(
            self.username,
            self.password,
            baseurl=self.url,
            ssl_verify=False)
        self.jenkins = Jenkins(
            self.url,
            requester=self.jenkinsRequester)

    def _get_configuration(self, project_id, configuration):
        job = self.jenkins[project_id]
        configs = job._data['activeConfigurations']
        build_config = None
        for config in configs:
            print "Active config: %s" % config
            if config['name'] == configuration:
                return config
        return None

    def _get_build_with_config(self, project_id, build_id, configuration):
        job = self.jenkins[project_id]
        build_config = self._get_configuration(project_id, configuration)
        print "build config from Jenkins: %s" % build_config
        if build_config:
            conf_url = build_config['url'] + "%s/" % build_id
            try:
                japibuild = JenkinsApiBuild(conf_url, int(build_id), job)
                return JenkinsBuild(japibuild)
            except:
                print "Build not found!"
        return None

    def get_build(self, project_id, build_id, configuration=None):
        print "Retrieving build: %s %s %s" % (project_id, build_id, configuration)
        if not configuration:
            return JenkinsBuild(self.jenkins[project_id].get_build(int(build_id)))
        return self._get_build_with_config(project_id, build_id, configuration)

#    def get_build_status(self, build):
#        if build.is_running():
#            return "RUNNING"
#        return build.get_status()

    def get_last_build(self, project_id, configuration=None):
        if not configuration:
            return JenkinsBuild(self.jenkins[project_id].get_last_build())
        build_id = self.jenkins[project_id].get_last_build().get_number()
        return self._get_build_with_config(project_id, build_id, configuration)

    def get_last_successful_build(self, project_id, configuration=None):
        if not configuration:
            return JenkinsBuild(self.jenkins[project_id].get_last_good_build())
        build_id = self.jenkins[project_id].get_last_good_build().get_number()
        return self._get_build_with_config(project_id, build_id, configuration)

    def get_last_completed_build(self, project_id, configuration=None):
        number = self.jenkins[project_id].get_last_completed_buildnumber()
        return self.get_build(project_id, number, configuration)

    def get_available_configurations(self, project_id):
        job = self.jenkins[project_id]
        config_list = []
        if 'activeConfigurations' in job._data.keys():
            for config in job._data['activeConfigurations']:
                config_list.append(config['name'])
        return config_list

    @staticmethod
    def reduce_build_results(result_list):
        if len(result_list) < 1:
            return None # raise exception?
        result_set = set(result_list)
        if len(result_set) == 1:
            return result_set.pop()
        else:
            # in the order of importance (presumably)
            if 'FAILURE' in result_set:
                return 'FAILURE'
            if 'ABORTED' in result_set:
                return 'ABORTED'
            if 'UNSTABLE' in result_set:
                return 'UNSTABLE'
            if 'NOT_BUILD' in result_set:
                return 'NOT_BUILD'
        return None


class LinaroMultiConfigHwpackBuildSystem(JenkinsBuildSystem):
    def get_test_job_ids(self, build):
        lava_job_regexp = re.compile('http://validation.linaro.org/scheduler/job/(?P<lava_job_id>\d+)')
        job_id_list = []
        description = build.get_description()
        if description:
            for r in lava_job_regexp.finditer(description):
                job_id_list.append(r.group('lava_job_id'))
        return job_id_list

    def get_build_revision(self, build):
        try:
            return build.get_revision()
        except:
            return None

    def get_build_scm_url(self, build):
        try:
            url_list = build.job.get_scm_url()
            if len(url_list) == 1:
                return url_list[0]
            elif len(url_list) == 0:
                return None
            # rise exception?
            #return build.job.get_scm_url()
        except:
            return None

    def get_build_scm_branch(self, build):
        try:
            branch_list = build.job.get_scm_branch()
            if len(branch_list) == 1:
                return branch_list[0]
            elif len(branch_list) == 0:
                return None
        except:
            return None

class LinaroMultiConfigUpstreamKernelBuildSystem(LinaroMultiConfigHwpackBuildSystem):
    def get_test_job_ids(self, build):
        lava_job_regexp = re.compile('http://validation.linaro.org/scheduler/job/(?P<lava_job_id>\d+)')
        job_id_list = []
        for line in build.get_console().iter_lines():
            job_id_line = lava_job_regexp.search(line)
            if job_id_line:
                for job_id in job_id_line.groups():
                    job_id_list.append(job_id)
        return job_id_list

class LinaroAndroidBuildSystem(JenkinsBuildSystem):
    def get_build_revision(self, build):
        """
        Should we return the version of manifest here?
        """
        return None

    def get_build_scm_url(self, build):
        return None

    def get_build_scm_branch(self, build):
        return None

    def get_test_job_ids(self, build):
        job_id_list = []
        #params = {"stream": True}
        #print self.url
        #console_stream = build.job.jenkins.requester.get_url(
        #    "%s/consoleText" % build.baseurl,
        #    params=params)
        lava_match = re.compile("LAVA Job Id:\s\[?\'?(?P<master_job_id>\d+)")
        #for line in console_stream.iter_lines():
        for line in build.get_console().iter_lines():
            job_id_line = lava_match.search(line)
            if job_id_line:
                for job_id in job_id_line.groups():
                    job_id_list.append(job_id)

        return job_id_list


class AnonymousLinaroAndroidBuildSystem(LinaroAndroidBuildSystem):
    def __init__(self, base_url, username=None, password=None):
        self.url = base_url
        self.jenkins = Jenkins(self.url)