validate.py

"""
SFCC Schema Validator
Author: Charles Lavery <clavery@salesforce.com>

Requires packages:
    - lxml
    - jsonschema

Place XSD and JSON schemas in a `./schemas` subdirectory relative to this file (download from 
https://documentation.b2c.commercecloud.salesforce.com/DOC3/index.jsp?topic=%2Fcom.demandware.dochelp%2FDWAPI%2Fxsd%2FSchemas.html 
and https://documentation.b2c.commercecloud.salesforce.com/DOC3/topic/com.demandware.dochelp/DWAPI/content/JsonSchemas.html?resultof=%22%6a%73%6f%6e%22%20%22%73%63%68%65%6d%61%22%20)

then run

./validate.py [file_or_directory]

where file or directory is an xml file, a directory of files of XML and JSON PD components
"""
import sys

from lxml import etree as ET
import os
import re
import json

from jsonschema import validate as validate_jsonschema
import jsonschema

SCHEMA_MAP = {
    "abtest" : "abtest.xsd",
    "bmext" : "bmext.xsd",
    "cache-settings" : "cachesettings.xsd",
    "catalog" : "catalog.xsd",
    "coupons" : "coupon.xsd",
    "coupon-redemptions" : "couponredemption.xsd",
    "customers" : "customer.xsd",
    "customer-groups" : "customergroup.xsd",
    "customer-lists" : "customerlist.xsd",
    "customer-list" : "customerlist2.xsd",
    "customerpaymentinstrument" : "customerpaymentinstrument.xsd",
    "custom-objects" : "customobject.xsd",
    "csrf-whitelists" : "csrfwhitelists.xsd",
    "extensions" : "bmext.xsd",
    "feeds" : "feed.xsd",
    "form" : "form.xsd",
    "geolocations" : "geolocation.xsd",
    "gift-certificates" : "giftcertificate.xsd",
    "inventory" : "inventory.xsd",
    "library" : "library.xsd",
    "locales" : "locales.xsd",
    "metadata" : "metadata.xsd",
    "oauth" : "oauth.xsd",
    "orders" : "order.xsd",
    "payment-settings" : "paymentmethod.xsd",
    "payment-processors" : "paymentprocessor.xsd",
    "preferences" : "preferences.xsd",
    "pricebooks" : "pricebook.xsd",
    "product-lists" : "productlist.xsd",
    "promotions" : "promotion.xsd",
    "redirect-urls" : "redirecturl.xsd",
    "schedules" : "schedules.xsd",
    "services" : "services.xsd",
    "shipping" : "shipping.xsd",
    "site" : "site.xsd",
    "slot-configurations" : "slot.xsd",
    "sitemap-configuration" : "sitemapconfiguration.xsd",
    "sort" : "sort.xsd",
    "sourcecodes" : "sourcecode.xsd",
    "stores" : "store.xsd",
    "tax" : "tax.xsd",
    "url-rules" : "urlrules.xsd",
    "xml" : "xml.xsd",
    "jobs" : "jobs.xsd",
    "page-meta-tags" : "pagemetatag.xsd",
}

NSMAP = {
    "http://www.demandware.com/xml/impex/search/2007-02-28" : "search.xsd",
    "http://www.demandware.com/xml/impex/search2/2010-02-19" : "search2.xsd",
}

JSON_SCHEMAS = [
    "attributedefinition.json",
    "attributedefinitiongroup.json",
    "common.json",
    "componentconstructor.json",
    "componenttype.json",
    "componenttypeexclusion.json",
    "contentassetcomponentconfig.json",
    "contentassetcomponentdata.json",
    "contentassetpageconfig.json",
    "customeditortype.json",
    "editordefinition.json",
    "image.json",
    "pagetype.json",
    "regiondefinition.json",
    "visibilityrule.json",
]

def load_json_schema_ref_resolver():
    store = {}
    for schema in JSON_SCHEMAS:
        store[schema] = json.load(open(os.path.join(os.path.dirname(os.path.realpath(__file__)), 'schemas/' + schema)))
    return jsonschema.RefResolver(base_uri='', referrer='', store=store)

