50 lines
1.5 KiB
Python
50 lines
1.5 KiB
Python
"""Load the JSON Schema contracts and validate canon logs / origin seeds.
|
|
|
|
Schemas are the single source of truth in /docs/schemas. This module is the
|
|
api's half of the contract (charter §11): every canon log the client posts is
|
|
validated here before it could ever reach a prompt.
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
from functools import lru_cache
|
|
from pathlib import Path
|
|
|
|
from jsonschema import Draft202012Validator
|
|
|
|
|
|
def _schema_dir() -> Path:
|
|
"""Locate the schema directory.
|
|
|
|
Honours CANON_SCHEMA_DIR, else walks up from this file looking for
|
|
docs/schemas (local checkout) or schemas (bundled into the Docker image).
|
|
"""
|
|
env = os.environ.get("CANON_SCHEMA_DIR")
|
|
if env:
|
|
return Path(env)
|
|
here = Path(__file__).resolve()
|
|
for parent in here.parents:
|
|
for candidate in (parent / "docs" / "schemas", parent / "schemas"):
|
|
if candidate.is_dir():
|
|
return candidate
|
|
raise RuntimeError("schema directory not found")
|
|
|
|
|
|
@lru_cache(maxsize=None)
|
|
def _validator(schema_name: str) -> Draft202012Validator:
|
|
with open(_schema_dir() / schema_name) as f:
|
|
return Draft202012Validator(json.load(f))
|
|
|
|
|
|
def validation_errors(doc: dict, schema_name: str) -> list[str]:
|
|
validator = _validator(schema_name)
|
|
return [e.message for e in sorted(validator.iter_errors(doc), key=lambda e: list(e.path))]
|
|
|
|
|
|
def validate_origin(doc: dict) -> list[str]:
|
|
return validation_errors(doc, "origin.schema.json")
|
|
|
|
|
|
def validate_canon_log(doc: dict) -> list[str]:
|
|
return validation_errors(doc, "canon-log.schema.json")
|