ltp2JUnit

#!/usr/bin/env python
#

from sys                             import argv, stdout
from getopt                          import getopt, GetoptError
from os                              import path
from datetime                        import datetime, date
from traceback                       import format_exc
from string                          import maketrans
from xml.sax.saxutils                import escape
import uuid

from ktl.utils                       import stdo, error, dump
from ktl.dbg                         import Dbg

import JUnit_api as api

# Exit
#
class Exit():
    """
    If an error message has already been displayed and we want to just exit the app, this
    exception is raised.
    """
    pass

# CmdlineError
#
# The type of exception that will be raised by Cmdline.process() if there
# are command line processing errors.
#
class CmdlineError(Exception):
    # __init__
    #
    def __init__(self, error):
        self.msg = error

# Cmdline
#
class Cmdline:
    """
    Handle all the command line processing for the application.
    """
    # __init__
    #
    def __init__(self):
        self.cfg = {}

    # error
    #
    def error(self, e, defaults):
        """
        Simple helper which prints out an error message and then prints out the usage.
        """
        if e != '': error(e)
        self.usage(defaults)

    # usage
    #
    def usage(self, defaults):
        """
        Prints out the help text which explains the command line options.
        """
        stdo("    Usage:                                                                                   \n")
        stdo("        %s <Options> <log-file>                                                              \n" % defaults['app_name'])
        stdo("                                                                                             \n")
        stdo("    Options:                                                                                 \n")
        stdo("        --help           Prints this text.                                                   \n")
        stdo("                                                                                             \n")
        stdo("        --debug=<debug options>                                                              \n")
        stdo("                         Performs additional output related to the option enabled and        \n")
        stdo("                         the application defined support for the option.                     \n")
        stdo("                                                                                             \n")
        stdo("        --raise                                                                              \n")
        stdo("                         Raise exceptions encountered during processing                      \n")
        stdo("                         (default behavior is to include them in the xml output)             \n")
        stdo("                                                                                             \n")
        stdo("    Examples:                                                                                \n")
        stdo("        %s ltp-results.log                                                                   \n" % defaults['app_name'])

    # process
    #
    def process(self, argv):
        """
        This method is responsible for calling the getopt function to process the command
        line. All parameters are processed into class variables for use by other methods.
        """
        cfg = {}
        result = True
        try:
            optsShort = ''
            optsLong  = ['help', 'debug=', 'raise']
            opts, args = getopt(argv[1:], optsShort, optsLong)

            for opt, val in opts:
                if (opt == '--help'):
                    raise CmdlineError('')

                elif (opt == '--raise'):
                    cfg['raise'] = True

                elif opt in ('--debug'):
                    cfg['debug'] = val.split(',')
                    for level in cfg['debug']:
                        if level not in Dbg.levels:
                            Dbg.levels.append(level)

            if result: # No errors yet

                # At lease one source package must be specified.
                #
                if len(args) > 0:
                    cfg['logs'] = args

        except GetoptError, error:
            raise CmdlineError(error)

        return cfg

