48 lines
2 KiB
Python
48 lines
2 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from typing import Any
|
|
|
|
from jsonschema import Draft202012Validator, FormatChecker
|
|
|
|
from hub_core.contracts import extension_contract_root
|
|
from hub_core.runtime.models import RegistryRegistration
|
|
|
|
|
|
class ContractValidator:
|
|
"""Validate runtime registration input against the packaged contract."""
|
|
|
|
def __init__(self) -> None:
|
|
contract_root = extension_contract_root()
|
|
schema_root = contract_root.joinpath("schemas")
|
|
self._descriptor = _validator(schema_root.joinpath("hub-descriptor.schema.json"))
|
|
self._manifest = _validator(schema_root.joinpath("hub-manifest.schema.json"))
|
|
catalog = json.loads(
|
|
contract_root.joinpath("catalogs", "event-types.json").read_text(encoding="utf-8")
|
|
)
|
|
self._event_families = {
|
|
entry["type"]: entry["family"] for entry in catalog["event_types"]
|
|
}
|
|
|
|
def validate_registration(self, registration: RegistryRegistration) -> None:
|
|
self._descriptor.validate(registration.descriptor)
|
|
self._manifest.validate(registration.manifest)
|
|
descriptor_id = registration.descriptor.get("reuse_surface_id")
|
|
manifest_id = registration.manifest.get("reuse_surface_id")
|
|
if descriptor_id != manifest_id:
|
|
raise ValueError("descriptor and manifest reuse_surface_id must match")
|
|
|
|
def validate_event_family(self, event_type: str, expected_family: str) -> None:
|
|
actual_family = self._event_families.get(event_type)
|
|
if actual_family is None:
|
|
raise ValueError(f"event type '{event_type}' is not cataloged")
|
|
if actual_family != expected_family:
|
|
raise ValueError(
|
|
f"event type '{event_type}' belongs to '{actual_family}', not '{expected_family}'"
|
|
)
|
|
|
|
|
|
def _validator(resource: Any) -> Draft202012Validator:
|
|
schema = json.loads(resource.read_text(encoding="utf-8"))
|
|
Draft202012Validator.check_schema(schema)
|
|
return Draft202012Validator(schema, format_checker=FormatChecker())
|