Files
yovision/contracts/tests/events-v1/contract_support.py
T

92 lines
4.5 KiB
Python

"""Small dependency-free validator for the JSON Schema keywords used by v1 contracts."""
from __future__ import annotations
import json
import math
import re
from datetime import datetime
from pathlib import Path
from typing import Any
def load_json(path: Path) -> Any:
return json.loads(path.read_text(encoding="utf-8"))
def canonical_bytes(value: Any) -> bytes:
"""Canonical bytes for checked-in JCS vectors (all vector numbers are JCS-safe)."""
return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"), allow_nan=False).encode("utf-8")
def validate(instance: Any, schema: dict[str, Any], schema_path: Path, location: str = "$") -> list[str]:
if "$ref" in schema:
ref = schema["$ref"]
if ref.startswith("#"):
return [f"{location}: local fragments are not supported by the contract checker"]
target = (schema_path.parent / ref).resolve()
return validate(instance, load_json(target), target, location)
errors: list[str] = []
for subschema in schema.get("allOf", []):
errors.extend(validate(instance, subschema, schema_path, location))
if "anyOf" in schema and not any(not validate(instance, item, schema_path, location) for item in schema["anyOf"]):
errors.append(f"{location}: does not match anyOf")
if "not" in schema and not validate(instance, schema["not"], schema_path, location):
errors.append(f"{location}: matches forbidden schema")
if "if" in schema and not validate(instance, schema["if"], schema_path, location):
errors.extend(validate(instance, schema.get("then", {}), schema_path, location))
expected = schema.get("type")
type_ok = {
"object": lambda x: isinstance(x, dict),
"array": lambda x: isinstance(x, list),
"string": lambda x: isinstance(x, str),
"integer": lambda x: isinstance(x, int) and not isinstance(x, bool),
"number": lambda x: isinstance(x, (int, float)) and not isinstance(x, bool) and math.isfinite(x),
"boolean": lambda x: isinstance(x, bool),
}
if expected and (expected not in type_ok or not type_ok[expected](instance)):
return errors + [f"{location}: expected {expected}"]
if "const" in schema and instance != schema["const"]:
errors.append(f"{location}: expected constant {schema['const']!r}")
if "enum" in schema and instance not in schema["enum"]:
errors.append(f"{location}: value not in enum")
if isinstance(instance, dict):
required = schema.get("required", [])
errors.extend(f"{location}: missing {name}" for name in required if name not in instance)
properties = schema.get("properties", {})
if schema.get("additionalProperties") is False:
errors.extend(f"{location}: unknown property {name}" for name in instance if name not in properties)
for name, value in instance.items():
if name in properties:
errors.extend(validate(value, properties[name], schema_path, f"{location}.{name}"))
elif isinstance(instance, list):
if len(instance) < schema.get("minItems", 0):
errors.append(f"{location}: too few items")
if "maxItems" in schema and len(instance) > schema["maxItems"]:
errors.append(f"{location}: too many items")
if schema.get("uniqueItems") and len({canonical_bytes(item) for item in instance}) != len(instance):
errors.append(f"{location}: duplicate items")
for index, value in enumerate(instance):
errors.extend(validate(value, schema.get("items", {}), schema_path, f"{location}[{index}]"))
elif isinstance(instance, str):
if len(instance) < schema.get("minLength", 0):
errors.append(f"{location}: string too short")
if "maxLength" in schema and len(instance) > schema["maxLength"]:
errors.append(f"{location}: string too long")
if "pattern" in schema and re.fullmatch(schema["pattern"], instance) is None:
errors.append(f"{location}: pattern mismatch")
if schema.get("format") == "date-time":
try:
datetime.fromisoformat(instance.replace("Z", "+00:00"))
except ValueError:
errors.append(f"{location}: invalid date-time")
elif isinstance(instance, (int, float)) and not isinstance(instance, bool):
if "minimum" in schema and instance < schema["minimum"]:
errors.append(f"{location}: below minimum")
if "maximum" in schema and instance > schema["maximum"]:
errors.append(f"{location}: above maximum")
return errors