def _validate_payload_schema()

in azext_iot/operations/hub.py [0:0]

• 55 lines of code
• 10 McCabe index (conditional complexity)

def _validate_payload_schema(content):
    import json
    from os.path import join
    from azext_iot.models.validators import JsonSchemaType, JsonSchemaValidator
    from azext_iot.constants import EDGE_DEPLOYMENT_ROOT_SCHEMAS_PATH as root_schema_path
    from azext_iot.common.utility import shell_safe_json_parse

    EDGE_AGENT_SCHEMA_PATH = "azure-iot-edgeagent-deployment-{}.json"
    EDGE_HUB_SCHEMA_PATH = "azure-iot-edgehub-deployment-{}.json"
    EDGE_SCHEMA_PATH_DICT = {
        "$edgeAgent": EDGE_AGENT_SCHEMA_PATH,
        "$edgeHub": EDGE_HUB_SCHEMA_PATH,
    }

    modules_content = content["modulesContent"]
    system_modules_for_validation = ["$edgeAgent", "$edgeHub"]

    for sys_module in system_modules_for_validation:
        if sys_module in modules_content:
            if (
                "properties.desired" in modules_content[sys_module]
                and "schemaVersion"
                in modules_content[sys_module]["properties.desired"]
            ):
                target_schema_ver = modules_content[sys_module][
                    "properties.desired"
                ]["schemaVersion"]
                target_schema_def_path = join(root_schema_path, f"{EDGE_SCHEMA_PATH_DICT[sys_module].format(target_schema_ver)}")

                logger.info("Attempting to fetch schema content from %s...", target_schema_def_path)
                if not exists(target_schema_def_path):
                    logger.info("Invalid schema path %s, skipping validation...", target_schema_def_path)
                    continue

                try:
                    target_schema_def = str(read_file_content(target_schema_def_path))
                    target_schema_def = shell_safe_json_parse(target_schema_def)
                except Exception:
                    logger.info(
                        "Unable to fetch schema content from %s skipping validation...",
                        target_schema_def_path,
                    )
                    continue

                logger.info(f"Validating {sys_module} of deployment payload against schema...")
                to_validate_content = {
                    sys_module: modules_content[sys_module]
                }
                draft_version = JsonSchemaType.draft4
                if "$schema" in target_schema_def and "/draft-07/" in target_schema_def["$schema"]:
                    draft_version = JsonSchemaType.draft7

                v = JsonSchemaValidator(target_schema_def, draft_version)
                errors = v.validate(to_validate_content)
                if errors:
                    # Pretty printing schema validation errors
                    raise ValidationError(
                        json.dumps(
                            {"validationErrors": errors},
                            separators=(",", ":"),
                            indent=2,
                        )
                    )