libxml2/xstc/xstc.py

#!/usr/bin/env python

#
# This is the MS subset of the W3C test suite for XML Schemas.
# This file is generated from the MS W3c test suite description file.
#

import sys, os
import exceptions, optparse
import libxml2

opa = optparse.OptionParser()

opa.add_option("-b", "--base", action="store", type="string", dest="baseDir",
               default="",
               help="""The base directory; i.e. the parent folder of the
               "nisttest", "suntest" and "msxsdtest" directories.""")

opa.add_option("-o", "--out", action="store", type="string", dest="logFile",
               default="test.log",
               help="The filepath of the log file to be created")
               
opa.add_option("--no-log", action="store_true", dest="disableLog",
               default=False,
               help="The filepath of the log file to be created")    
               
opa.add_option("--no-test-out", action="store_true", dest="disableTestStdOut",
               default=False,
               help="The filepath of the log file to be created")                           

opa.add_option("-s", "--silent", action="store_true", dest="silent", default=False,
               help="Disables display of all tests")

opa.add_option("-v", "--verbose", action="store_true", dest="verbose",
               default=False,
               help="Displays all tests (only if --silent is not set)")

opa.add_option("-x", "--max", type="int", dest="maxTestCount",
               default="-1",
               help="The maximum number of tests to be run")

opa.add_option("-t", "--test", type="string", dest="singleTest",
               default=None,
               help="Runs the specified test only")
               
opa.add_option("--rieo", "--report-internal-errors-only", action="store_true",
               dest="reportInternalErrOnly", default=False,
               help="Display erroneous tests of type 'internal' only")
               
opa.add_option("--rmleo", "--report-mem-leak-errors-only", action="store_true",
               dest="reportMemLeakErrOnly", default=False,
               help="Display erroneous tests of type 'memory leak' only")

opa.add_option("-c", "--combines", type="string", dest="combines",
               default=None,
               help="Combines to be run (all if omitted)")

opa.add_option("--rc", "--report-combines", action="store_true",
               dest="reportCombines", default=False,
               help="Display combine reports")

opa.add_option("--rec", "--report-err-combines", action="store_true",
               dest="reportErrCombines", default=False,
               help="Display erroneous combine reports only")

opa.add_option("--debug", action="store_true",
               dest="debugEnabled", default=False,
               help="Displays debug messages")
               
opa.add_option("--info", action="store_true",
               dest="info", default=False,
               help="Displays info on the suite only. Does not run any test.")            

(options, args) = opa.parse_args()

if options.combines is not None:
    options.combines = options.combines.split()
    
################################################
# The vars below are not intended to be changed.
#

msgSchemaNotValidButShould =  "The schema should be valid."
msgSchemaValidButShouldNot = "The schema should be invalid."
msgInstanceNotValidButShould = "The instance should be valid."
msgInstanceValidButShouldNot = "The instance should be invalid."
testFolderNIST = "nisttest"
testFolderMS   = "msxsdtest"
testFolderSUN  = "suntest"

###################
# Helper functions.
#

def handleError(test, msg):
    test.addLibLog("'%s'   LIB: %s" % (test.name, msg))
    if msg.find("Unimplemented") > -1:
        test.failUnimplemented()
    elif msg.find("Internal") > -1:
        test.failInternal()        
    

##################
# Test case class.
#