# Ltp
#
class Ltp():

    # __init__
    #
    def __init__(self):
        self.raiseExceptions = False
        self.all_bytes = maketrans('', '')
        return

    # escape_string
    #
    def escape_string(self, instr):
        outstr = instr.translate(self.all_bytes, self.all_bytes[:32])  # All bytes < 32 are deleted (second argument)
        return escape(outstr)


    # parse_header
    #
    def parse_header(self, stream):
        """
        Parse meta information about the ltp run from the first part of the
        output file. When we reach the end of the header, return what we have.
        """

        Dbg.enter("Ltp.parse_header")
        # List of items which are in the format MATCHTERM=VALUE, possibly quoted (to be stripped)
        # may not contain spaces
        equal_delimited_list = ['DISTRIB_ID', 'DISTRIB_RELEASE', 'DISTRIB_CODENAME', 'DISTRIB_DESCRIPTION']

        # List of terms which when matched at the start of the line, will include the entire rest of the line as the VALUE
        # These may contain spaces
        match_list = ['Modules Loaded']

        # TODO possibly add code to extract CPU information and amount of memory

        properties = {}
        # read until we get "Running tests......."
        for line in stream:
            if 'Running tests.......' in line:
                # we're done!
                return properties

            tag, sep, therest = line.partition('=')
            if sep != '':
                if tag in equal_delimited_list:
                    properties[tag] = therest.strip('"').strip()
                continue

            for term in match_list:
                if line.startswith(term):
                    lt = len(term)
                    properties[term.replace(' ', '_')] = line[lt:].strip()
                    continue

            # find the uname output to get the hostname and kernel version
            l = line.strip()
            if l.startswith('Linux') and l.endswith('GNU/Linux'):
                properties['hostname'] = l.split()[1]
                properties['kernel_version'] = l.split()[2]

        raise ValueError('EOF Before end of header')
        # create some data indicating a major failure
        # TODO

    # parse_testcases
    #
    def parse_testcases(self, stream):
        """
        Parse information about each test in the ltp run from the log file
        and create the testcase objects for the output. Keep track of totals
        for erors and fails, and the total test time spent.
        """
        Dbg.enter("Ltp.parse_testcases")
        total_tests = 0
        total_time = 0
        total_failures = 0
        total_errors = 0
        testcases = []

        while True:
            # Clear all the collection variables
            tr_output = []
            tr_duration = None
            tr_termination_type = None
            tr_termination_id = None
            tr_corefile = ''
            tr_cutime = 0
            tr_cstime = 0
            nextline = ''

            # read the first line of testcases
            line = stream.next()
            # Tests start with '<<<test_start>>>' and end with '<<<test_end>>>'
            # if next line is not a test start, we assume we're at the end of all tests
            if not line.strip().startswith('<<<test_start>>>'):
                # TODO calculate totals if needed
                return total_tests, total_failures, total_errors, total_time, nextline, testcases

            tr_output.append(line.strip())

            # iterate over the first test case part
            for line in stream:
                l = line.strip()
                tr_output.append(l)
                if l.startswith('<<<test_output>>>'):
                    break
                # collect information from the top of the test output
                if l.startswith('tag'):
                    tr_tag = l.split()[0].split('=')[1]
                if l.startswith('cmdline'):
                    tag, sep, therest = l.partition('=')
                    tr_cmdline = self.escape_string(therest.strip('"'))
                if l.startswith('contacts'):
                    tag, sep, therest = l.partition('=')
                    tr_contacts = therest.strip('"')
                if l.startswith('analysis'):
                    tag, sep, therest = l.partition('=')
                    tr_analysis = therest

            for line in stream:
                l = line.strip()
                tr_output.append(l)
                if l.startswith('<<<execution_status>>>'):
                    break

            for line in stream:
                l = line.strip()
                tr_output.append(l)
                if l.startswith('<<<test_end>>>'):
                    break

                # collect execution status
                if l.startswith('initiation_status'):
                    tag, sep, therest = line.partition('=')
                    tr_initiation_status = therest.strip('"')
                if l.startswith('duration') or l.startswith('cutime'):
                    # lines containing multiple items
                    # duration=0 termination_type=exited termination_id=0 corefile=no
                    # cutime=0 cstime=0

                    for item in l.split():
                        tag, value = item.split('=')
                        if tag == 'duration':
                            tr_duration = value
                        elif tag == 'termination_type':
                            tr_termination_type = value
                        elif tag == 'termination_id':
                            tr_termination_id = value
                        elif tag == 'corefile':
                            tr_corefile = value
                        elif tag == 'cutime':
                            tr_cutime = value
                        elif tag == 'cstime':
                            tr_cstime = value
                        else:
                            pass

            # Now package it all into a test object and update the totals
            # things we collected (* ignored)
            # tr_tag: a text tag for the test
            # * tr_cmdline: The command line from the test
            # * tr_contacts: unknown, always empty string
            # tr_analysis: ['exit']
            # tr_output: output text
            # * tr_duration: time in seconds (often 0) - not the sum of cutime and cstime, don't know what it is
            # tr_termination_type: [exited]
            # tr_termination_id: nonzero means fail
            # * tr_corefile: [no, yes]
            # tr_cutime: user time in seconds (often 0)
            # tr_cstime: system time in seconds (often 0)

            # check for some things I don't understand well, error for not as expected
            total_tests = total_tests + 1
            if tr_analysis != 'exit':
                tc = api.testcaseType()
                tr_output.append('LOGFILTER: Interpreted as an error because (analysis != exit)')
                iserror = True
                tcerror = api.errorType(message='Unexpected value for analysis tag in test result', type_ = 'Logparse', valueOf_ = "analysis=%s" % tr_analysis)
                tc.error = tcerror
                tc.name = 'Logfilter'
                tc.classname = 'logfilter'
                tc.time = 0
                total_errors = total_errors + 1
                testcases.append(tc)
            elif tr_termination_type not in ['exited', 'signaled']:
                tc = api.testcaseType()
                tr_output.append('LOGFILTER: Interpreted as an error because (termination_type != exited)')
                isError = True
                tcerror = api.errorType(message='Unexpected value for termination type tag in test result', type_ = 'Logparse', valueOf_ = "termination_type=%s" % tr_termination_type)
                tc.error = tcerror
                tc.name = 'Logfilter'
                tc.classname = 'logfilter'
                tc.time = 0
                total_errors = total_errors + 1
                testcases.append(tc)
            elif int(tr_termination_id) != 0:
                tc = api.testcaseType()
                tr_output.append('LOGFILTER: Interpreted as a failure because (termination_id != 0)')
                total_failures = total_failures + 1

                outputtext = ""
                for line in tr_output:
                    outputtext = outputtext + self.escape_string(line) + '\n'

                tcfailure = api.failureType(message='Test tag %s has failed' % tr_tag, type_ = 'Failure', valueOf_ = "%s" % outputtext)
                tc.failure = tcfailure
                tc.name = tr_tag
                tc.classname = 'ltp'
                tc.time = int(tr_cutime) + int(tr_cstime)
                total_time = total_time +  int(tr_cutime) + int(tr_cstime)
                testcases.append(tc)
            else:
                # success, we append the testcase without an error or fail
                tc = api.testcaseType()
                outputtext = ""
                for line in tr_output:
                    outputtext = outputtext + self.escape_string(line) + '\n'
                tc.name = tr_tag
                tc.classname = 'ltp'
                tc.time = int(tr_cutime) + int(tr_cstime)
                total_time = total_time +  int(tr_cutime) + int(tr_cstime)
                testcases.append(tc)

    # output_log_to_dict
    #
    def output_log_to_JUnit(self, stream):
        Dbg.enter("Ltp.output_log_to_JUnit")

        ts = api.testsuite(name='Ubuntu LTP tests')
        # get the properties we want from the log before we parse any tests
        props = self.parse_header(stream)

        properties = api.propertiesType()
        for key, val in props.items():
            tp = api.propertyType(key, val)
            properties.add_property(tp)
        ts.set_properties(properties)

        ts.timestamp = date.isoformat(date.today())
        if 'hostname' in props:
            ts.hostname = props['hostname']
        else:
            ts.hostname = 'localhost'

        # iterate over all test results in the log
        testcases = []
        numtests = 0
        failures = 0
        errors = 0
        time = 0
        nextline = ''
        try:
            numtests, failures, errors, time, nextline, testcases = self.parse_testcases(stream)
        except Exception as ex:
            if self.raiseExceptions:
                raise
            # TODO This is an error, probably unexpected EOF - report it
            tc = api.testcaseType()
            tc.name = 'Logfilter'
            tc.classname = 'logfilter'
            tc.time = 0
            tcerror = api.errorType(message='LOGFILTER: Exception occurred while parsing tests', type_ = 'Logparse', valueOf_ = format_exc())
            tc.error = tcerror
            testcases.append(tc)
            failures = 0
            errors = 1
            time = 0
            nextline = ''

        # TODO deal with case of zero testcases, assume it's an error
        if len(testcases) == 0:
            tc = api.testcaseType()
            tc.name = 'Logfilter'
            tc.classname = 'logfilter'
            tc.time = 0
            tcerror = api.errorType(message='LOGFILTER: No test cases found while parsing log', type_ = 'Logparse', valueOf_ = 'nothing to show')
            tc.error = tcerror
            testcases.append(tc)
        else:
            for tc in testcases:
                ts.add_testcase(tc)

        ts.failures = failures
        ts.errors = errors
        ts.time = time
        ts.tests = numtests

        return ts

