path: root/report_automation.py

#!/usr/bin/python

import xmlrpclib
import json
import re
import os
import ast
import sys
import zipfile
import base64
import shutil
import argparse
import logging
from argparse import RawTextHelpFormatter
import netrc
from StringIO import StringIO
import xml.etree.ElementTree as ET
from py3o.template import Template
from jenkinsapi.jenkins import Jenkins

CTS_RESULT_FILE_NAME = "testResult.xml"
logging.basicConfig(format='%(levelname)s:  %(message)s', level=logging.INFO)
logger = logging.getLogger(__name__)
#logger.setLevel(logging.INFO)

class TestJobParseError(Exception):
    def __init__(self, job_id):
        self.job_type = job_id

    def __str__(self):
        return repr(self.job_id)


class LinaroAndroidBuildSystem(object):
    def __init__(self, base_url, build_job,
                 username=None, password=None):
        logger.info("Jenkins build job: %s" %  build_job)
        self.base_url  = base_url
        self.build_job = build_job
        self.username  = username
        self.password  = password
        try:
            self.jenkins    = Jenkins(self.base_url, self.username, self.password)
            self.project    = self.jenkins[self.build_job]
        except Exception as e:
            logger.warning("Can not get Jenkins job: %s" % project)

    def get_build(self, build_no):
        try:
            build = self.project.get_build(build_no)
            return build
        except Exception as e:
            logger.warning("Can not get build: #%d" % build_no)

    def get_test_job_ids(self, build_no):
        logger.info("Querying build #%d test job ids from Jenkins build job: %s" % 
              (build_no, self.build_job))
        job_id_list = []
        build = self.get_build(build_no)
        job_id_list = re.findall("LAVA Job Id:\s\[?\'?(?P<master_job_id>[\d\.]+)", build.get_console())

        return job_id_list


class LAVA(object):
    def __init__(self, username, token, lava_server="https://validation.linaro.org/"):
        try:
            self.lava_server = lava_server
            self.username    = username
            self.token       = token
            server_url       = "https://%s:%s@%sRPC2/" % (self.username, self.token, self.lava_server.split("//")[1])
            self.server      = xmlrpclib.ServerProxy(server_url)
        except:
            logger.warning("Can not connect to LAVA server: %s" % lava_server)
            exit(-1)

    def get_job_info(self, job_no):
        job_status = None
        job_info   = {}
        try:
            job_status      = self.server.scheduler.job_status(job_no)
        except xmlrpclib.ProtocolError as err:
            if err.errcode == 401:
                logger.error("Invalid LAVA credentials. Please check your .netrc")
                sys.exit(1)
        except:
            logger.warning("Can not get any information for test job: %s" % job_no)

        if job_status == None or job_status['job_status'] != 'Complete':
            logger.warning("!!Job #%s is not completed.\n\tJob status: %s!! " % (job_no, job_status['job_status']))
        else:
            job_detail                  = self.server.scheduler.job_details(job_no)
            job_info['job_no']          = job_no
            job_info['result_link']     = job_detail['_results_link']
            job_info['result_bundle']   = json.loads(self.server.dashboard.get(job_status['bundle_sha1'])['content'])
            if any(t['test_id'] == 'busybox' for t in job_info['result_bundle']['test_runs']):
                job_info['type']    = "lava_test"
            elif any(t['test_id'] == 'cts-host' for t in job_info['result_bundle']['test_runs']):
                job_info['type']    = "cts"
            elif any(t['test_id'] == 'lava-android-benchmark-target' or t['test_id'] == 'wa2-host-postprocessing'
                 for t in job_info['result_bundle']['test_runs']):
                job_info['type']    = "benchmark"
            else:
                job_info['type']    = "other"

        return job_info


class Report(object):
    def __init__(self):
        self.compiler_version       = "GCC 4.9\nGCC 4.8"
        self.firmware_version       = "3.11"
        self.android_version        = "5.1.1 Lollipop"
        self.kernel_version         = "4.10.79"
        self.cts_tests                    = {}
        self.cts_tests['total_pkgs']      = 0
        self.cts_tests['total_passed']    = 0
        self.cts_tests['total_failed']    = 0
        self.cts_tests['total_skipped']   = 0
        self.cts_tests['total_pass_rate'] = 0
        self.cts_tests['test_pkgs']       = []
#        self.cts_tests = {
#            "total_pkgs":     2000,
#            "total_passed":   1950,
#            "total_failed":   50,
#            "total_pass_rate": 98,
#            test_pkgs: [
#               {"package_name": "android.app",
#                "total": 100,
#                "passed": 90,
#                "failed": 5,
#                "skipped": 5,
#                "pass_rate": 85
#               }
#            ]
#        }
        self.lava_tests = {
            "result_link": "",
            "test_cases": []
        }
