path: root/tools/cts.py

from common import get_auth, write_to_file
from constants import (
    GREEN,
    YELLOW,
    ORANGE,
    RED,
)
from jinja2 import Environment, FileSystemLoader
from lava import LavaServerProxy
from sets import Set
from StringIO import StringIO
from tabulate import tabulate
from urlparse import urlsplit
from xml.dom.minidom import parse
from xml.etree import ElementTree as ET
import lxml.etree as LET
import base64
import codecs
import os
import shutil
import zipfile


RESULT_FILE_NAME = "test_result.xml"
FINAL_RESULTS = "results.xml"
FINAL_REPORT = "results.html"
XSLT = "compatibility_result.xsl"
CSS = "compatibility_result.css"
LOGO = "logo.png"
FOCUSED_LIST = [
    "CtsAndroidAppTestCases",
    "CtsAppTestCases",
    "CtsAppWidgetTestCases",
    "CtsBionicTestCases",
    "CtsContentTestCases",
    "CtsGraphicsTestCases",
    "CtsLibcoreLegacy22TestCases ",
    "CtsLibcoreOkHttpTestCases",
    "CtsLibcoreTestCases",
    "CtsWidgetTestCases",
    "CtsSecurityHostTestCases",
    "CtsSecurityTestCases",
    "CtsViewTestCases",
    "CtsVmTestCases",
    "CtsWebkitTestCases",
    "vm-tests-tf"
]

class CtsRow(object):
    def __init__(self, row, colour=GREEN):
        self.row = row
        self.colour = colour


def download_results(args):
    lava_base_url = urlsplit(args.server).netloc
    lava_username, _, lava_password = get_auth(lava_base_url)
    lava_proxy = LavaServerProxy(lava_base_url, lava_username, lava_password)
    basedir = os.path.join(os.getcwd(), args.output)
    try:
        os.makedirs(basedir)
    except:
        if not args.force:
            print "{0} already exists".format(basedir)
            exit(1)
        else:
            shutil.rmtree(basedir)
            os.makedirs(basedir)

    for lava_job_id in args.joblist:
        if not lava_proxy.test_results_available(lava_job_id):
            print "Results not available for %s" % lava_job_id
            continue

        print "Downloading results for job: %s" % lava_job_id
        bundle = lava_proxy.get_test_job_results(lava_job_id)
        host = [t for t in bundle['test_runs'] if t['test_id'] == 'cts-host']
        if host:
            host = host[0]
        else:
            os.exit(1)
        cts_results_base64 = [a['content'] for a in host['attachments'] if a['pathname'].startswith('android-cts')]
        if len(cts_results_base64) == 0:
            print "No CTS results found in job: %s" % lava_job_id
            continue

        print "Extracting results for job: %s" % lava_job_id
        infile = StringIO(cts_results_base64[0])
        outfile = StringIO()
        base64.decode(infile, outfile)
        compressed = zipfile.ZipFile(outfile)
        print "CTS results for job %s: %s" % (lava_job_id, compressed.namelist()[0])

        dirpath = os.path.join(basedir, compressed.namelist()[0])
        print dirpath
        if os.path.exists(dirpath) and os.path.isdir(dirpath):
            if not args.force:
                print "Directory %s exists. Skipping" % compressed.namelist()[0]
                continue
            else:
                shutil.rmtree(dirpath)
        compressed.extractall(basedir)


def combine_results(args):
    basedom = None
    current_dir = os.path.join(os.getcwd(), args.output)
    current_index = 0
    print "Combining CTS results"
    for (index, directory) in enumerate(os.walk(current_dir).next()[1]):
        print directory
        if not directory.startswith(".") and directory not in ['device-info-files', 'tools']:
            if basedom is None:
                shutil.copy("%s/%s" % (os.path.join(current_dir, directory),RESULT_FILE_NAME), current_dir)
                currentsource = "%s/%s" % (os.path.join(current_dir, directory), RESULT_FILE_NAME)
                basedom = parse(currentsource)
                testpackage_list = basedom.getElementsByTagName("Module")
                for testpackage in testpackage_list:
                    packagename = testpackage.getAttribute("name")
                    packageabi = testpackage.getAttribute("abi")
                    print "\t%s (%s)" % (packagename, packageabi)
            else:
                currentsource = "%s/%s" % (os.path.join(current_dir, directory), RESULT_FILE_NAME)
                currentdom = parse(currentsource)
                result = basedom.getElementsByTagName("Result")[0]
                basesummary = basedom.getElementsByTagName("Summary")[0]
                summary = currentdom.getElementsByTagName("Summary")[0]
                for attr in summary.attributes.keys():
                    basevalue = basesummary.getAttribute(attr)
                    basesummary.setAttribute(attr, str(int(summary.getAttribute(attr)) + int(basevalue)))
                testpackage_list = currentdom.getElementsByTagName("Module")
                for testpackage in testpackage_list:
                    packagename = testpackage.getAttribute("name")
                    packageabi = testpackage.getAttribute("abi")
                    print "\t%s (%s)" % (packagename, packageabi)
                    added = False
                    for package in result.getElementsByTagName("Module"):
                        #name = package.getAttribute("appPackageName")
                        name = package.getAttribute("name")
                        if sorted([name, packagename])[0] == packagename:
                            result.insertBefore(testpackage, package)
                            added = True
                            break
                    if not added:
                        result.appendChild(testpackage)

    finalfile_path = os.path.join(current_dir, FINAL_RESULTS)
    finalfile = codecs.open(finalfile_path, "w", "utf-8")
    basedom.writexml(finalfile)
    finalfile.close()