class MSTestCase:
           
    def __init__(self, name, descr, tFolder, sFolder, sFile, sVal, iExists, iFolder, iFile, iVal):
        global testFolderNIST, testFolderSUN, testFolderMS
        #
        # Init.
        #
        self.name = name
        self.descr = descr
        self.test_Folder = tFolder
        self.schema_Folder = sFolder
        self.schema_File = sFile
        self.schema_Val = sVal
        self.instance_Exists = iExists
        self.instance_Folder = iFolder
        self.instance_File = iFile
        self.instance_Val = iVal
        self.failed = False
        self.log = []
        self.libLog = []
        self.phase = ""
        self.initialMemUsed = 0
        self.memLeak = 0
        self.excepted = False
        self.bad = False
        self.unimplemented = False
        self.internalErr = False
        #
        # Compute combine name of this test.
        #       
        if self.test_Folder == testFolderMS or self.test_Folder == testFolderSUN:
            #
            # Use the last given directory for the combine name.
            #
            dirs = self.schema_Folder.split("/")
            self.combineName = dirs[len(dirs) -1]
	    if self.test_Folder == testFolderMS:
	        if self.combineName == "group":		    
	            self.schema_Folder = "Group"
		    self.instance_Folder = "Group"
        elif self.test_Folder == testFolderNIST:
            #
            # NIST files are named in the following form: 
            # "NISTSchema-short-pattern-1.xsd"
            #
            tokens = self.schema_File.split("-")
            self.combineName = tokens[1]            
        else:
            self.combineName = "unkown"
            raise Exception("Could not compute the combine name of a test.")
        #
        # Init the log.
        #
        self.log.append("'%s'   descr: %s\n" % (self.name, self.descr))
        self.log.append("'%s'   exp schema   valid: %d\n" % (self.name, self.schema_Val))
        if (self.instance_Exists):
            self.log.append("'%s'   exp instance valid: %d\n" % (self.name, self.instance_Val))                       
       
    def addLibLog(self, msg):
        """This one is intended to be used by the error handler
        function"""
        self.libLog.append(msg)

    def fail(self, msg):       
        self.failed = True         
        self.log.append("'%s' ( FAILED: %s\n" % (self.name, msg))
        
    def failInternal(self):
        self.failed = True
        self.internalErr = True
        self.log.append("'%s' * INTERNAL\n" % self.name)
        
    def failUnimplemented(self):
        self.failed = True
        self.unimplemented = True
        self.log.append("'%s' ? UNIMPLEMENTED\n" % self.name)

    def failCritical(self, msg):        
        self.failed = True
        self.bad = True
        self.log.append("'%s' ! BAD: %s\n" % (self.name, msg))  

    def failExcept(self, e):      
        self.failed = True
        self.excepted = True
        self.log.append("'%s' # EXCEPTION: %s\n" % (self.name, e.__str__()))
    
    def setUp(self):            
        #
        # Set up Libxml2.
        #   
        self.initialMemUsed = libxml2.debugMemory(1)
        libxml2.initParser()
        libxml2.lineNumbersDefault(1)
        libxml2.registerErrorHandler(handleError, self)
        
    def tearDown(self):        
        libxml2.schemaCleanupTypes()
        libxml2.cleanupParser()      
        self.memLeak = libxml2.debugMemory(1) - self.initialMemUsed

    def isIOError(self, file, docType):
        err = None
        try:
            err = libxml2.lastError()
        except:
            # Suppress exceptions.
            pass
        if (err is None):
            return False
        if err.domain() == libxml2.XML_FROM_IO:
            self.failCritical("failed to access the %s resource '%s'\n" % (docType, file))

    def debugMsg(self, msg):
        global options 
        if options.debugEnabled:
            sys.stdout.write("'%s'   DEBUG: %s\n" % (self.name, msg))
            
    def finalize(self):
        """Adds additional info to the log."""
        #
        # Add libxml2 messages.
        #
        self.log.extend(self.libLog)
        #
        # Add memory leaks.
        #        
        if self.memLeak != 0:            
            self.log.append("%s + memory leak: %d bytes\n" % (self.name, self.memLeak))
            
    def processSchema(self, filePath):
        global msgSchemaNotValidButShould, msgSchemaValidButShouldNot
        schema = None
        
        #
        # Parse the schema.
        #
        self.debugMsg("loading schema: %s" % filePath)
        schema_ParserCtxt = libxml2.schemaNewParserCtxt(filePath)
        try:
            try:
                schema = schema_ParserCtxt.schemaParse()
            except:
                pass
        finally:
            self.debugMsg("after loading schema")
            del schema_ParserCtxt
        if schema is None:
            self.debugMsg("schema is None")
            self.debugMsg("checking for IO errors...")
            if self.isIOError(file, "schema"):
                return None
        self.debugMsg("checking schema result")
        if (schema is None and self.schema_Val) or (schema is not None and self.schema_Val == 0):
            self.debugMsg("schema result is BAD")
            if (schema == None):
                self.fail(msgSchemaNotValidButShould)
            else:
                self.fail(msgSchemaValidButShouldNot)
        else:
	    self.debugMsg("schema result is OK")
            return schema

    def processInstance(self, filePath, schema):
        global msgInstanceNotValidButShould, msgInstanceValidButShouldNot
        
        instance = None
        self.debugMsg("loading instance: %s" % filePath)            
        instance_parserCtxt = libxml2.newParserCtxt()
        if (instance_parserCtxt is None):
            # TODO: Is this one necessary, or will an exception 
            # be already raised?
            raise Exception("Could not create the instance parser context.")
        try:
            try:
                instance = instance_parserCtxt.ctxtReadFile(filePath, None, libxml2.XML_PARSE_NOWARNING)
            except:
                # Suppress exceptions.
                pass
        finally:
            del instance_parserCtxt
        self.debugMsg("after loading instance")
        if instance is None:
            self.debugMsg("instance is None")
            self.failCritical("Failed to parse the instance for unknown reasons.")
            return
        else:
            try:
                #
                # Validate the instance.
                #
		
                validation_Ctxt = schema.schemaNewValidCtxt()
		#validation_Ctxt = libxml2.schemaNewValidCtxt(None)
                if (validation_Ctxt is None):
                    self.failCritical("Could not create the validation context.")
                    return
                try:
                    self.debugMsg("validating instance") 
                    instance_Err = validation_Ctxt.schemaValidateDoc(instance)
                    self.debugMsg("after instance validation") 
                    self.debugMsg("instance-err: %d" % instance_Err)
                    if (instance_Err != 0 and self.instance_Val == 1) or (instance_Err == 0 and self.instance_Val == 0):
                        self.debugMsg("instance result is BAD")
                        if (instance_Err != 0):
                            self.fail(msgInstanceNotValidButShould)
                        else:
                            self.fail(msgInstanceValidButShouldNot)
                            
                    else:                        
                                self.debugMsg("instance result is OK")
                finally:
                    del validation_Ctxt
            finally:
                instance.freeDoc()
            

    def run(self):
        """Runs a test.""" 
        global options
        
        # os.path.join(options.baseDir, self.test_Folder, self.schema_Folder, self.schema_File)
        filePath = "%s/%s/%s/%s" % (options.baseDir, self.test_Folder, self.schema_Folder, self.schema_File)
        schema = None
        try:                
            schema = self.processSchema(filePath)
            try:
                if self.instance_Exists and (schema is not None) and (not self.failed):
                    filePath = "%s/%s/%s/%s" % (options.baseDir, self.test_Folder, self.instance_Folder, self.instance_File)
                    self.processInstance(filePath, schema)
            finally:
                if schema is not None:
                   del schema

        except (Exception, libxml2.parserError, libxml2.treeError), e:
            self.failExcept(e)

            