#        self.lava_tests = {
#            "result_link": "RESULT BUNDLE URL",
#            "test_cases": [
#                {"test_id"  : "busybox",
#                 "Pass"     : 12,
#                 "Fail"     : 5,
#                 "Skip"     : 5,
#                 "Unknown"  : 5,
#                 "failed_test_cases" : "ls, pwd"}
#            ]
#        }
        self.benchmark_tests = []
#        self.benchmark_tests = [
#            {"test_id"  : "Pi",
#             "status"   : "Pass",
#             "result_link"    : "RESULT BUNDLE URL",
#             "sub_tests": [
#                    {"test_id"  : "Pi sub - 1",
#                     "result"   : "Pass",
#                     "score"    : "100ms"},
#                    {"test_id"  : "Pi sub - 2",
#                     "result"   : "Pass",
#                     "score"    : "100ms"}
#             ]
#            }

    def set_compiler_version(self, compiler_version):
        #ToDo create similar method for each property
        self.compiler_version = compiler_version

    def set_firmware_version(self, firmware_version):
        self.firmware_version = firmware_version

    def set_android_version(self, android_version):
        self.android_version = android_version

    def set_kernel_version(self, kernel_version):
        self.kernel_version = kernel_version

    def parse_lava_test(self, job_info):
        logger.info("Parsing lava test job #%s" % job_info['job_no'])
        lava_tests  = {}
        lava_tests['result_link'] = job_info['result_link']
        lava_tests['test_cases']  = []
        for test_run in job_info['result_bundle']['test_runs']:
            if test_run['test_id'] == 'lava':
                continue
            # Count passed and failed test cases
            test_case                      = {}
            test_case['Pass']              = 0
            test_case['Fail']              = 0
            test_case['Skip']              = 0
            test_case['Unknown']           = 0
            test_case['failed_test_cases'] = ""
            test_case['test_id']           = test_run['test_id']
            for test_result in test_run['test_results']:
                result = test_result['result'].title()
                if(result == 'Fail'):
                    test_case['failed_test_cases'] += "\n%s" % test_result['test_case_id']
                exec("test_case['%s'] = test_case['%s'] + 1" % (result, result))
            lava_tests['test_cases'].append(test_case)
            self.lava_tests = lava_tests

    def parse_benchmark_test(self, job_info):
        logger.info("Parsing benchmark test job #%s" % job_info['job_no'])
        benchmark_test = {}
        sub_tests      = []
        try:
            test_run = (t for t in job_info['result_bundle']['test_runs']
                        if t['test_id'] == 'lava-android-benchmark-host' or
                           t['test_id'] == 'wa2-host-postprocessing').next()
            src = (s for s in test_run['software_context']['sources'] if 'test_params' in s).next()
        except Exception as e:
            logger.warning("\tJob %s seems not be a completed job." % job_info['job_no'])
            raise TestJobParseError(job_info['job_no'])

        if test_run['test_id'] == 'lava-android-benchmark-host':
            benchmark_test['test_id'] = ast.literal_eval(src['test_params'])['TEST_NAME']
        else:
            benchmark_test['test_id'] = ast.literal_eval(src['test_params'])['JOB_NAME']
        benchmark_test['result_link']   = job_info['result_link']
        for test in test_run['test_results']:
            if not str(test['test_case_id']).startswith('lava-test-shell'):
                test_case = {}
                test_case['score']   = ""
                test_case['test_id'] = test['test_case_id']
                test_case['result']  = test['result'].upper()
                if 'measurement' in test:
                    test_case['score'] = "%s %s" % (test['measurement'], test['units'])
                sub_tests.append(test_case)
        if len(sub_tests) > 0:
            benchmark_test['status'] = "PASS"
        else:
            benchmark_test['status'] = "FAIL"
        benchmark_test['sub_tests'] = sub_tests
    
        self.benchmark_tests.append(benchmark_test)

    def parse_cts(self, job_info):
        logger.info("Parsing cts test job #%s" % job_info['job_no'])
        attachment = None
        try:
            test_run   = (r for r in job_info['result_bundle']['test_runs']
                          if r['test_id'] == 'cts-host').next()
            attachment = (a for a in test_run['attachments']
                          if a['pathname'].startswith('android-cts')).next()
        except Exception as e:
            logger.warning("\tJob %s seems not be a completed job." % job_info['job_no'])
            raise TestJobParseError(job_info['job_no'])

        if attachment is not None:
            try:
                infile = StringIO(attachment['content'])
                outfile = StringIO()
                base64.decode(infile, outfile)
                compressed = zipfile.ZipFile(outfile)
                compressed.extractall()
            except Exception as e:
                logger.warning("\tCan not extract attachment: %s" % attachment['pathname'])
            CTS_TEST_RESULT_DIR = "./%s" % attachment['pathname'].split('/')[-1].split('.zip')[0]
            CTS_RESULT_FILE     = "%s/%s" % (CTS_TEST_RESULT_DIR, CTS_RESULT_FILE_NAME)
            tree = ET.parse(CTS_RESULT_FILE)
            root = tree.getroot()
            for pkg in root.iter("TestPackage"):
                cts_test = {}
                cts_test['package_name'] = pkg.get('appPackageName')
                cts_test['total']        = len(pkg.findall(".//Test"))
                cts_test['passed']       = len(pkg.findall(".//Test[@result='pass']"))
                cts_test['failed']       = len(pkg.findall(".//Test[@result='fail']"))
                cts_test['skipped']      = len(pkg.findall(".//Test[@result='skip']"))
                cts_test['pass_rate']    = 100 * cts_test['passed']/cts_test['total']
                self.cts_tests['test_pkgs'].append(cts_test)
            shutil.rmtree(CTS_TEST_RESULT_DIR)
        else:
            logger.warning("\tNo attachment in it")

    def to_dict(self):
        # Calculate numbers of CTS tests
        for pkg in self.cts_tests['test_pkgs']:
            self.cts_tests['total_pkgs']    += pkg['total']
            self.cts_tests['total_passed']  += pkg['passed']
            self.cts_tests['total_failed']  += pkg['failed']
            self.cts_tests['total_skipped'] += pkg['skipped']
        if self.cts_tests['total_pkgs'] != 0:
            self.cts_tests['total_pass_rate'] = 100 * self.cts_tests['total_passed']/self.cts_tests['total_pkgs']

        return {
            "kernel_version"        : self.kernel_version,
            "compiler_version"      : self.compiler_version,
            "firmware_version"      : self.firmware_version,
            "android_version"       : self.android_version,
            "cts_tests"             : self.cts_tests,
            "lava_tests"            : self.lava_tests,
            "benchmark_tests"       : self.benchmark_tests
        }