def validate_xml(xml, throw=True):
    """Validates XML against DWRE schemas"""

    root_el = xml.getroot()
    root_tag = root_el.tag[root_el.tag.find('}')+1:]
    schema_name = None
    if None in root_el.nsmap:
        schema_name = NSMAP.get(root_el.nsmap[None], SCHEMA_MAP.get(root_tag))
    elif root_tag in SCHEMA_MAP:
        schema_name = SCHEMA_MAP.get(root_tag)

    if not schema_name:
        return

    schema = ET.XMLSchema(file=os.path.join(os.path.dirname(os.path.realpath(__file__)), 'schemas', schema_name))
    schema.validate(xml)
    if throw:
        schema.assertValid(xml)
    return schema


SCHEMALINE_RE = re.compile(r'^(.*?:\d+:\d+:)')
def validate_file(full_filename):
    print("%s"  % full_filename, end=' ')
    try:
        xml = ET.parse(full_filename)
        schema = validate_xml(xml, throw=False)
        if schema is None:
            print("no schema [WARNING]")
            return True
        elif schema.error_log:
            print("[ERROR]")
            for e in schema.error_log:
                print(SCHEMALINE_RE.sub('\\1', str(e)))
            return False
        else:
            print("[OK]")
            return True
    except ET.XMLSchemaParseError as e:
        print("bad schema [WARNING]")
        return True
    except ET.XMLSyntaxError as e:
        print("[ERROR]")
        print(e)
        return False

def validate_json_schemas(directory):
    results = []
    ref_resolver = load_json_schema_ref_resolver()
    for (dirpath, dirnames, filenames) in os.walk(directory):
        for fname in filenames:
            (root, ext) = os.path.splitext(fname)

            if ext == ".json":
                full_filename = os.path.join(dirpath, fname)
                result = validate_json_schema(full_filename, ref_resolver=ref_resolver)
                results.append(result)

    return all(results)

def validate_json_schema(full_filename, ref_resolver=None):
    if not ref_resolver:
        ref_resolver = load_json_schema_ref_resolver()
    try:
        if os.path.normpath("experience/components") in os.path.normpath(full_filename):
            print("%s"  % full_filename, end=' ')
            # validate component
            component_schema = json.load(open(os.path.join(os.path.dirname(os.path.realpath(__file__)), 'schemas/componenttype.json')))
            with open(full_filename, "r") as f:
                instance = json.load(f) # TODO: check json failure
                validate_jsonschema(instance=instance, schema=component_schema,
                                    resolver=ref_resolver)
            print("[OK]")
            return True
        elif os.path.normpath("experience/pages") in os.path.normpath(full_filename):
            print("%s"  % full_filename, end=' ')
            # validate component
            component_schema = json.load(open(os.path.join(os.path.dirname(os.path.realpath(__file__)), 'schemas/pagetype.json')))
            with open(full_filename, "r") as f:
                instance = json.load(f) # TODO: check json failure
                validate_jsonschema(instance=instance, schema=component_schema,
                                    resolver=ref_resolver)
            print("[OK]")
            return True
        return True
    except json.decoder.JSONDecodeError as e:
        print("[ERROR] Cannot Decode JSON")
        print(e.msg)
        return False
    except jsonschema.exceptions.ValidationError as e:
        print("[ERROR] jsonschema validation error")
        print(e.message)
        return False

def validate_directory(directory):
    results = []
    for (dirpath, dirnames, filenames) in os.walk(directory):
        for fname in filenames:
            (root, ext) = os.path.splitext(fname)

            if ext == ".xml":
                full_filename = os.path.join(dirpath, fname)
                result = validate_file(full_filename)
                results.append(result)

    return all(results)

def main(target):
    results = []
    if os.path.isdir(target):
        results.append(validate_directory(target))
        results.append(validate_json_schemas(target))
    elif os.path.isfile(target):
        (root, ext) = os.path.splitext(target)
        if ext == ".json":
            results.append(validate_json_schema(target))
        else:
            results.append(validate_file(target))
    else:
        raise IOError("file not found")
    if not all(results):
        raise RuntimeError("Not all files validated")

if len(sys.argv) < 2:
    print("Usage: %s [file_or_directory]" % sys.argv[0])
else:
    main(sys.argv[1])