####################
# Test runner class.
#
              
class MSTestRunner:

    CNT_TOTAL = 0
    CNT_RAN = 1
    CNT_SUCCEEDED = 2
    CNT_FAILED = 3
    CNT_UNIMPLEMENTED = 4
    CNT_INTERNAL = 5
    CNT_BAD = 6
    CNT_EXCEPTED = 7
    CNT_MEMLEAK = 8

    def __init__(self):
        self.logFile = None
        self.counters = self.createCounters()
        self.testList = []
        self.combinesRan = {}
        
    def createCounters(self):
        counters = {self.CNT_TOTAL:0, self.CNT_RAN:0, self.CNT_SUCCEEDED:0,
        self.CNT_FAILED:0, self.CNT_UNIMPLEMENTED:0, self.CNT_INTERNAL:0, self.CNT_BAD:0, 
        self.CNT_EXCEPTED:0, self.CNT_MEMLEAK:0}
        
        return counters

    def addTest(self, test):
        self.testList.append(test)
        
    def updateCounters(self, test, counters):
        if test.memLeak != 0:
           counters[self.CNT_MEMLEAK] += 1
        if not test.failed:
           counters[self.CNT_SUCCEEDED] +=1
        if test.failed:
           counters[self.CNT_FAILED] += 1
        if test.bad:
           counters[self.CNT_BAD] += 1
        if test.unimplemented:
           counters[self.CNT_UNIMPLEMENTED] += 1   
        if test.internalErr:
           counters[self.CNT_INTERNAL] += 1                      
        if test.excepted:
           counters[self.CNT_EXCEPTED] += 1
        return counters
           
    def displayResults(self, out, all, combName, counters):
        out.write("\n")
        if all:
            if options.combines is not None:
                out.write("combine(s): %s\n" % str(options.combines))
        elif combName is not None:             
            out.write("combine : %s\n" % combName)
        out.write("  total             : %d\n" % counters[self.CNT_TOTAL])
        if all or options.combines is not None:
            out.write("    ran             : %d\n" % counters[self.CNT_RAN])
        # out.write("    succeeded       : %d\n" % counters[self.CNT_SUCCEEDED])
        if counters[self.CNT_FAILED] > 0:
            out.write("    failed          : %d\n" % counters[self.CNT_FAILED])
            out.write("     -> internal    : %d\n" % counters[self.CNT_INTERNAL])
            out.write("     -> unimpl.     : %d\n" % counters[self.CNT_UNIMPLEMENTED])
            out.write("     -> bad         : %d\n" % counters[self.CNT_BAD])            
            out.write("     -> exceptions  : %d\n" % counters[self.CNT_EXCEPTED])
        if counters[self.CNT_MEMLEAK] > 0:
            out.write("    memory leaks    : %d\n" % counters[self.CNT_MEMLEAK])

    def displayShortResults(self, out, all, combName, counters):
        out.write("Ran %d of %d tests:" % (counters[self.CNT_RAN],
                  counters[self.CNT_TOTAL]))
        # out.write("    succeeded       : %d\n" % counters[self.CNT_SUCCEEDED])
        if counters[self.CNT_FAILED] > 0 or counters[self.CNT_MEMLEAK] > 0:
            out.write(" %d failed" % (counters[self.CNT_FAILED]))
            if counters[self.CNT_INTERNAL] > 0:
                out.write(" %d internal" % (counters[self.CNT_INTERNAL]))
            if counters[self.CNT_UNIMPLEMENTED] > 0:
                out.write(" %d unimplemented" % (counters[self.CNT_UNIMPLEMENTED]))
            if counters[self.CNT_BAD] > 0:
                out.write(" %d bad" % (counters[self.CNT_BAD]))
            if counters[self.CNT_EXCEPTED] > 0:
                out.write(" %d exception" % (counters[self.CNT_EXCEPTED]))
            if counters[self.CNT_MEMLEAK] > 0:
                out.write(" %d leaks" % (counters[self.CNT_MEMLEAK]))
            out.write("\n")
        else:
            out.write(" all passed\n")
    
    def reportCombine(self, combName):
        global options
        
        counters = self.createCounters()
        #
        # Compute evaluation counters.
        #
        for test in self.combinesRan[combName]:
            counters[self.CNT_TOTAL] += 1
            counters[self.CNT_RAN] += 1
            counters = self.updateCounters(test, counters)
        if options.reportErrCombines and (counters[self.CNT_FAILED] == 0) and (counters[self.CNT_MEMLEAK] == 0):
            pass
        else:
            if not options.disableLog:
                self.displayResults(self.logFile, False, combName, counters)
            self.displayResults(sys.stdout, False, combName, counters)
        
    def displayTestLog(self, test):
        sys.stdout.writelines(test.log)
        sys.stdout.write("~~~~~~~~~~\n")
    
    def reportTest(self, test):
        global options
        
        error = test.failed or test.memLeak != 0
        #
        # Only erroneous tests will be written to the log,
        # except @verbose is switched on.
        #        
        if not options.disableLog and (options.verbose or error):
            self.logFile.writelines(test.log)
            self.logFile.write("~~~~~~~~~~\n")
        #
        # if not @silent, only erroneous tests will be
        # written to stdout, except @verbose is switched on.
        #
        if not options.silent: 
            if options.reportInternalErrOnly and test.internalErr:
                self.displayTestLog(test)
            if options.reportMemLeakErrOnly and test.memLeak != 0: 
                self.displayTestLog(test)
            if (options.verbose or error) and (not options.reportInternalErrOnly) and (not options.reportMemLeakErrOnly):
                self.displayTestLog(test)
                
    def addToCombines(self, test):
        found = False
        if self.combinesRan.has_key(test.combineName):
            self.combinesRan[test.combineName].append(test)
        else:
            self.combinesRan[test.combineName] = [test]

    def run(self):

        global options
        
        if options.info:
            for test in self.testList:
                self.addToCombines(test)               
            sys.stdout.write("Combines: %d\n" % len(self.combinesRan))
            sys.stdout.write("%s\n" % self.combinesRan.keys())
            return
        
        if not options.disableLog:
            self.logFile = open(options.logFile, "w")
        try:
            for test in self.testList:
                self.counters[self.CNT_TOTAL] += 1
                #
                # Filter tests.
                #   
                if options.singleTest is not None and options.singleTest != "":
                    if (test.name != options.singleTest):
                        continue
                elif options.combines is not None:
                    if not options.combines.__contains__(test.combineName):
                        continue
                if options.maxTestCount != -1 and self.counters[self.CNT_RAN] >= options.maxTestCount:
                    break
                self.counters[self.CNT_RAN] += 1
                #
                # Run the thing, dammit.
                #
                try:
                    test.setUp()
                    try:
                        test.run()
                    finally:
                        test.tearDown()
                finally:
                    #
                    # Evaluate.
                    #
                    test.finalize()
                    self.reportTest(test)
                    if options.reportCombines or options.reportErrCombines:
                        self.addToCombines(test)
                    self.counters = self.updateCounters(test, self.counters)
        finally:        
            if options.reportCombines or options.reportErrCombines:
                #
                # Build a report for every single combine.
                #
                # TODO: How to sort a dict?
                #
                self.combinesRan.keys().sort(None)
                for key in self.combinesRan.keys():
                    self.reportCombine(key)
            
            #
            # Display the final report.
            #
            if options.silent:
                self.displayShortResults(sys.stdout, True, None, self.counters)
            else:
                sys.stdout.write("===========================\n")
                self.displayResults(sys.stdout, True, None, self.counters)