class FunctionalTestReport(Report):
    def parse_lava_test(self, job_info):
        logger.info("Parsing lava test job #%s" % job_info['job_no'])
        lava_tests  = {}
        lava_tests['result_link'] = job_info['result_link']
        lava_tests['testsuites']  = []
        for test_run in job_info['result_bundle']['test_runs']:
            if test_run['test_id'] == 'lava':
                continue
            # Count passed and failed test cases
            test_case                      = {}
            test_case['Pass']              = 0
            test_case['Fail']              = 0
            test_case['Skip']              = 0
            test_case['Unknown']           = 0
            test_case['name']              = test_run['test_id']
            test_case['results']           = []
            for test_result in test_run['test_results']:
                result = test_result['result'].title()
                name = test_result['test_case_id']
                measurement = ""
                if 'measurement' in test_result.keys():
                    measurement = test_result['measurement']
                test_case[result] = test_case[result] + 1
                test_case['results'].append({'name': name, 'result': result, 'measurement': measurement})
            lava_tests['testsuites'].append(test_case)
            self.lava_tests = lava_tests

    def to_dict(self):
        return {
            "testsuites"            : self.lava_tests['testsuites']
        }


if __name__ == '__main__':

    JENKINS_URL     = "https://android-build.linaro.org/jenkins/"
    JENKINS_JOB     = "linaro-android_lcr-member-juno"
    REPORT_TEMPLATE = "py3o_report_juno_template.odt"
    REPORT_OUTPUT   = "report_output.odt"
    FUNCT_REPORT_TEMPLATE = "functional_test_results_template.ods"
    FUNCT_REPORT_OUTPUT   = "functional_report_output.ods"

    parser          = argparse.ArgumentParser(formatter_class=RawTextHelpFormatter,
                      description="""This script will collect all data and make a report.
The authentication credential for LAVA/Jenkins server should be put in ~/.netrc
e.g.
   machine   validation.linaro.org
   login     USERNAME
   password  AUTHENTICATION TOKEN""")
    parser.add_argument("-bn", "--build-no", dest="build_no", type=int, required=True,
                        help="Specify the Jenkins build number. This is MANDATORY.")
    parser.add_argument("-u", "--jenkins-url", dest="jenkins_url", default=JENKINS_URL,
                        help="Specify the Jenkins URL. This is optional.\nDefault: %s" % JENKINS_URL)
    parser.add_argument("-job", "--jenkins-job-name", dest="jenkins_job_name", default=JENKINS_JOB,
                        help="Specify the Jenkins job name.This is optional.\nDefault: %s" % JENKINS_JOB)
    parser.add_argument("-a", "--manual-lava-jobs", dest="manual_lava_jobs", nargs = '+',
                        help="Specify the test jobs that were submitted manually. This is optional.")
    parser.add_argument("-d", "--exclude-lava-jobs", dest="exclude_lava_jobs", nargs = '+',
                        help="Specify the test jobs that don't want to be included in the report. This is optional.")
    parser.add_argument("-temp", "--template", dest="template_file", default=REPORT_TEMPLATE,
                        help="Specify the report template. This is optional.\nDefault: %s" % REPORT_TEMPLATE)
    parser.add_argument("-ftemp", "--functional_template", dest="functional_template_file", default=FUNCT_REPORT_TEMPLATE,
                        help="Specify the report template for functional tests. This is optional.\nDefault: %s" % FUNCT_REPORT_TEMPLATE)
    parser.add_argument("-o", "--output", dest="output_file", default=REPORT_OUTPUT,
                        help="Specify the output file name for functional tests. This is optional.\nDefault: %s" % FUNCT_REPORT_OUTPUT)
    parser.add_argument("-fo", "--functional-output", dest="functional_output_file", default=FUNCT_REPORT_OUTPUT,
                        help="Specify the output file name. This is optional.\nDefault: %s" % REPORT_OUTPUT)
    parser.add_argument("-f", "--force-report", dest="force_report", 
                        help="Generate report even there're some errors", action="store_true")

    args = parser.parse_args()

    generate_report = True
    try:
        lava_login_info = netrc.netrc().authenticators("validation.linaro.org")
    except:
        logger.error("Can not get authentication credential for LAVA Server from ~/.netrc")
        exit(-1)
    if lava_login_info is None:
        logger.error("Can not get authentication credential for LAVA Server from ~/.netrc")
        exit(-1)
    jenkins_ligin_info = netrc.netrc().authenticators(args.jenkins_url.split('/')[2])
    t           = Template(args.template_file, args.output_file)
    ft          = Template(args.functional_template_file, args.functional_output_file)
    rep_obj     = Report()
    fun_rep_obj = FunctionalTestReport()
    if jenkins_ligin_info is not None:
        jenkins = LinaroAndroidBuildSystem(base_url=args.jenkins_url, build_job=args.jenkins_job_name,
                                           username=jenkins_ligin_info[0], password=jenkins_ligin_info[2])
    else:
        jenkins = LinaroAndroidBuildSystem(base_url=args.jenkins_url, build_job=args.jenkins_job_name)
    lava_server = LAVA(lava_login_info[0], lava_login_info[2])

    # Query test job IDs
    test_job_ids = []
    test_job_ids = jenkins.get_test_job_ids(args.build_no)
    if not test_job_ids:
        logger.error("There is no test jobs in this build!")
        exit(-1)
    logger.info("Get the test job IDs:")
    if args.exclude_lava_jobs:
        test_job_ids = [id for id in test_job_ids if id not in args.exclude_lava_jobs]
    if args.manual_lava_jobs:
        test_job_ids.extend(args.manual_lava_jobs)

    logger.info(test_job_ids)
    logger.info("Going to parse these test jobs!!")
    uncompleted_job = []
    parse_error_job = []
    for job in test_job_ids:
        job_info    = {}
        job_info    = lava_server.get_job_info(job)
        if job_info:
            try:
                if job_info['type']    == 'lava_test':
                    rep_obj.parse_lava_test(job_info)
                    fun_rep_obj.parse_lava_test(job_info)
                elif job_info['type']  == 'benchmark':
                    rep_obj.parse_benchmark_test(job_info)
                elif job_info['type']   == 'cts':
                    rep_obj.parse_cts(job_info)
            except TestJobParseError as e:
                parse_error_job.append(job)
        else:
            uncompleted_job.append(job)

    print("------------------------------------------------------")
    if uncompleted_job or parse_error_job:
        if args.force_report:
            generate_report = True
        else:
            generate_report = False
            logger.info("Something wrong, the report will not be generated!!")
        logger.info("%d uncompleted test jobs: %s" % (len(uncompleted_job), ', '.join(uncompleted_job)))
        logger.info("%d test jobs parse error: %s" % (len(parse_error_job), ', '.join(parse_error_job)))

    if generate_report:
        logger.info("Prepare for the report!!")
        data = rep_obj.to_dict()
        fun_data = fun_rep_obj.to_dict()
        with open("backup_fun_data.json", "w") as bf:
            json.dump(fun_data, bf)
        t.render(data)
        ft.render(fun_data)
        logger.info("Done, the output report is \"%s\"" % REPORT_OUTPUT)