def cts_summary(args, focused_list=None, generate_style=False):
    sourcepath = os.path.join(os.getcwd(), args.output)
    sourcefile = os.path.join(sourcepath, args.sourcefile)
    tree = ET.parse(sourcefile)
    root = tree.getroot()

    testpackages = root.findall(".//Module[@abi='%s']" % args.abi)
    results = []
    cts_styles = []
    totalpass = 0
    totalfail = 0
    totaltimeout = 0
    totalnotexec = 0
    for package in testpackages:
        if focused_list:
            if package.attrib['name'] not in focused_list:
                continue
        testpass = package.findall(".//Test[@result='pass']")
        totalpass = totalpass + len(testpass)
        testfail = package.findall(".//Test[@result='fail']")
        totalfail = totalfail + len(testfail)
        testtimeout = package.findall(".//Test[@result='timeout']")
        totaltimeout = totaltimeout + len(testtimeout)
        testnotexec = package.findall(".//Test[@result='notExecuted']")
        totalnotexec = totalnotexec + len(testnotexec)
        #import pdb;pdb.set_trace()
        if len(testpass) + len(testfail) + len(testtimeout) + len(testnotexec) == 0:
            print "No results for {0}".format(package.attrib['name'])
            continue
        passrate = float(
            len(testpass))/float(len(testpass) + len(testfail) + \
                len(testtimeout) + len(testnotexec))
        results.append([
            package.attrib['name'],
            len(testpass),
            len(testfail),
            len(testtimeout),
            len(testnotexec),
            "{0:.2f}".format(passrate*100)])
        index = len(results)
        ctsrow = CtsRow(index)
        if passrate < 1 and passrate >= 0.5:
            ctsrow.colour = YELLOW
        elif passrate < 0.5 and passrate > 0:
            ctsrow.colour = ORANGE
        elif passrate == 0:
            ctsrow.colour = RED
        cts_styles.append(ctsrow)

    if totalpass + totalfail + totaltimeout + totalnotexec == 0:
        print "No CTS results"
        return
    results.append(["**Total**",
        "**{}**".format(totalpass),
        "**{}**".format(totalfail),
        "**{}**".format(totaltimeout),
        "**{}**".format(totalnotexec),
        "**{0:.2f}**".format(100*float(totalpass)/float(totalpass + \
                totalfail + \
                totaltimeout + \
                totalnotexec))])
    headers=["**package**",
            "**pass**",
            "**fail**",
            "**timeout**",
            "**not executed**",
            "**%**"]

    name = "summary"
    if focused_list is not None:
        name = "focused"
    env = Environment(loader=FileSystemLoader('templates'))
    cts_template = env.get_template("cts")
    cts_output = cts_template.render(
        name = name,
        ctstable = tabulate(results,
        headers=headers,
        floatfmt=".2f",
        tablefmt="rst")
    )
    write_to_file(args, "cts-{0}.rst".format(name), cts_output)

    cts_style = env.get_template("cts.styles")
    styles_output = cts_style.render(
        name = name,
        ctsrows = cts_styles
    )
    write_to_file(args, "cts-{0}.styles".format(name), styles_output)

    if args.savecsv:
        import csv
        ctspath = os.path.join(os.getcwd(), args.output)
        ctsfile = os.path.join(ctspath, "cts.csv")
        with open(ctsfile, "wb") as fp:
            writer = csv.writer(fp, quoting=csv.QUOTE_ALL)
            writer.writerows(results)

def focused_summary(args):
    cts_summary(args, FOCUSED_LIST)

def purge_results(args):
    #remove only files produced by CTS module
    # go through subdirs and check if they contain CTS results
    results_dir = os.path.join(os.getcwd(), args.output)
    cts_dirs = Set()
    # remove files generated by script
    files = ['cts-summary.rst',
        'cts-focused.rst',
        'summary.styles',
        'focused.styles',
        'results.xml',
        'test_result.xml',
        'compatibility_result.css',
        'results.html',
        'logo.png'
    ]
    for f in files:
        path = os.path.join(results_dir, f)
        if os.path.exists(path) and os.path.isfile(path):
            os.unlink(path)
    for subdir, dirs, files in os.walk(results_dir):
        if RESULT_FILE_NAME in files:
            cts_dirs.add(subdir)
            continue
    print cts_dirs
    for subdir in cts_dirs:
        shutil.rmtree(os.path.join(results_dir, subdir))

def __copy_file(expected, final_report_path):
    filename = expected.rsplit("/", 1)[1]
    if not os.path.exists(expected):
        for subdir, dirs, files in os.walk(final_report_path):
            if filename in files:
                shutil.copyfile(os.path.join(final_report_path, subdir, filename),
                    expected)


def produce_html(args):
    final_report_path = os.path.join(os.path.join(os.getcwd(), args.output))
    final_result_file = os.path.join(final_report_path, FINAL_RESULTS)
    final_report_file = os.path.join(final_report_path, FINAL_REPORT)
    xslt_result_file = os.path.join(os.getcwd(), XSLT)
    css_result_file = os.path.join(final_report_path, CSS)
    logo_result_file = os.path.join(final_report_path, LOGO)
    if not os.path.exists(final_result_file):
        print "{0} not found".format(final_result_file)
        return
    for filename in [css_result_file, logo_result_file]:
        __copy_file(filename, final_report_path)

    if os.path.exists(final_report_file) and not args.force:
        return
    dom = LET.parse(final_result_file)
    xslt = LET.parse(xslt_result_file)
    transform = LET.XSLT(xslt)
    newdom = transform(dom)
    write_to_file(args, FINAL_REPORT, LET.tostring(newdom, pretty_print=True))

def produce_pdf(args):
    # call wkhtmltopdf
    pass