check_xsd.py

import logging

import glob
import sys
import argparse
from pathlib import Path
from pprint import pprint
import os

from lxml import etree
from xml.etree.ElementTree import ParseError
from xmlschema import XMLSchema, XMLSchemaParseError

def BuildSchemaDictonary (fileList):
    if len(fileList) == 0:
        logging.info("No schema files provided")
        return []

    logging.info("Schema locations:")
    schemaLocations = []
    for schemaFile in fileList:
        try:
            xs = XMLSchema(schemaFile, validation='skip')
            schemaLocations.append((xs.target_namespace, str(Path(schemaFile).resolve())))
            logging.info(" [ {0}  ->  {1} ]".format(xs.default_namespace, schemaFile))
        except ParseError as ex:
            logging.warning (" [ {0} failed to parse:  {1} ]".format(schemaFile, ex))
    return schemaLocations


def BuildSchema (coreFile, fileList = None):
    schemaLocations = []
    if fileList and len(fileList) > 0:
        schemaLocations = BuildSchemaDictonary(fileList)

    coreSchema = XMLSchema(str(Path(coreFile)), locations=schemaLocations)
    return coreSchema

def ValidateSingleFile (schemaFile):
    try:
        xs = XMLSchema(schemaFile, validation='skip')
    except ParseError as ex:
        logging.warning (" [ {0} failed to parse:  {1} ]".format(schemaFile, ex))
        return ex
    return None


def ValidateXSDFiles (fileList):
    if len(fileList) == 0:
        logging.info("No schema files provided")
        return {}
    
    schemaLocations = BuildSchemaDictonary(fileList)
    errors = {}
    schemaDictionary = {}

    logging.info("Schema validation:")
    for schemaFile in fileList:
        try:
            schema = XMLSchema(schemaFile, locations = schemaLocations)
            logging.info(schemaFile + ": OK")
            errors[schemaFile] = []
            schemaDictionary[schema.target_namespace] = schema
        except XMLSchemaParseError as ex:
            if (ex.schema_url) and (ex.schema_url != ex.origin_url):
                logging.info("  Error {1} comes from {0}, suppressing".format(ex.schema_url, ex.message))
                errors[schemaFile] = []
            else:
                logging.warning(schemaFile + ": Failed validation ({0})".format(ex))
                errors[schemaFile] = [ex.message]
    return errors, schemaDictionary


def ValidateInstanceDocuments (coreFile, supportingSchemas, instanceDocs):
    if (instanceDocs is None) or len(instanceDocs) == 0:
        logging.warning ("No instance documents provided")
        return []

    schema = BuildSchema(coreFile, supportingSchemas)
    errors = []
    for instanceDoc in instanceDocs:
        try:
            schema.validate(instanceDoc)
            logging.info ("{0} passed validation".format(instanceDoc))
        except Exception as ex:
            logging.error ("{0} failed validation: {1}".format(instanceDoc, ex))
    return errors
    


if __name__ == '__main__':
    parser = argparse.ArgumentParser()
    parser.add_argument("-v", "--verbosity", help="verbosity level", action="count", default=0)
    parser.add_argument("input", help="include a directory or file", action="append", nargs="+")
    parser.add_argument("-p", "--primaryNamespace", help="Primary schema namespace for instance doc validation")
    args = parser.parse_args()

    logging.getLogger().setLevel(logging.WARNING)
    if (args.verbosity >= 1):
        logging.getLogger().setLevel(logging.INFO)
    if (args.verbosity >= 2):
        logging.getLogger().setLevel(logging.DEBUG)
        logging.debug("Very verbose selected")


    logging.debug(f"Path: {args.input}")
    includeFileList = []
    includeInstanceDocList = []
    for path in args.input[0]:
        p = Path(path)
        if not p.exists():
            logging.error(f"Include path {path} not found")
            exit(1)
        if p.is_dir():
            logging.debug(f"Expanding directory")
            for g in glob.glob(os.path.join(str(p), "*.xsd")):
                logging.info(f">Including {g}")
                includeFileList.append(g)
            for g in glob.glob(os.path.join(str(p), "*.xml")):
                logging.info(f">Including instance doc {g}")
                includeInstanceDocList.append(g)
        else:
            logging.info(f">Including {p.absolute()}")
            if str(p.absolute()).endswith('.xml'):
                includeInstanceDocList.append(str(p.absolute()))
            elif str(p.absolute()).endswith('.xsd'):
                includeFileList.append(str(p.absolute()))
            else:
                logging.warning(f'Ignoring file {p.absolute()}')
    
    if len(includeInstanceDocList) and (args.primaryNamespace is None):
        print("Cannot validate instance documents without specifying a primary namespace (use -h for usage guidelines)")
        exit(-1)

    syntaxErrors = 0
    
    print ("=============================")
    print ("XSD syntax checks:")
    print ("-----------------------------")
    for file in includeFileList:
        error = ValidateSingleFile(file)
        if (error):
            print (f"  {file} : Syntax error [{error}]")
            syntaxErrors += 1
        else:
            print (f"  {file} : OK")

    print ("-----------------------------")
    if (syntaxErrors > 0):
        print (f"{syntaxErrors} syntax errors detected")
        exit(syntaxErrors)
    else:
        print ("0 syntax errors detected")

    results, schemaDict = ValidateXSDFiles(includeFileList)
    
    print ("=============================")
    print ("XSD build checks:")
    print ("-----------------------------")
    errorCount = 0
    for fileName, errors in results.items():
        if len(errors) > 0:
            errorCount += len(errors)
            print (f"  {fileName}: {len(errors)} errors")
            for error in errors:
                if isinstance(error, XMLSchemaParseError):
                    print (error.msg)
                else:
                    print (f"      {str(error.strip())}")
        else:
            print (f"  {fileName}: OK")

    print ("-----------------------------")
    print (f"{errorCount} build errors detected")
    if (errorCount > 0):
        exit(errorCount)

    print ("=============================")
    print ("Instance document checks")
    print ("-----------------------------")
    errorCount = 0

    primarySchema = schemaDict[args.primaryNamespace]
    for instanceDoc in includeInstanceDocList:
        try:
            results = primarySchema.validate(instanceDoc)
            print (f"  {instanceDoc} : OK")
        except Exception as ex:
            errorCount += 1
            print (f"  {instanceDoc} : {str(ex)}")
    print ("-----------------------------")
    print (f"{errorCount} instance doc errors detected")
    print ("=============================")
    exit(errorCount)