# LtpLogFilter
#
class LtpLogFilter():
    # initialize
    #
    def initialize(self):
        """
        A separate initialize that we can control when it gets called (not
        when the object is instantiated).
        """
        defaults = {}                   # A dictionary of application defaults
        defaults['app_name'] = argv[0]  # Used if an exception is thrown processing the command line (in the usage)
        defaults['logs'] = []

        try:
            # Process the command line arguments, and any configuration file options. Make sure
            # that any required parameters have been provided.
            #
            cmdline = Cmdline()
            self.cfg = cmdline.process(argv)
            self.verify_options(self.cfg)

        except CmdlineError as e:
            cmdline.error(e.msg, defaults)
            raise Exit()

        return

    # verify_options
    #
    def verify_options(self, cfg):
        """
        Used to verify that all required parameters are present and correct.
        """
        retval = False

        # At least one log file is required.
        #
        if len(cfg['logs']) == 0:
            CmdlineError("At least one log file must be specified.")

        for fid in cfg['logs']:
            if not path.exists(fid):
                error("The log file (%s) does not exist." % (fid))
                raise Exit

        return

    # main
    #
    def main(self):
        try:
            self.initialize()

            ltp = Ltp()
            if 'raise' in self.cfg:
                ltp.raiseExceptions = True
            testsuites = api.testsuites()
            for fid in self.cfg['logs']:
                with open(fid, 'r') as f:
                    ts = ltp.output_log_to_JUnit(f)
                # TODO Add the stdout to this by re-reading
                f.close()
                outputtext = '<![CDATA['
                with open(fid, 'r') as f:
                    for line in f:
                        outputtext = outputtext + ltp.escape_string(line.strip()) + '\n'
                    outputtext = outputtext + ']]>'
                    ts.set_system_out(outputtext)
                testsuites.add_testsuite(ts)
            testsuites.export(stdout, 0)

        # Handle the user presses <ctrl-C>.
        #
        except KeyboardInterrupt:
            pass

        except Exit:
            pass

        return

if __name__ == '__main__':
    app = LtpLogFilter()
    app.main()

# vi:set ts=4 sw=4 expandtab: