Files
troupe-standup/wizard-work/spark/json_schema_validator_20241021.py
T

245 lines
8.6 KiB
Python

#!/usr/bin/env python3
"""
Simple JSON Schema Validator - validates JSON files against basic schema constraints.
Created: 2024-10-21
"""
import json
import sys
from pathlib import Path
from typing import Any, Dict, List, Tuple
def validate_type(value: Any, type_name: str) -> bool:
"""Check if value matches the expected JSON type."""
type_validators = {
'string': lambda v: isinstance(v, str),
'number': lambda v: isinstance(v, (int, float)) and not isinstance(v, bool),
'integer': lambda v: isinstance(v, int) and not isinstance(v, bool),
'boolean': lambda v: isinstance(v, bool),
'array': lambda v: isinstance(v, list),
'object': lambda v: isinstance(v, dict),
'null': lambda v: v is None,
}
validator = type_validators.get(type_name)
return validator(value) if validator else False
def validate_against_schema(data: Any, schema: Dict, path: str = "") -> Tuple[bool, List[str]]:
"""Recursively validate data against a JSON schema. Returns (valid, errors)."""
errors = []
# Type checking
if 'type' in schema:
if not validate_type(data, schema['type']):
errors.append(f"{path or '(root)'}: Expected type '{schema['type']}', got '{type(data).__name__}'")
return False, errors
# Object validation
if isinstance(data, dict) and 'properties' in schema:
props = schema['properties']
required = schema.get('required', [])
for prop in required:
if prop not in data:
errors.append(f"{path}.{prop}: Required property missing")
for prop, prop_schema in props.items():
if prop in data:
sub_path = f"{path}.{prop}" if path else prop
valid, sub_errors = validate_against_schema(data[prop], prop_schema, sub_path)
errors.extend(sub_errors)
# Array validation
if isinstance(data, list) and 'items' in schema:
for i, item in enumerate(data):
sub_path = f"{path}[{i}]"
valid, sub_errors = validate_against_schema(item, schema['items'], sub_path)
errors.extend(sub_errors)
# Enum validation
if 'enum' in schema:
if data not in schema['enum']:
errors.append(f"{path or '(root)'}: Value '{data}' not in enum {schema['enum']}")
# String constraints
if isinstance(data, str):
if 'minLength' in schema and len(data) < schema['minLength']:
errors.append(f"{path or '(root)'}: String length {len(data)} < minLength {schema['minLength']}")
if 'maxLength' in schema and len(data) > schema['maxLength']:
errors.append(f"{path or '(root)'}: String length {len(data)} > maxLength {schema['maxLength']}")
if 'pattern' in schema:
import re
if not re.match(schema['pattern'], data):
errors.append(f"{path or '(root)'}: String '{data}' doesn't match pattern '{schema['pattern']}'")
# Numeric constraints
if isinstance(data, (int, float)) and not isinstance(data, bool):
if 'minimum' in schema and data < schema['minimum']:
errors.append(f"{path or '(root)'}: Value {data} < minimum {schema['minimum']}")
if 'maximum' in schema and data > schema['maximum']:
errors.append(f"{path or '(root)'}: Value {data} > maximum {schema['maximum']}")
return len(errors) == 0, errors
def validate_json_file(json_path: str, schema_path: str) -> bool:
"""Validate a JSON file against a schema file."""
try:
with open(json_path) as f:
data = json.load(f)
except FileNotFoundError:
print(f"Error: JSON file not found: {json_path}")
return False
except json.JSONDecodeError as e:
print(f"Error: Invalid JSON in {json_path}: {e}")
return False
try:
with open(schema_path) as f:
schema = json.load(f)
except FileNotFoundError:
print(f"Error: Schema file not found: {schema_path}")
return False
except json.JSONDecodeError as e:
print(f"Error: Invalid JSON in {schema_path}: {e}")
return False
valid, errors = validate_against_schema(data, schema)
if valid:
print(f"VALID: {json_path} conforms to schema")
return True
else:
print(f"INVALID: {json_path} does not conform to schema")
for err in errors:
print(f" {err}")
return False
def run_self_test() -> bool:
"""Run self-validation tests against known test files."""
import tempfile
import os
test_passed = 0
test_failed = 0
# Test 1: Valid JSON against valid schema
schema1 = {
"type": "object",
"properties": {
"name": {"type": "string"},
"value": {"type": "integer", "minimum": 0}
},
"required": ["name", "value"]
}
data1 = {"name": "test", "value": 42}
valid, _ = validate_against_schema(data1, schema1)
if valid:
print("Test 1 PASSED: Valid object with required properties")
test_passed += 1
else:
print("Test 1 FAILED: Valid object should pass")
test_failed += 1
# Test 2: Missing required property
data2 = {"name": "test"}
valid, errors = validate_against_schema(data2, schema1)
if not valid and any("value" in e for e in errors):
print("Test 2 PASSED: Missing required property detected")
test_passed += 1
else:
print("Test 2 FAILED: Should detect missing 'value'")
test_failed += 1
# Test 3: Wrong type
data3 = {"name": "test", "value": "not a number"}
valid, errors = validate_against_schema(data3, schema1)
if not valid and any("integer" in e for e in errors):
print("Test 3 PASSED: Wrong type detected")
test_passed += 1
else:
print("Test 3 FAILED: Should detect wrong type")
test_failed += 1
# Test 4: Value below minimum
data4 = {"name": "test", "value": -5}
valid, errors = validate_against_schema(data4, schema1)
if not valid and any("minimum" in e for e in errors):
print("Test 4 PASSED: Value below minimum detected")
test_passed += 1
else:
print("Test 4 FAILED: Should detect value below minimum")
test_failed += 1
# Test 5: Array validation
schema5 = {
"type": "object",
"properties": {
"items": {
"type": "array",
"items": {"type": "string"}
}
}
}
data5 = {"items": ["a", "b", "c"]}
valid, _ = validate_against_schema(data5, schema5)
if valid:
print("Test 5 PASSED: Valid array of strings")
test_passed += 1
else:
print("Test 5 FAILED: Valid array should pass")
test_failed += 1
# Test 6: Array with wrong item type
data6 = {"items": ["a", 123, "c"]}
valid, errors = validate_against_schema(data6, schema5)
if not valid and any("[1]" in e for e in errors):
print("Test 6 PASSED: Array item type mismatch detected")
test_passed += 1
else:
print("Test 7 FAILED: Should detect array item type mismatch")
test_failed += 1
# Test 7: Enum validation
schema7 = {
"type": "object",
"properties": {
"status": {"enum": ["active", "inactive", "pending"]}
}
}
data7 = {"status": "active"}
valid, _ = validate_against_schema(data7, schema7)
if valid:
print("Test 7 PASSED: Valid enum value")
test_passed += 1
else:
print("Test 7 FAILED: Valid enum should pass")
test_failed += 1
# Test 8: Invalid enum value
data8 = {"status": "unknown"}
valid, errors = validate_against_schema(data8, schema7)
if not valid and any("enum" in e for e in errors):
print("Test 8 PASSED: Invalid enum value detected")
test_passed += 1
else:
print("Test 8 FAILED: Should detect invalid enum value")
test_failed += 1
print(f"\nSelf-test results: {test_passed} passed, {test_failed} failed")
return test_failed == 0
def main():
if len(sys.argv) >= 2 and sys.argv[1] == '--self-test':
success = run_self_test()
sys.exit(0 if success else 1)
if len(sys.argv) < 3:
print("Usage: json_schema_validator.py <json_file> <schema_file>")
print(" json_schema_validator.py --self-test")
sys.exit(1)
json_path = sys.argv[1]
schema_path = sys.argv[2]
success = validate_json_file(json_path, schema_path)
sys.exit(0 if success else 1)
if __name__ == '__main__':
main()