41 lines
1.8 KiB
Python
41 lines
1.8 KiB
Python
"""Проверка payload фактов качества по контракту `contracts/metrics/quality-facts.schema.json` (JSON Schema 2020-12).
|
|
|
|
Контракт живёт в docs, копия — `api/contracts/metrics/` (идентичность проверяет `test_contract.py`). Корневая схема
|
|
не инстанцируется: `quality_intake_fact.payload` валидируется против `#/$defs/intake-payload`, `quality_outcome_fact.
|
|
payload` — против `#/$defs/outcome-payload`.
|
|
"""
|
|
|
|
import json
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from jsonschema import Draft202012Validator, FormatChecker
|
|
|
|
SCHEMA_PATH = Path(__file__).resolve().parents[1] / "contracts" / "metrics" / "quality-facts.schema.json"
|
|
_SCHEMA = json.loads(SCHEMA_PATH.read_text(encoding="utf-8"))
|
|
|
|
|
|
def _validator(def_name: str) -> Draft202012Validator:
|
|
schema = {"$schema": _SCHEMA["$schema"], "$defs": _SCHEMA["$defs"], "$ref": f"#/$defs/{def_name}"}
|
|
return Draft202012Validator(schema, format_checker=FormatChecker())
|
|
|
|
|
|
_INTAKE = _validator("intake-payload")
|
|
_OUTCOME = _validator("outcome-payload")
|
|
|
|
|
|
def _errors(validator: Draft202012Validator, payload: dict[str, Any]) -> list[str]:
|
|
return [f"{'/'.join(map(str, e.absolute_path))}: {e.message}" for e in validator.iter_errors(payload)]
|
|
|
|
|
|
def assert_intake_payload_valid(payload: dict[str, Any]) -> None:
|
|
"""Payload факта приёма соответствует `$defs/intake-payload`."""
|
|
problems = _errors(_INTAKE, payload)
|
|
assert not problems, problems[:5]
|
|
|
|
|
|
def assert_outcome_payload_valid(payload: dict[str, Any]) -> None:
|
|
"""Payload факта исхода соответствует `$defs/outcome-payload`."""
|
|
problems = _errors(_OUTCOME, payload)
|
|
assert not problems, problems[:5]
|