diff --git a/packages/tangle-cli/src/tangle_cli/schema_validation.py b/packages/tangle-cli/src/tangle_cli/schema_validation.py index c2081e8..842bb40 100644 --- a/packages/tangle-cli/src/tangle_cli/schema_validation.py +++ b/packages/tangle-cli/src/tangle_cli/schema_validation.py @@ -104,20 +104,27 @@ def load_dehydrated_schema() -> dict[str, Any]: ) from e +@lru_cache(maxsize=1) +def _dehydrated_validator() -> Any: + """Build (and ``check_schema``) the packaged-schema validator once.""" + schema = load_dehydrated_schema() + validator_cls = validator_for(schema) + validator_cls.check_schema(schema) + return validator_cls(schema) + + def validate_dehydrated_data(data: Mapping[str, Any]) -> None: """Validate ``data`` against the dehydrated pipeline schema. Uses the draft declared in the schema's ``$schema`` (Draft 2020-12) - via :func:`jsonschema.validators.validator_for`. + via :func:`jsonschema.validators.validator_for`. The validator is built + once; ``data`` is fully validated on every call. Raises: SchemaValidationError: with the single best/most-specific message when ``data`` does not conform. """ - schema = load_dehydrated_schema() - validator_cls = validator_for(schema) - validator_cls.check_schema(schema) - validator = validator_cls(schema) + validator = _dehydrated_validator() errors = sorted(validator.iter_errors(data), key=lambda e: list(e.absolute_path)) if not errors: diff --git a/tests/test_schema_validation.py b/tests/test_schema_validation.py index d81939a..ecfc5dc 100644 --- a/tests/test_schema_validation.py +++ b/tests/test_schema_validation.py @@ -9,8 +9,12 @@ """ import copy +import jsonschema import pytest +from jsonschema import Draft202012Validator +from jsonschema.validators import validator_for +from tangle_cli import schema_validation from tangle_cli.schema_validation import ( SchemaValidationError, assert_no_template_delimiters, @@ -177,6 +181,80 @@ def test_validate_dehydrated_data_rejects_unsupported_execution_options( validate_dehydrated_data(data) +# --------------------------------------------------------------------------- +# Compiled-validator caching: check_schema runs once, every payload +# still fully validated with unchanged error semantics. + + +@pytest.fixture +def check_schema_calls(monkeypatch): + """Cold validator cache; record every check_schema call.""" + calls: list[object] = [] + original = Draft202012Validator.check_schema + + def counting(cls, s, *args, **kwargs): + calls.append(s) + return original(s, *args, **kwargs) + + monkeypatch.setattr(Draft202012Validator, "check_schema", classmethod(counting)) + schema_validation._dehydrated_validator.cache_clear() + yield calls + schema_validation._dehydrated_validator.cache_clear() + + +def _uncached_error(data) -> str | None: + """Reference: the pre-cache path (build a validator every call).""" + schema = load_dehydrated_schema() + cls = validator_for(schema) + errors = sorted(cls(schema).iter_errors(data), key=lambda e: list(e.absolute_path)) + if not errors: + return None + best = jsonschema.exceptions.best_match(errors) + loc = ".".join(str(p) for p in best.absolute_path) or "root" + return f"dehydrated pipeline failed schema validation at {loc}: {best.message}" + + +def _invalid_payloads() -> list[dict]: + missing_name = _valid_pipeline() + del missing_name["name"] + no_tasks = _valid_pipeline() + no_tasks["implementation"]["graph"]["tasks"] = {} + inline_spec = _valid_pipeline() + inline_spec["implementation"]["graph"]["tasks"]["extract"]["componentRef"] = { + "spec": {"name": "Extract"} + } + bad_retry = _valid_pipeline() + bad_retry["implementation"]["graph"]["tasks"]["load"]["executionOptions"] = { + "retryStrategy": {"maxRetries": -1} + } + return [missing_name, no_tasks, inline_spec, bad_retry] + + +def test_check_schema_runs_once_across_many_validations(check_schema_calls): + for _ in range(25): + validate_dehydrated_data(_valid_pipeline()) + for payload in _invalid_payloads(): + with pytest.raises(SchemaValidationError): + validate_dehydrated_data(payload) + assert len(check_schema_calls) == 1 + + +def test_cached_validator_keeps_error_messages(check_schema_calls): + # Interleave valid/invalid payloads on a warm cache; each result must + # match the uncached reference exactly (message + location). + for _ in range(3): + validate_dehydrated_data(_valid_pipeline()) + for payload in _invalid_payloads(): + expected = _uncached_error(payload) + assert expected is not None + with pytest.raises(SchemaValidationError) as excinfo: + validate_dehydrated_data(payload) + assert str(excinfo.value) == expected + assert excinfo.value.location == expected.split(" at ", 1)[1].split(":", 1)[0] + validate_dehydrated_data(_valid_pipeline()) + assert len(check_schema_calls) == 1 + + # --------------------------------------------------------------------------- # Template-delimiter output contract.