path: root/report_automation.py

#!/usr/bin/python

import xmlrpclib
import json
import re
import os
import ast
import zipfile
import base64
import shutil
import argparse
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"

class LinaroAndroidBuildSystem(object):
    def __init__(self, base_url, build_job,
                 username=None, password=None):
        print("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:
            print("Can not get Jenkins job: %s" % project)
            print e

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

    def get_test_job_ids(self, build_no):
        print("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:
            print("Can not connect to LAVA server: %s" % lava_server)

    def get_job_info(self, job_no):
        job_info = {}
        try:
            job_status      = self.server.scheduler.job_status(job_no)
            job_detail      = self.server.scheduler.job_details(job_no)
        except:
            print("Can not get any information for test job: %d" % 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'] == 'bionic_libc_tests' 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' 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 = [
#            {"package_name": "android.app",
#             "total": 100,
#             "passed": 90,
#             "failed": 5,
#             "skipped": 5,
#             "pass_rate": 85
#             }
#        ]
        self.lava_tests = {}
#        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):
        print("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

#        return lava_tests

    def parse_benchmark_test(self, job_info):
        print("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').next()
            src      = (s for s in test_run['software_context']['sources'] if 'test_params' in s).next()
        except Exception as e:
            print("Job %s seems not be a completed job." % job_info['job_no'])
            print e
        benchmark_test['test_id']       = ast.literal_eval(src['test_params'])['TEST_NAME']
        benchmark_test['result_link']   = job_info['result_link']
        for test in test_run['test_results']:
            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):
        print("Parsing cts test job #%s" % job_info['job_no'])
        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:
            print("Job %s seems not be a completed job." % job_info['job_no'])
            print e
        try:
            infile = StringIO(attachment['content'])
            outfile = StringIO()
            base64.decode(infile, outfile)
            compressed = zipfile.ZipFile(outfile)
            compressed.extractall()
        except Exception as e:
            print("Can not extract attachment: %s" % attachment['pathname'])
            print e
        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.append(cts_test)

        shutil.rmtree(CTS_TEST_RESULT_DIR)


    def to_dict(self):
        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
        }

if __name__ == '__main__':

    JENKINS_URL     = "https://android-build.linaro.org/jenkins/"
    JENKINS_JOB     = "linaro-android_lcr-member-juno"
    LAVA_USERNAME   = "arthur.she"
    LAVA_TOKEN      = "a6eajjn85bp11n87u30oaxb11ewn687843btsdw63wc3gt10t72rqj4x3bi0bd0czuqotz8rp2w3moa5gja5r03u920f0duux10z78t7fcdslqi8omaxtnsgttgfcfs7"
    REPORT_TEMPLATE = "py3o_report_juno_template.odt"
    REPORT_OUTPUT   = "py3o_report_juno_output.odt"

    parser          = argparse.ArgumentParser()
    parser.add_argument("-job", "--jenkins_job_name", type=str, default=JENKINS_JOB,
                        help="Specify the Jenkins job name. Default: %s" % JENKINS_JOB)
    parser.add_argument("-bn", "--build_no", type=int, required=True,
                        help="Specify the Jenkins build number")
    parser.add_argument("-a", "--manual_lava_jobs", nargs = '+',
                        help="Specify the test jobs that were submitted manually")
    parser.add_argument("-d", "--exclude_lava_jobs", nargs = '+',
                        help="Specify the test jobs that wouldn't be included in the report")

    args = parser.parse_args()

    t           = Template(REPORT_TEMPLATE, REPORT_OUTPUT)
    rep_obj     = Report()
    jenkins     = LinaroAndroidBuildSystem(base_url=JENKINS_URL, build_job=args.jenkins_job_name)
    lava_server = LAVA(LAVA_USERNAME, LAVA_TOKEN)

    # Query test job IDs
    test_job_ids = []
    test_job_ids = jenkins.get_test_job_ids(args.build_no)
    print("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)

    print(test_job_ids)
    print("Going to parse these test jobs!!")
    for job in test_job_ids:
        job_info    = {}
        job_info    = lava_server.get_job_info(job)
        if job_info['type']    == 'lava_test':
            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)

    print("Prepare for the report!!")
    data = rep_obj.to_dict()
    t.render(data)
    print("Done, the output report is \"%s\"" % REPORT_OUTPUT)