diff --git a/Makefile b/Makefile index fcfce56..b4b2700 100644 --- a/Makefile +++ b/Makefile @@ -26,7 +26,7 @@ $(error PYTHON must be set for local package targets (examples: venv: PYTHON=.ve endif endif -.PHONY: help py-env install build install-local uninstall-local clean dev-fmt all check fmt lint clippy mypy test security update-spec e2e coverage +.PHONY: help py-env install build install-local uninstall-local clean dev-fmt all check fmt lint clippy mypy test security update-spec e2e coverage gts-server # Default target - show help .DEFAULT_GOAL := help @@ -115,6 +115,11 @@ coverage: install $(PYTHON) -m pip install 'pytest-cov>=5,<7' $(PYTHON) -m pytest tests/ --cov=gts --cov-report=xml --cov-report=term +PORT ?= 8000 + +gts-server: install + $(PYTHON) -m gts server --host 127.0.0.1 --port $(PORT) + # Run end-to-end tests against gts-spec e2e: install @echo "Starting server in background..." diff --git a/README.md b/README.md index 3af6258..a356620 100644 --- a/README.md +++ b/README.md @@ -4,7 +4,7 @@ A minimal, idiomatic Python library for working with **GTS** ([Global Type System](https://github.com/gts-spec/gts-spec)) identifiers and JSON/JSON Schema artifacts. -Current supported GTS spec version: `0.13.4` +Current supported GTS spec version: `0.14.0` ## Roadmap diff --git a/gts/README.md b/gts/README.md index 739fb10..43f66ff 100644 --- a/gts/README.md +++ b/gts/README.md @@ -2,7 +2,7 @@ Python helpers and a reference HTTP service for the [Global Type System (GTS)](https://github.com/globaltypesystem/gts-spec). The package supports GTS identifier parsing, JSON Schema-backed validation, schema compatibility and derivation checks, traits, casting, queries, file loading, a CLI, and a FastAPI application. -The package targets GTS specification v0.13.4 and requires Python 3.9 or later. +The package targets GTS specification v0.14.0 and requires Python 3.9 or later. ## Installation diff --git a/gts/openapi.json b/gts/openapi.json index 3e9b962..e4a2b05 100644 --- a/gts/openapi.json +++ b/gts/openapi.json @@ -2,7 +2,7 @@ "openapi": "3.1.0", "info": { "title": "GTS Server", - "version": "0.13.4" + "version": "0.14.0" }, "paths": { "/entities": { diff --git a/gts/pyproject.toml b/gts/pyproject.toml index 2405c35..2d5a04a 100644 --- a/gts/pyproject.toml +++ b/gts/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "gts" -version = "0.13.4" +version = "0.14.0" description = "Global Type System (GTS) helpers: identifiers, parsing, validation, and operations" readme = "README.md" authors = [{ name = "GTS Community" }] diff --git a/gts/src/gts/__init__.py b/gts/src/gts/__init__.py index 87f7530..3bc3e95 100644 --- a/gts/src/gts/__init__.py +++ b/gts/src/gts/__init__.py @@ -14,6 +14,7 @@ GtsIdSegment, GtsWildcard, ) +from .gts_ref_validation import GtsRefValidationMode from .path_resolver import GtsPathResolver from .store import ( GtsReader, @@ -30,6 +31,7 @@ "GtsIdSegment", "GtsPathResolver", "GtsReader", + "GtsRefValidationMode", "GtsStore", "GtsWildcard", "JsonEntity", diff --git a/gts/src/gts/_server.py b/gts/src/gts/_server.py index f5135ae..c02cd8f 100644 --- a/gts/src/gts/_server.py +++ b/gts/src/gts/_server.py @@ -1,9 +1,11 @@ from __future__ import annotations +# ruff: noqa: B008 + import logging import sys import time -from typing import Any +from typing import Annotated, Any from fastapi import Body, FastAPI, Query from fastapi.responses import JSONResponse @@ -11,8 +13,12 @@ from starlette.middleware.base import BaseHTTPMiddleware from .ops import GtsOps +from .gts_ref_validation import GtsRefValidationMode logger = logging.getLogger(__name__) +GTS_REF_VALIDATION_QUERY = Query( + GtsRefValidationMode.ANY_VALID, alias="gts-ref-validation" +) # ANSI color codes @@ -188,7 +194,7 @@ def __init__( self.host = host self.port = port self.base_url = f"http://{self.host}:{self.port}" - self.app = FastAPI(title="GTS Server", version="0.13.4") + self.app = FastAPI(title="GTS Server", version="0.14.0") self.app.add_middleware( _RequestLoggingMiddleware, verbose=self.ops.verbose, @@ -345,8 +351,14 @@ async def add_entity( self, body: dict[str, Any] = Body(...), # noqa: B008 - FastAPI dependency pattern validate: bool = Query(False), + validation: bool = Query(False), + gts_ref_validation: GtsRefValidationMode = GTS_REF_VALIDATION_QUERY, ) -> JSONResponse: - result = self.ops.add_entity(body, validate=validate) + result = self.ops.add_entity( + body, + validate=validate is True or validation is True, + gts_ref_validation=gts_ref_validation, + ) status_code = 200 if result.ok else 409 if result.conflict else 422 return JSONResponse(result.to_dict(), status_code=status_code) @@ -384,8 +396,12 @@ async def match_id_pattern( async def id_to_uuid(self, id: str = Query(..., alias="gts_id")) -> dict[str, Any]: return self.ops.uuid(id).to_dict() - async def validate_instance(self, body: ValidateInstanceRequest) -> dict[str, Any]: - return self.ops.validate_instance(body.instance_id).to_dict() + async def validate_instance( + self, + body: ValidateInstanceRequest, + gts_ref_validation: GtsRefValidationMode = GTS_REF_VALIDATION_QUERY, + ) -> dict[str, Any]: + return self.ops.validate_instance(body.instance_id, gts_ref_validation).to_dict() async def validate_json( self, @@ -401,12 +417,18 @@ async def validate_json_as_type( return self.ops.validate_json(body, explicit_type_id=gts_type).to_dict() async def validate_type_schema( - self, body: ValidateTypeSchemaRequest + self, + body: ValidateTypeSchemaRequest, + gts_ref_validation: GtsRefValidationMode = GTS_REF_VALIDATION_QUERY, ) -> dict[str, Any]: - return self.ops.validate_schema(body.type_id).to_dict() + return self.ops.validate_schema(body.type_id, gts_ref_validation).to_dict() - async def validate_entity(self, body: ValidateEntityRequest) -> dict[str, Any]: - return self.ops.validate_entity(body.resolved_id).to_dict() + async def validate_entity( + self, + body: ValidateEntityRequest, + gts_ref_validation: GtsRefValidationMode = GTS_REF_VALIDATION_QUERY, + ) -> dict[str, Any]: + return self.ops.validate_entity(body.resolved_id, gts_ref_validation).to_dict() async def schema_graph( self, id: str = Query(..., alias="gts_id") diff --git a/gts/src/gts/gts_ref_validation.py b/gts/src/gts/gts_ref_validation.py new file mode 100644 index 0000000..d3582b0 --- /dev/null +++ b/gts/src/gts/gts_ref_validation.py @@ -0,0 +1,7 @@ +from enum import Enum + + +class GtsRefValidationMode(str, Enum): + NONE = "none" + ANY_PRESENT = "any-present" + ANY_VALID = "any-valid" diff --git a/gts/src/gts/ops.py b/gts/src/gts/ops.py index 893f4cd..f9709b9 100644 --- a/gts/src/gts/ops.py +++ b/gts/src/gts/ops.py @@ -10,6 +10,7 @@ from .entities import DEFAULT_GTS_CONFIG, GtsConfig, GtsEntity from .files_reader import GtsFileReader from .gts import GtsID, GtsWildcard +from .gts_ref_validation import GtsRefValidationMode from .path_resolver import GtsPathResolver from .schema_cast import GtsEntityCastResult from .store import GtsStore, GtsStoreQueryResult @@ -17,6 +18,15 @@ # Interface helpers +def _normalize_gts_ref_validation(value: Any) -> GtsRefValidationMode: + if isinstance(value, GtsRefValidationMode): + return value + try: + return GtsRefValidationMode(value) + except (TypeError, ValueError): + return GtsRefValidationMode.ANY_VALID + + @dataclass class GtsIdValidationResult: """Result of validating a GTS ID format.""" @@ -384,8 +394,12 @@ def reload_from_path(self, path: str | builtins.list[str]) -> None: self.store = GtsStore(self._reader) def add_entity( - self, content: dict[str, Any], validate: bool = False + self, + content: dict[str, Any], + validate: bool = False, + gts_ref_validation: GtsRefValidationMode = GtsRefValidationMode.ANY_VALID, ) -> GtsAddEntityResult: + gts_ref_validation = _normalize_gts_ref_validation(gts_ref_validation) entity = GtsEntity(content=content, cfg=self.cfg) # For instances (non-schemas), require an id field from entity_id_fields @@ -429,9 +443,11 @@ def add_entity( if entity.is_schema: self.store.validate_schema_basic(entity.gts_id.id) if validate: - self.store.validate_schema(entity.gts_id.id) + self.store.validate_schema(entity.gts_id.id, gts_ref_validation) elif validate: - self.store.validate_instance(entity.raw_id or entity.gts_id.id) + self.store.validate_instance( + entity.raw_id or entity.gts_id.id, gts_ref_validation + ) except Exception as e: # noqa: BLE001 - converted to a result object at API boundary self.store.unregister(store_key) if previous: @@ -652,34 +668,52 @@ def validate_json( is_type_schema=entity.is_schema, ) - def validate_instance(self, gts_id: str) -> GtsValidationResult: + def validate_instance( + self, + gts_id: str, + gts_ref_validation: GtsRefValidationMode = GtsRefValidationMode.ANY_VALID, + ) -> GtsValidationResult: + gts_ref_validation = _normalize_gts_ref_validation(gts_ref_validation) try: - self.store.validate_instance(gts_id) + self.store.validate_instance(gts_id, gts_ref_validation) return GtsValidationResult(id=gts_id, ok=True) except Exception as e: # noqa: BLE001 - converted to a result object at API boundary return GtsValidationResult(id=gts_id, ok=False, error=str(e)) - def validate_schema(self, gts_id: str) -> GtsValidationResult: + def validate_schema( + self, + gts_id: str, + gts_ref_validation: GtsRefValidationMode = GtsRefValidationMode.ANY_VALID, + ) -> GtsValidationResult: + gts_ref_validation = _normalize_gts_ref_validation(gts_ref_validation) try: - self.store.validate_schema(gts_id) + self.store.validate_schema(gts_id, gts_ref_validation) return GtsValidationResult(id=gts_id, ok=True) except Exception as e: # noqa: BLE001 - converted to a result object at API boundary return GtsValidationResult(id=gts_id, ok=False, error=str(e)) - def validate_entity(self, gts_id: str) -> GtsEntityValidationResult: - try: - parsed = GtsID(gts_id) - except Exception as e: # noqa: BLE001 - converted to a result object at API boundary - return GtsEntityValidationResult( - id=gts_id, ok=False, entity_type="", error=str(e) - ) + def validate_entity( + self, + gts_id: str, + gts_ref_validation: GtsRefValidationMode = GtsRefValidationMode.ANY_VALID, + ) -> GtsEntityValidationResult: + gts_ref_validation = _normalize_gts_ref_validation(gts_ref_validation) + entity = self.store.get(gts_id) + if entity: + entity_type = "schema" if entity.is_schema else "instance" + else: + try: + parsed = GtsID(gts_id) + entity_type = "schema" if parsed.is_type else "instance" + except Exception as e: # noqa: BLE001 - converted at API boundary + return GtsEntityValidationResult( + id=gts_id, ok=False, entity_type="", error=str(e) + ) - if parsed.is_type: - entity_type = "schema" - result = self.validate_schema(gts_id) + if entity_type == "schema": + result = self.validate_schema(gts_id, gts_ref_validation) else: - entity_type = "instance" - result = self.validate_instance(gts_id) + result = self.validate_instance(gts_id, gts_ref_validation) return GtsEntityValidationResult( id=result.id, ok=result.ok, entity_type=entity_type, error=result.error diff --git a/gts/src/gts/schema_validation.py b/gts/src/gts/schema_validation.py index 95b61b7..9496e8a 100644 --- a/gts/src/gts/schema_validation.py +++ b/gts/src/gts/schema_validation.py @@ -1,6 +1,7 @@ from __future__ import annotations -from collections.abc import Iterator +import copy +from collections.abc import Callable, Iterator from typing import Any import regex @@ -9,6 +10,107 @@ PATTERN_TIMEOUT_SECONDS = 1.0 +_SCHEMA_MAP_KEYWORDS = { + "$defs", + "definitions", + "dependentSchemas", + "patternProperties", + "properties", +} +_SCHEMA_ARRAY_KEYWORDS = {"allOf", "anyOf", "oneOf", "prefixItems"} +_DRAFT3_SCHEMA_KEYWORDS = {"disallow", "extends", "type"} +_SCHEMA_SINGLE_KEYWORDS = { + "additionalItems", + "additionalProperties", + "contains", + "contentSchema", + "else", + "if", + "not", + "propertyNames", + "then", + "unevaluatedItems", + "unevaluatedProperties", + "x-gts-traits-schema", +} + + +def iter_schema_nodes( + schema: Any, path: str = "" +) -> Iterator[tuple[dict[str, Any], str]]: + if not isinstance(schema, dict): + return + yield schema, path + for keyword, value in schema.items(): + keyword_path = f"{path}/{keyword}" if path else keyword + if keyword in _SCHEMA_MAP_KEYWORDS and isinstance(value, dict): + for name, child in value.items(): + yield from iter_schema_nodes(child, f"{keyword_path}/{name}") + elif keyword in _SCHEMA_ARRAY_KEYWORDS and isinstance(value, list): + for index, child in enumerate(value): + yield from iter_schema_nodes(child, f"{keyword_path}[{index}]") + elif keyword in _SCHEMA_SINGLE_KEYWORDS: + yield from iter_schema_nodes(value, keyword_path) + elif keyword in _DRAFT3_SCHEMA_KEYWORDS: + if isinstance(value, dict): + yield from iter_schema_nodes(value, keyword_path) + elif isinstance(value, list): + for index, child in enumerate(value): + if isinstance(child, dict): + yield from iter_schema_nodes(child, f"{keyword_path}[{index}]") + elif keyword == "items": + if isinstance(value, list): + for index, child in enumerate(value): + yield from iter_schema_nodes(child, f"{keyword_path}[{index}]") + else: + yield from iter_schema_nodes(value, keyword_path) + elif keyword == "dependencies" and isinstance(value, dict): + for name, child in value.items(): + if isinstance(child, (dict, bool)): + yield from iter_schema_nodes(child, f"{keyword_path}/{name}") + + +def map_schema_nodes(schema: Any, transform: Callable[[Any], Any]) -> Any: + if not isinstance(schema, dict): + return copy.deepcopy(schema) + mapped = copy.deepcopy(schema) + for keyword, value in schema.items(): + if keyword in _SCHEMA_MAP_KEYWORDS and isinstance(value, dict): + mapped[keyword] = { + name: map_schema_nodes(child, transform) + for name, child in value.items() + } + elif keyword in _SCHEMA_ARRAY_KEYWORDS and isinstance(value, list): + mapped[keyword] = [map_schema_nodes(child, transform) for child in value] + elif keyword in _SCHEMA_SINGLE_KEYWORDS: + mapped[keyword] = map_schema_nodes(value, transform) + elif keyword in _DRAFT3_SCHEMA_KEYWORDS: + if isinstance(value, dict): + mapped[keyword] = map_schema_nodes(value, transform) + elif isinstance(value, list): + mapped[keyword] = [ + map_schema_nodes(child, transform) + if isinstance(child, dict) + else copy.deepcopy(child) + for child in value + ] + elif keyword == "items": + if isinstance(value, list): + mapped[keyword] = [ + map_schema_nodes(child, transform) for child in value + ] + else: + mapped[keyword] = map_schema_nodes(value, transform) + elif keyword == "dependencies" and isinstance(value, dict): + mapped[keyword] = { + name: map_schema_nodes(child, transform) + if isinstance(child, (dict, bool)) + else copy.deepcopy(child) + for name, child in value.items() + } + return transform(mapped) + + # Shared format checker for instance/trait validation. # # A bare ``FormatChecker()`` draws from jsonschema's shared, class-level checker diff --git a/gts/src/gts/store.py b/gts/src/gts/store.py index 1a489b4..f8c601e 100644 --- a/gts/src/gts/store.py +++ b/gts/src/gts/store.py @@ -14,8 +14,9 @@ from ._naming import looks_like_gts, strip_scheme, with_scheme from .entities import GtsEntity from .gts import GtsID, GtsRef, GtsWildcard +from .gts_ref_validation import GtsRefValidationMode from .schema_cast import GtsEntityCastResult -from .schema_validation import FORMAT_CHECKER, validator_for +from .schema_validation import FORMAT_CHECKER, iter_schema_nodes, validator_for from .x_gts_ref import XGtsRefValidator, _without_x_gts_ref logger = logging.getLogger(__name__) @@ -198,7 +199,7 @@ def get(self, entity_id: str) -> GtsEntity | None: def get_schema_content(self, type_id: str) -> dict[str, Any]: """Get schema content as dict (legacy method for backward compatibility).""" entity = self.get(type_id) - if entity and isinstance(entity.content, dict): + if entity and entity.is_schema and isinstance(entity.content, dict): return entity.content raise KeyError(f"Schema not found: {type_id}") @@ -295,6 +296,33 @@ def _validate_schema_refs(schema: dict[str, Any], path: str = "") -> None: nested_path = f"{path}[{idx}]" GtsStore._validate_schema_refs(item, nested_path) + def _validate_schema_ref_targets( + self, schema: Any, path: str = "", visited: set[str] | None = None + ) -> None: + visited = visited if visited is not None else set() + if isinstance(schema, dict): + ref_uri = schema.get("$ref") + if isinstance(ref_uri, str): + ref = GtsRef.parse(ref_uri) + if not ref.is_local and ref.is_gts and ref.has_scheme: + current_path = f"{path}.$ref" if path else "$ref" + try: + target = self.get_schema_content(ref.target_id) + except KeyError as error: + raise ValueError( + f"Unresolvable $ref at '{current_path}': '{ref_uri}'" + ) from error + if ref.target_id not in visited: + visited.add(ref.target_id) + self._validate_schema_ref_targets(target, current_path, visited) + for key, value in schema.items(): + if key != "$ref": + nested_path = f"{path}.{key}" if path else key + self._validate_schema_ref_targets(value, nested_path, visited) + elif isinstance(schema, list): + for index, item in enumerate(schema): + self._validate_schema_ref_targets(item, f"{path}[{index}]", visited) + def _validate_schema_x_gts_refs(self, gts_id: str) -> None: """ Validate a schema's x-gts-ref fields. @@ -340,15 +368,6 @@ def _validate_gts_keywords(content: dict[str, Any]) -> None: } supported_keywords = top_level_keywords | {"x-gts-ref"} - def _contains_key_recursive(value: Any, key: str) -> bool: - if isinstance(value, dict): - if key in value: - return True - return any(_contains_key_recursive(v, key) for v in value.values()) - elif isinstance(value, list): - return any(_contains_key_recursive(v, key) for v in value) - return False - # Validate x-gts-final final_val = content.get("x-gts-final") if final_val is not None and not isinstance(final_val, bool): @@ -369,26 +388,19 @@ def _contains_key_recursive(value: Any, key: str) -> bool: "schema cannot declare both x-gts-final and x-gts-abstract as true" ) - def _validate_extensions(value: Any) -> None: - if isinstance(value, dict): - for key, nested_value in value.items(): - if key.startswith("x-gts-") and key not in supported_keywords: - raise ValueError(f"Unsupported GTS extension keyword: {key}") - _validate_extensions(nested_value) - elif isinstance(value, list): - for item in value: - _validate_extensions(item) - - _validate_extensions(content) + for schema_node, _ in iter_schema_nodes(content): + for key in schema_node: + if key.startswith("x-gts-") and key not in supported_keywords: + raise ValueError(f"Unsupported GTS extension keyword: {key}") # Check that x-gts-final/x-gts-abstract/x-gts-traits/x-gts-traits-schema - # appear only at the top level - for key, value in content.items(): - if key in top_level_keywords: + # appear only at the top level. + for schema_node, path in iter_schema_nodes(content): + if not path: continue - for kw in top_level_keywords: - if _contains_key_recursive(value, kw): - raise ValueError(f"{kw} must be at the schema top level") + for keyword in top_level_keywords: + if keyword in schema_node: + raise ValueError(f"{keyword} must be at the schema top level") @staticmethod def _content_is_abstract(content: dict[str, Any]) -> bool: @@ -600,11 +612,15 @@ def _validate_traits( gts_id: str, is_abstract: bool, transient_schema: dict[str, Any] | None = None, + gts_ref_validation: GtsRefValidationMode = GtsRefValidationMode.ANY_VALID, ) -> None: """Validate OP#13: schema traits for a type.""" effective = self._build_effective_traits(gts_id, transient_schema) errors = effective.validate( - check_unresolved=not is_abstract, reference_store=self + check_unresolved=not is_abstract, + reference_store=self, + gts_ref_validation=gts_ref_validation, + selected_type_id=gts_id, ) if errors: raise ValueError( @@ -654,7 +670,10 @@ def validate_schema_basic(self, gts_id: str) -> None: self._validate_gts_keywords(schema_content) def validate_schema_content( - self, gts_id: str, schema_content: dict[str, Any] + self, + gts_id: str, + schema_content: dict[str, Any], + gts_ref_validation: GtsRefValidationMode = GtsRefValidationMode.ANY_VALID, ) -> None: """Validate a schema using the registry only for its dependencies.""" schema_id = _require_schema_id(gts_id) @@ -671,6 +690,7 @@ def validate_schema_content( logger.info(f"Validating schema {schema_id.id}") self._validate_schema_refs(schema_content, "") + self._validate_schema_ref_targets(schema_content) self._validate_schema_x_gts_refs_content(schema_id.id, schema_content) self._validate_gts_keywords(schema_content) self._validate_schema_chain(schema_id.id, schema_content) @@ -696,11 +716,28 @@ def validate_schema_content( schema_id.id, self._content_is_abstract(schema_content), schema_content, + gts_ref_validation, ) - def validate_schema(self, gts_id: str) -> None: + def validate_schema( + self, + gts_id: str, + gts_ref_validation: GtsRefValidationMode = GtsRefValidationMode.ANY_VALID, + ) -> None: """Validate a registered schema and all of its dependencies.""" + self._validate_schema_transitive(gts_id, set(), set(), gts_ref_validation) + + def _validate_schema_transitive( + self, + gts_id: str, + visiting: set[str], + validated: set[str], + gts_ref_validation: GtsRefValidationMode, + ) -> None: schema_id = _require_schema_id(gts_id) + key = f"schema:{schema_id.id}" + if key in validated or key in visiting: + return schema_entity = self.get(schema_id.id) if not schema_entity: @@ -711,9 +748,167 @@ def validate_schema(self, gts_id: str) -> None: raise ValueError( # noqa: TRY004 - keep ValueError for API compatibility f"Schema '{schema_id.id}' content must be a dictionary" ) - self.validate_schema_content(schema_id.id, schema_entity.content) - def validate_instance_content(self, content: dict[str, Any], type_id: str) -> None: + visiting.add(key) + try: + self.validate_schema_content( + schema_id.id, schema_entity.content, gts_ref_validation + ) + + schema_ref_validator = XGtsRefValidator(store=self, mode=gts_ref_validation) + schema_ref_errors = schema_ref_validator.validate_schema_ref_existence( + schema_entity.content, selected_type_id=schema_id.id + ) + if schema_ref_errors: + raise ValueError( + "x-gts-ref validation failed: " + + "; ".join(error.reason for error in schema_ref_errors) + ) + + effective_traits = self._build_effective_traits(schema_id.id) + trait_ref_validator = XGtsRefValidator(store=self, mode=gts_ref_validation) + trait_ref_validator.validate_schema_ref_existence( + effective_traits.schema, selected_type_id=schema_id.id + ) + trait_ref_validator.validate_instance( + effective_traits.values, + effective_traits.schema, + selected_type_id=schema_id.id, + ) + xref_ids = ( + schema_ref_validator.referenced_ids | trait_ref_validator.referenced_ids + ) + wildcard_patterns = ( + schema_ref_validator.referenced_wildcard_patterns + | trait_ref_validator.referenced_wildcard_patterns + ) + if gts_ref_validation is GtsRefValidationMode.ANY_VALID: + for dependency_id in xref_ids: + try: + self._validate_entity_transitive( + dependency_id, visiting, validated, gts_ref_validation + ) + except Exception as error: + raise ValueError( + f"Referenced x-gts-ref entity '{dependency_id}' is invalid: {error}" + ) from error + for pattern in wildcard_patterns: + if not self._has_valid_wildcard_match( + pattern, visiting, validated, gts_ref_validation + ): + raise ValueError( + f"x-gts-ref wildcard constraint '{pattern}' has no valid registered match" + ) + + chain_ids: list[str] = [] + prefix = "gts." + for segment in schema_id.gts_id_segments: + chain_ids.append(prefix + segment.segment) + prefix += segment.segment + for ancestor_id in chain_ids[:-1]: + try: + self._validate_schema_transitive( + ancestor_id, visiting, validated, gts_ref_validation + ) + except Exception as error: + raise ValueError( + f"Ancestor type '{ancestor_id}' is invalid: {error}" + ) from error + + for dependency_id, dependency_is_type in self._schema_dependencies( + schema_entity.content, include_gts_refs=False + ): + try: + if dependency_is_type: + self._validate_schema_transitive( + dependency_id, visiting, validated, gts_ref_validation + ) + else: + self._validate_entity_transitive( + dependency_id, visiting, validated, gts_ref_validation + ) + except Exception as error: + raise ValueError( + f"Referenced entity '{dependency_id}' is invalid: {error}" + ) from error + finally: + visiting.remove(key) + validated.add(key) + + def _schema_dependencies( + self, schema: Any, include_gts_refs: bool = True + ) -> Iterator[tuple[str, bool]]: + x_gts_ref_validator = XGtsRefValidator(mode=GtsRefValidationMode.NONE) + selected_type_id = ( + x_gts_ref_validator.selected_type_id(schema, None) + if isinstance(schema, dict) + else None + ) + for subschema, _path in iter_schema_nodes(schema): + ref_uri = subschema.get("$ref") + if isinstance(ref_uri, str): + ref = GtsRef.parse(ref_uri) + if not ref.is_local and ref.is_gts and ref.has_scheme: + yield ref.target_id, True + + if not include_gts_refs: + continue + x_gts_ref = x_gts_ref_validator.resolve_ref_pattern( + subschema.get("x-gts-ref"), selected_type_id + ) + if ( + isinstance(x_gts_ref, str) + and x_gts_ref.startswith("gts.") + and "*" not in x_gts_ref + ): + yield x_gts_ref, True + + def _validate_entity_transitive( + self, + gts_id: str, + visiting: set[str], + validated: set[str], + gts_ref_validation: GtsRefValidationMode, + ) -> None: + entity = self.get(gts_id) + if not entity: + raise StoreGtsEntityNotFound(gts_id) + if entity.is_schema: + self._validate_schema_transitive( + gts_id, visiting, validated, gts_ref_validation + ) + else: + self._validate_instance_transitive( + gts_id, visiting, validated, gts_ref_validation + ) + + def _has_valid_wildcard_match( + self, + pattern: str, + visiting: set[str], + validated: set[str], + gts_ref_validation: GtsRefValidationMode, + ) -> bool: + wildcard = GtsWildcard(pattern) + for entity_id in self._by_id: + try: + if not GtsID(entity_id).wildcard_match(wildcard): + continue + self._validate_entity_transitive( + entity_id, visiting, validated, gts_ref_validation + ) + return True + except Exception as error: # noqa: BLE001 - try another wildcard match + logger.debug("Invalid wildcard candidate %s: %s", entity_id, error) + continue + return False + + def validate_instance_content( + self, + content: dict[str, Any], + type_id: str, + gts_ref_validation: GtsRefValidationMode = GtsRefValidationMode.ANY_VALID, + ) -> set[str]: """Validate unregistered instance content against a registered type schema.""" schema_type = _require_schema_id(type_id) try: @@ -735,9 +930,9 @@ def validate_instance_content(self, content: dict[str, Any], type_id: str) -> No ) validator.validate(content) - x_gts_ref_validator = XGtsRefValidator(store=self) + x_gts_ref_validator = XGtsRefValidator(store=self, mode=gts_ref_validation) x_gts_ref_errors = x_gts_ref_validator.validate_instance( - content, self._resolve_schema_refs(schema) + content, self._resolve_schema_refs(schema), selected_type_id=schema_type.id ) if x_gts_ref_errors: error_messages = [ @@ -746,11 +941,65 @@ def validate_instance_content(self, content: dict[str, Any], type_id: str) -> No raise ValueError( f"x-gts-ref validation failed: {'; '.join(error_messages)}" ) + return x_gts_ref_validator.referenced_ids def validate_instance( self, gts_id: str, + gts_ref_validation: GtsRefValidationMode = GtsRefValidationMode.ANY_VALID, ) -> None: + """Validate an object instance and its complete dependency closure.""" + self._validate_instance_transitive(gts_id, set(), set(), gts_ref_validation) + + def _validate_instance_transitive( + self, + gts_id: str, + visiting: set[str], + validated: set[str], + gts_ref_validation: GtsRefValidationMode, + ) -> None: + key = f"instance:{gts_id}" + if key in validated or key in visiting: + return + visiting.add(key) + try: + referenced_ids = self._validate_instance_local(gts_id, gts_ref_validation) + + obj = ( + self.get(GtsID(gts_id).id) + if GtsID.is_valid(gts_id) + else self.get(gts_id) + ) + if not obj or not obj.type_id: + return + try: + self._validate_schema_transitive( + obj.type_id, visiting, validated, gts_ref_validation + ) + except Exception as error: + raise ValueError( + f"Instance type '{obj.type_id}' is invalid: {error}" + ) from error + + if gts_ref_validation is GtsRefValidationMode.ANY_VALID: + for dependency_id in referenced_ids: + try: + self._validate_entity_transitive( + dependency_id, visiting, validated, gts_ref_validation + ) + except Exception as error: + raise ValueError( + f"Referenced entity '{dependency_id}' is invalid: {error}" + ) from error + finally: + visiting.remove(key) + validated.add(key) + + def _validate_instance_local( + self, + gts_id: str, + gts_ref_validation: GtsRefValidationMode, + ) -> set[str]: """ Validate an object instance against its schema. @@ -781,7 +1030,9 @@ def validate_instance( raise TypeError(f"Instance '{lookup_id}' content must be a dictionary") logger.info(f"Validating instance {gts_id} against schema {obj.type_id}") - self.validate_instance_content(obj.content, obj.type_id) + return self.validate_instance_content( + obj.content, obj.type_id, gts_ref_validation + ) def cast( self, diff --git a/gts/src/gts/traits.py b/gts/src/gts/traits.py index 7d7f954..36c40c5 100644 --- a/gts/src/gts/traits.py +++ b/gts/src/gts/traits.py @@ -21,8 +21,9 @@ from . import derivation from ._json_pointer import resolve as resolve_json_pointer +from .gts_ref_validation import GtsRefValidationMode from .schema_validation import FORMAT_CHECKER as _FORMAT_CHECKER -from .schema_validation import validator_for +from .schema_validation import map_schema_nodes, validator_for from .x_gts_ref import XGtsRefValidator X_GTS_TRAITS_SCHEMA = "x-gts-traits-schema" @@ -53,7 +54,11 @@ def _has_explicit_values(self) -> bool: return isinstance(self.merged_traits, dict) and len(self.merged_traits) > 0 def validate( - self, check_unresolved: bool, reference_store: Any | None = None + self, + check_unresolved: bool, + reference_store: Any | None = None, + gts_ref_validation: GtsRefValidationMode = GtsRefValidationMode.ANY_VALID, + selected_type_id: str | None = None, ) -> list[str]: """Return a list of error strings (empty means valid).""" errors = _validate_trait_schema_integrity(self.resolved_trait_schemas) @@ -72,15 +77,23 @@ def validate( return [] if _effective_schema_is_false(self.schema): - if self._has_explicit_values(): + has_non_false_declaration = any( + declaration is not False for declaration in self.resolved_trait_schemas + ) + if self._has_explicit_values() or has_non_false_declaration: return [ f"{X_GTS_TRAITS_SCHEMA} resolves to `false` in the chain - " # noqa: ISC004 - f"{X_GTS_TRAITS} values are prohibited" + "trait declarations and values are prohibited" ] return [] return _validate_trait_values( - self.schema, self.values, check_unresolved, reference_store + self.schema, + self.values, + check_unresolved, + reference_store, + gts_ref_validation, + selected_type_id, ) @@ -274,6 +287,17 @@ def _materialize_traits(trait_schema: Any, traits: Any, depth: int = 0) -> Any: # --- validation ------------------------------------------------------------ +def _without_required(schema: Any) -> Any: + def strip(node: Any) -> Any: + if not isinstance(node, dict): + return node + result = dict(node) + result.pop("required", None) + return result + + return map_schema_nodes(copy.deepcopy(schema), strip) + + def _validate_trait_schema_integrity(resolved_trait_schemas: list[Any]) -> list[str]: for i, ts in enumerate(resolved_trait_schemas): if isinstance(ts, bool): @@ -324,28 +348,14 @@ def _validate_trait_schema_compatibility( return errors -def _strip_required(schema: Any, depth: int = 0) -> Any: - if depth >= MAX_RECURSION_DEPTH or not isinstance(schema, dict): - return schema - out = dict(schema) - out.pop("required", None) - all_of = out.get("allOf") - if isinstance(all_of, list): - out["allOf"] = [_strip_required(i, depth + 1) for i in all_of] - return out - - def _validate_traits_against_schema( trait_schema: Any, effective_traits: Any, check_unresolved: bool ) -> list[str]: errors: list[str] = [] - validation_schema = ( - trait_schema if check_unresolved else _strip_required(trait_schema) - ) try: - cls = validator_for(validation_schema) - validator = cls(validation_schema, format_checker=_FORMAT_CHECKER) + cls = validator_for(trait_schema) + validator = cls(trait_schema, format_checker=_FORMAT_CHECKER) for error in validator.iter_errors(effective_traits): errors.append(f"trait validation: {error.message}") except Exception as e: # noqa: BLE001 - surfaced as validation error message @@ -382,11 +392,26 @@ def _validate_trait_values( effective_traits: Any, check_unresolved: bool, reference_store: Any | None, + gts_ref_validation: GtsRefValidationMode, + selected_type_id: str | None, ) -> list[str]: + schema_for_values = ( + effective_traits_schema + if check_unresolved + else _without_required(effective_traits_schema) + ) errors = _validate_traits_against_schema( - effective_traits_schema, effective_traits, check_unresolved + schema_for_values, effective_traits, check_unresolved ) - xref = XGtsRefValidator(store=reference_store, require_registered_target=True) - for err in xref.validate_instance(effective_traits, effective_traits_schema, ""): + xref = XGtsRefValidator(store=reference_store, mode=gts_ref_validation) + for err in xref.validate_schema_ref_existence( + effective_traits_schema, selected_type_id=selected_type_id + ): + errors.append(f"trait x-gts-ref: {err.reason}") + for err in xref.validate_instance( + effective_traits, + effective_traits_schema, + selected_type_id=selected_type_id, + ): errors.append(f"trait x-gts-ref: {err.reason}") return errors diff --git a/gts/src/gts/x_gts_ref.py b/gts/src/gts/x_gts_ref.py index f410273..f0b6ce7 100644 --- a/gts/src/gts/x_gts_ref.py +++ b/gts/src/gts/x_gts_ref.py @@ -5,7 +5,7 @@ in the GTS specification section 9.5. Key optimizations: -1. Use jsonpointer library for JSON Pointer resolution +1. Resolve local JSON Schema $ref pointers during traversal 2. Consolidate duplicate validation logic 3. Simplify recursive traversal with a generic walker """ @@ -16,19 +16,20 @@ from jsonschema.validators import validator_for -from ._json_pointer import MISSING from ._json_pointer import resolve as resolve_json_pointer from ._naming import GTS_PREFIX, strip_scheme -from .gts import GtsID +from .gts import GtsID, GtsWildcard +from .gts_ref_validation import GtsRefValidationMode +from .schema_validation import iter_schema_nodes, map_schema_nodes + +X_GTS_REF_SELF = "/$id" def _without_x_gts_ref(schema: Any) -> Any: - if isinstance(schema, dict): - stripped = { - key: _without_x_gts_ref(value) - for key, value in schema.items() - if key != "x-gts-ref" - } + def strip(node: Any) -> Any: + if not isinstance(node, dict): + return node + stripped = {key: value for key, value in node.items() if key != "x-gts-ref"} for keyword in ("oneOf", "anyOf", "allOf"): branches = stripped.get(keyword) if ( @@ -38,9 +39,8 @@ def _without_x_gts_ref(schema: Any) -> Any: ): stripped.pop(keyword, None) return stripped - if isinstance(schema, list): - return [_without_x_gts_ref(value) for value in schema] - return schema + + return map_schema_nodes(schema, strip) def _is_x_gts_ref_only_combinator(branches: list[Any]) -> bool: @@ -78,19 +78,42 @@ class XGtsRefValidator: """Validator for x-gts-ref constraints in GTS schemas.""" def __init__( - self, store: Any | None = None, require_registered_target: bool = False + self, + store: Any | None = None, + mode: GtsRefValidationMode | bool | str = GtsRefValidationMode.ANY_VALID, + *, + enforce_existence: bool | None = None, ): - """ - Initialize validator. - - Args: - store: Optional GtsStore for resolving entity references - """ + if enforce_existence is not None: + mode = ( + GtsRefValidationMode.ANY_PRESENT + if enforce_existence + else GtsRefValidationMode.NONE + ) + elif isinstance(mode, bool): + mode = GtsRefValidationMode.ANY_PRESENT if mode else GtsRefValidationMode.NONE self.store = store - self.require_registered_target = require_registered_target + self.mode = GtsRefValidationMode(mode) + self.referenced_ids: set[str] = set() + self.referenced_wildcard_patterns: set[str] = set() + + @staticmethod + def is_self_reference(value: Any) -> bool: + return value == X_GTS_REF_SELF + + @staticmethod + def selected_type_id( + schema: dict[str, Any], selected_type_id: str | None + ) -> str | None: + candidate = selected_type_id or schema.get("$id") + return strip_scheme(candidate) if isinstance(candidate, str) else None def validate_instance( - self, instance: dict[str, Any], schema: dict[str, Any], instance_path: str = "" + self, + instance: dict[str, Any], + schema: dict[str, Any], + instance_path: str = "", + selected_type_id: str | None = None, ) -> list[XGtsRefValidationError]: """ Validate an instance against x-gts-ref constraints in schema. @@ -104,6 +127,7 @@ def validate_instance( List of validation errors (empty if valid) """ errors: list[XGtsRefValidationError] = [] + selected_type_id = self.selected_type_id(schema, selected_type_id) def resolve_local_ref(ref: str) -> Any | None: if ref != "#" and not ref.startswith("#/"): @@ -123,7 +147,9 @@ def visit_instance(inst, sch, path, errs, refs=None): visit_instance(inst, target, path, errs, refs | {ref}) if "x-gts-ref" in sch and isinstance(inst, str): - error = self._validate_ref_value(inst, sch["x-gts-ref"], path, schema) + error = self._validate_ref_value( + inst, sch["x-gts-ref"], path, selected_type_id + ) if error: errs.append(error) @@ -202,10 +228,30 @@ def visit_instance(inst, sch, path, errs, refs=None): prop_path = f"{path}.{prop_name}" if path else prop_name visit_instance(inst[prop_name], prop_schema, prop_path, errs) - if "items" in sch and isinstance(inst, list): - for idx, item in enumerate(inst): - item_path = f"{path}[{idx}]" - visit_instance(item, sch["items"], item_path, errs) + if isinstance(inst, list): + tuple_items = sch.get("prefixItems") + items = sch.get("items") + if isinstance(tuple_items, list): + for idx, item_schema in enumerate(tuple_items[: len(inst)]): + item_path = f"{path}[{idx}]" + visit_instance(inst[idx], item_schema, item_path, errs) + if isinstance(items, dict): + for idx in range(len(tuple_items), len(inst)): + item_path = f"{path}[{idx}]" + visit_instance(inst[idx], items, item_path, errs) + elif isinstance(items, list): + for idx, item_schema in enumerate(items[: len(inst)]): + item_path = f"{path}[{idx}]" + visit_instance(inst[idx], item_schema, item_path, errs) + additional_items = sch.get("additionalItems") + if isinstance(additional_items, dict): + for idx in range(len(items), len(inst)): + item_path = f"{path}[{idx}]" + visit_instance(inst[idx], additional_items, item_path, errs) + elif isinstance(items, dict): + for idx, item in enumerate(inst): + item_path = f"{path}[{idx}]" + visit_instance(item, items, item_path, errs) def _validate_branch(inst, branch, path): branch_errors: list[XGtsRefValidationError] = [] @@ -216,57 +262,88 @@ def _validate_branch(inst, branch, path): return errors def validate_schema( - self, - schema: dict[str, Any], - schema_path: str = "", - root_schema: dict[str, Any] | None = None, + self, schema: dict[str, Any], schema_path: str = "" ) -> list[XGtsRefValidationError]: - """ - Validate x-gts-ref fields in a schema definition. - - Args: - schema: The JSON schema to validate - schema_path: Current path in schema (for error reporting) - root_schema: The root schema (for resolving relative refs) - - Returns: - List of validation errors (empty if valid) - """ - if root_schema is None: - root_schema = schema - + """Validate x-gts-ref fields in a schema definition.""" errors = [] + for subschema, path in iter_schema_nodes(schema, schema_path): + if "x-gts-ref" not in subschema: + continue + ref_path = f"{path}/x-gts-ref" if path else "x-gts-ref" + error = self._validate_ref_pattern(subschema["x-gts-ref"], ref_path) + if error: + errors.append(error) + return errors - def visit_schema(sch, path): - """Recursively visit schema nodes.""" - if not isinstance(sch, dict): - return - - # Check for x-gts-ref field - if "x-gts-ref" in sch: - ref_value = sch["x-gts-ref"] - ref_path = f"{path}/x-gts-ref" if path else "x-gts-ref" - error = self._validate_ref_pattern(ref_value, ref_path, root_schema) - if error: - errors.append(error) - - # Recurse into nested structures - for key, value in sch.items(): - if key == "x-gts-ref": + def validate_schema_ref_existence( + self, + schema: Any, + schema_path: str = "", + selected_type_id: str | None = None, + ) -> list[XGtsRefValidationError]: + if self.store is None or self.mode == GtsRefValidationMode.NONE: + return [] + store = self.store + selected_type_id = self.selected_type_id(schema, selected_type_id) + + def matches(pattern: str) -> list[str]: + wildcard = GtsWildcard(pattern) + result = [] + for entity_id, _ in store.items(): # noqa: PERF102 - generic store protocol + try: + if GtsID(entity_id).wildcard_match(wildcard): + result.append(entity_id) + except ValueError: continue - nested_path = f"{path}/{key}" if path else key - if isinstance(value, dict): - visit_schema(value, nested_path) - elif isinstance(value, list): - for idx, item in enumerate(value): - if isinstance(item, dict): - visit_schema(item, f"{nested_path}[{idx}]") - - visit_schema(schema, schema_path) + return result + + errors: list[XGtsRefValidationError] = [] + for subschema, path in iter_schema_nodes(schema, schema_path): + ref_pattern = subschema.get("x-gts-ref") + resolved = self.resolve_ref_pattern(ref_pattern, selected_type_id) + if not isinstance(resolved, str) or not resolved.startswith(GTS_PREFIX): + continue + ref_path = f"{path}/x-gts-ref" if path else "x-gts-ref" + if "*" in resolved: + if matches(resolved): + self.referenced_wildcard_patterns.add(resolved) + else: + errors.append( + XGtsRefValidationError( + ref_path, + ref_pattern, + resolved, + f"x-gts-ref wildcard constraint '{resolved}' has no registered match", + ) + ) + elif store.get(resolved) is None: + errors.append( + XGtsRefValidationError( + ref_path, + ref_pattern, + resolved, + f"x-gts-ref constraint type '{resolved}' is not registered", + ) + ) + else: + self.referenced_ids.add(resolved) return errors + def resolve_ref_pattern( + self, ref_pattern: Any, selected_type_id: str | None + ) -> str | None: + if not isinstance(ref_pattern, str): + return None + if self.is_self_reference(ref_pattern): + return selected_type_id + return strip_scheme(ref_pattern) + def _validate_ref_value( - self, value: str, ref_pattern: str, field_path: str, schema: dict[str, Any] + self, + value: str, + ref_pattern: str, + field_path: str, + selected_type_id: str | None, ) -> XGtsRefValidationError | None: """ Validate an instance value against its x-gts-ref constraint. @@ -275,7 +352,7 @@ def _validate_ref_value( value: The field value to validate ref_pattern: The x-gts-ref pattern field_path: Path to the field (for error reporting) - schema: The complete schema (for resolving relative refs) + selected_type_id: Canonical identifier of the selected leaf type Returns: XGtsRefValidationError if validation fails, None otherwise @@ -288,32 +365,21 @@ def _validate_ref_value( f"Value must be a string, got {type(value).__name__}", ) - # Resolve pattern if it's a relative reference - if ref_pattern.startswith("/"): - resolved_pattern = self._resolve_pointer(schema, ref_pattern) - if resolved_pattern is None: + if self.is_self_reference(ref_pattern): + if selected_type_id is None: return XGtsRefValidationError( field_path, value, ref_pattern, - f"Cannot resolve reference path '{ref_pattern}'", + "Cannot resolve /$id without a selected GTS Type Schema", ) - if not isinstance(resolved_pattern, str) or not resolved_pattern.startswith( - "gts." - ): - return XGtsRefValidationError( - field_path, - value, - ref_pattern, - f"Resolved reference '{ref_pattern}' -> '{resolved_pattern}' is not a GTS pattern", - ) - ref_pattern = resolved_pattern + ref_pattern = selected_type_id # Validate against GTS pattern return self._validate_gts_pattern(value, ref_pattern, field_path) def _validate_ref_pattern( - self, ref_pattern: str, field_path: str, root_schema: dict[str, Any] + self, ref_pattern: str, field_path: str ) -> XGtsRefValidationError | None: """ Validate an x-gts-ref pattern in a schema definition. @@ -321,7 +387,6 @@ def _validate_ref_pattern( Args: ref_pattern: The x-gts-ref value field_path: Path to the field (for error reporting) - root_schema: The root schema (for resolving relative refs) Returns: XGtsRefValidationError if validation fails, None otherwise @@ -338,30 +403,14 @@ def _validate_ref_pattern( if ref_pattern.startswith(GTS_PREFIX): return self._validate_gts_id_or_pattern(ref_pattern, field_path) - # Case 2: Relative reference - if ref_pattern.startswith("/"): - resolved = self._resolve_pointer(root_schema, ref_pattern) - if resolved is None: - return XGtsRefValidationError( - field_path, - ref_pattern, - ref_pattern, - f"Cannot resolve reference path '{ref_pattern}'", - ) - if not isinstance(resolved, str) or not GtsID.is_valid(resolved): - return XGtsRefValidationError( - field_path, - ref_pattern, - ref_pattern, - f"Resolved reference '{ref_pattern}' -> '{resolved}' is not a valid GTS identifier", - ) + if self.is_self_reference(ref_pattern): return None return XGtsRefValidationError( field_path, ref_pattern, ref_pattern, - f"Invalid x-gts-ref value: '{ref_pattern}' must start with 'gts.' or '/'", + f"Invalid x-gts-ref value: '{ref_pattern}' must be a GTS identifier, wildcard, or '{X_GTS_REF_SELF}'", ) def _validate_gts_id_or_pattern( @@ -433,10 +482,8 @@ def _validate_gts_pattern( f"Value '{value}' does not match pattern '{pattern}'", ) - # Optionally check if entity exists in store - if self.store and ( - not self.require_registered_target or self.store.get(pattern) - ): + # Referenced values use exact registry lookup in presence/full modes. + if self.store and self.mode != GtsRefValidationMode.NONE: entity = self.store.get(value) if not entity: return XGtsRefValidationError( @@ -445,34 +492,6 @@ def _validate_gts_pattern( pattern, f"Referenced entity '{value}' not found in registry", ) - - return None - - def _resolve_pointer(self, schema: dict[str, Any], pointer: str) -> str | None: - """ - Resolve a JSON Pointer in the schema to a GTS identifier. - - Args: - schema: The schema to search - pointer: JSON Pointer (e.g., "/$id", "/properties/type") - - Returns: - The resolved GTS identifier (bare form) or None if not found. - """ - current = resolve_json_pointer(schema, pointer, default=MISSING) - if current is MISSING or current is None: - return None - - # If current is a string, return it (normalized to the bare form). - if isinstance(current, str): - return strip_scheme(current) - - # If current is a dict with x-gts-ref, resolve it - if isinstance(current, dict) and "x-gts-ref" in current: - ref_value = current["x-gts-ref"] - if isinstance(ref_value, str): - if ref_value.startswith("/"): - return self._resolve_pointer(schema, ref_value) - return strip_scheme(ref_value) + self.referenced_ids.add(value) return None diff --git a/tests/test_ops.py b/tests/test_ops.py index 0db6446..77e42fb 100644 --- a/tests/test_ops.py +++ b/tests/test_ops.py @@ -1,10 +1,8 @@ """Tests for gts.ops.GtsOps (the high-level CLI/HTTP operations facade).""" import pytest - from gts.ops import GtsOps - SCHEMA = { "$schema": "http://json-schema.org/draft-07/schema#", "$id": "gts://gts.x.test._.foo.v1~", @@ -144,6 +142,16 @@ def test_add_entities_batch(self, ops): assert result.ok is True assert len(result.results) == 2 + def test_add_entities_rejects_unsupported_x_gts_ref_pointer(self, ops): + schema = { + "$schema": "http://json-schema.org/draft-07/schema#", + "$id": "gts://gts.x.test._.relative.v1~", + "properties": {"ref": {"x-gts-ref": "/missing"}}, + } + result = ops.add_entities([schema]) + assert result.ok is False + assert "must be a GTS identifier" in result.results[0].error + class TestAddSchemaLegacy: def test_add_schema_legacy_success(self, ops): diff --git a/tests/test_server.py b/tests/test_server.py index 3d99c3f..d6a6856 100644 --- a/tests/test_server.py +++ b/tests/test_server.py @@ -75,6 +75,15 @@ def test_add_entity_failure(self, server): resp = run(server.add_entity(body={"no": "id"}, validate=False)) assert resp.status_code == 422 + def test_add_entity_validation_alias(self, server): + schema = { + **SCHEMA, + "$id": "gts://gts.x.test._.relative.v1~", + "properties": {"ref": {"x-gts-ref": "/missing"}}, + } + resp = run(server.add_entity(body=schema, validate=False, validation=True)) + assert resp.status_code == 422 + def test_add_changed_entity_conflict(self, server): assert run(server.add_entity(body=SCHEMA, validate=False)).status_code == 200 changed_schema = {**SCHEMA, "properties": {"name": {"type": "integer"}}} diff --git a/tests/test_store_extra.py b/tests/test_store_extra.py index ac90fad..1850bcc 100644 --- a/tests/test_store_extra.py +++ b/tests/test_store_extra.py @@ -1,17 +1,18 @@ """Additional coverage-focused tests for gts.store.GtsStore.""" -import pytest -from typing import Iterator, Optional +from collections.abc import Iterator +from typing import Optional +import pytest +from gts.entities import DEFAULT_GTS_CONFIG, GtsEntity +from gts.gts import GtsID +from gts.schema_validation import PATTERN_TIMEOUT_SECONDS, validator_for from gts.store import ( - GtsStore, GtsReader, + GtsStore, StoreGtsEntityNotFound, StoreGtsObjectNotFound, ) -from gts.entities import GtsEntity, DEFAULT_GTS_CONFIG -from gts.gts import GtsID -from gts.schema_validation import PATTERN_TIMEOUT_SECONDS, validator_for class MockGtsReader(GtsReader): @@ -108,6 +109,77 @@ def test_recurses_into_list(self): {"allOf": [{"$ref": "http://example.com/schema"}]} ) + def test_registered_gts_ref_target_resolves(self): + target = _schema_entity("gts.x.test._.target.v1~") + store = GtsStore(MockGtsReader([target])) + store._validate_schema_ref_targets( + {"allOf": [{"$ref": "gts://gts.x.test._.target.v1~"}]} + ) + + def test_non_schema_gts_ref_target_raises(self): + target_id = "gts.x.test._.target.v1~" + target = GtsEntity( + content={"$id": target_id}, gts_id=GtsID(target_id), is_schema=False + ) + store = GtsStore(MockGtsReader([target])) + with pytest.raises(ValueError, match="Unresolvable \\$ref"): + store._validate_schema_ref_targets({"$ref": f"gts://{target_id}"}) + + def test_transitive_missing_gts_ref_target_raises(self): + target = _schema_entity( + "gts.x.test._.target.v1~", + {"$ref": "gts://gts.x.test._.missing.v1~"}, + ) + store = GtsStore(MockGtsReader([target])) + with pytest.raises(ValueError, match="Unresolvable \\$ref"): + store._validate_schema_ref_targets( + {"$ref": "gts://gts.x.test._.target.v1~"} + ) + + def test_missing_gts_ref_target_raises(self): + store = GtsStore(reader=None) + with pytest.raises(ValueError, match="Unresolvable \\$ref"): + store._validate_schema_ref_targets( + {"allOf": [{"$ref": "gts://gts.x.test._.missing.v1~"}]} + ) + + def test_missing_derived_gts_ref_target_raises(self): + target = _schema_entity("gts.x.test._.target.v1~") + store = GtsStore(MockGtsReader([target])) + with pytest.raises(ValueError, match="Unresolvable \\$ref"): + store._validate_schema_ref_targets( + {"$ref": ("gts://gts.x.test._.target.v1~x.test._.missing.v1~")} + ) + + +class TestSchemaDependencies: + def test_ignores_x_gts_ref_in_annotation_data(self): + store = GtsStore(reader=None) + assert ( + list( + store._schema_dependencies( + {"const": {"x-gts-ref": "gts.x.test._.missing.v1~"}} + ) + ) + == [] + ) + + def test_finds_constraint_under_property_named_x_gts_ref(self): + store = GtsStore(reader=None) + assert list( + store._schema_dependencies( + {"properties": {"x-gts-ref": {"x-gts-ref": "gts.x.test._.missing.v1~"}}} + ) + ) == [("gts.x.test._.missing.v1~", True)] + + def test_unsupported_x_gts_ref_pointer_is_not_a_dependency(self): + store = GtsStore(reader=None) + schema = { + "target": "gts.x.test._.target.v1~", + "properties": {"ref": {"x-gts-ref": "/target"}}, + } + assert list(store._schema_dependencies(schema)) == [] + class TestValidateGtsKeywords: def test_final_must_be_bool(self): @@ -134,6 +206,11 @@ def test_valid_top_level_keywords_pass(self): GtsStore._validate_gts_keywords({"x-gts-final": True}) GtsStore._validate_gts_keywords({"x-gts-abstract": True}) + def test_extension_shaped_annotation_data_is_ignored(self): + GtsStore._validate_gts_keywords( + {"const": {"x-gts-final": True, "x-gts-unknown": True}} + ) + def test_content_is_abstract_and_final(self): assert GtsStore._content_is_abstract({"x-gts-abstract": True}) is True assert GtsStore._content_is_abstract({}) is False @@ -170,6 +247,22 @@ def test_invalid_x_gts_ref_raises(self): with pytest.raises(Exception, match="x-gts-ref validation failed"): store._validate_schema_x_gts_refs("gts.x.test._.foo.v1~") + def test_basic_validation_rejects_unsupported_pointer(self): + schema = _schema_entity( + "gts.x.test._.foo.v1~", + { + "x-gts-traits-schema": { + "properties": { + "ref": {"x-gts-ref": "/x-gts-traits-schema/missingTarget"} + } + } + }, + ) + store = GtsStore(reader=None) + store.register(schema) + with pytest.raises(Exception, match="must be a GTS identifier"): + store.validate_schema_basic("gts.x.test._.foo.v1~") + class TestValidateSchemaChain: def test_single_segment_no_parent_ok(self): diff --git a/tests/test_traits.py b/tests/test_traits.py index c75bffe..3b8a6da 100644 --- a/tests/test_traits.py +++ b/tests/test_traits.py @@ -1,6 +1,7 @@ """Tests for gts.traits (OP#13 schema traits validation).""" from gts._json_pointer import resolve +from gts.schema_validation import map_schema_nodes from gts.traits import ( build_effective_traits, build_effective_traits_schema, @@ -11,6 +12,26 @@ ) +class TestSchemaTraversal: + def test_maps_draft3_schema_forms_only(self): + schema = { + "extends": {"required": ["extended"]}, + "type": ["object", {"required": ["typed"]}], + "disallow": ["array", {"required": ["disallowed"]}], + "const": {"required": ["data"]}, + } + mapped = map_schema_nodes( + schema, + lambda node: {k: v for k, v in node.items() if k != "required"}, + ) + assert mapped == { + "extends": {}, + "type": ["object", {}], + "disallow": ["array", {}], + "const": {"required": ["data"]}, + } + + class TestCollection: def test_collect_trait_schema_from_value_direct(self): out = [] @@ -221,6 +242,59 @@ def test_abstract_skips_unresolved_check(self): errors = effective.validate(check_unresolved=False) assert errors == [] + def test_abstract_validates_provided_trait_values(self): + schema = { + "type": "object", + "properties": {"a": {"type": "string"}}, + } + effective = build_effective_traits([schema], {"a": 5}, None) + errors = effective.validate(check_unresolved=False) + assert any("is not of type 'string'" in error for error in errors) + + def test_abstract_preserves_required_in_const_value(self): + schema = { + "type": "object", + "properties": {"config": {"const": {"required": ["a"]}}}, + "required": ["missing"], + } + effective = build_effective_traits( + [schema], {"config": {"required": ["a"]}}, None + ) + assert effective.validate(check_unresolved=False) == [] + + def test_abstract_skips_required_in_nested_dialect_schema(self): + schema = { + "type": "object", + "allOf": [ + { + "$schema": "http://json-schema.org/draft-07/schema#", + "type": "object", + "required": ["missing"], + } + ], + } + effective = build_effective_traits([schema], {}, None) + assert effective.validate(check_unresolved=False) == [] + + def test_abstract_checks_x_gts_ref_constraint_type_existence(self): + class FakeStore: + def get(self, value): + return None + + schema = { + "type": "object", + "properties": { + "ref": {"type": "string", "x-gts-ref": "gts.x.test._.foo.v1~"} + }, + } + errors = build_effective_traits([schema], {}, None).validate( + check_unresolved=False, reference_store=FakeStore() + ) + assert any( + "constraint type 'gts.x.test._.foo.v1~' is not registered" in e + for e in errors + ) + def test_incompatible_trait_schema_chain_flagged(self): # Second schema narrows type incompatibly with the ancestor. effective = build_effective_traits( @@ -238,7 +312,10 @@ def test_dialect_applied_to_effective_schema(self): effective = build_effective_traits( [{"type": "object"}], {}, "https://json-schema.org/draft/2020-12/schema" ) - assert effective.schema["$schema"] == "https://json-schema.org/draft/2020-12/schema" + assert ( + effective.schema["$schema"] + == "https://json-schema.org/draft/2020-12/schema" + ) def test_x_gts_ref_errors_prefixed(self): schema = { diff --git a/tests/test_x_gts_ref.py b/tests/test_x_gts_ref.py index ceb78e2..e2ed54d 100644 --- a/tests/test_x_gts_ref.py +++ b/tests/test_x_gts_ref.py @@ -1,6 +1,6 @@ """Tests for gts.x_gts_ref (x-gts-ref schema & instance validation, spec sec 9.5).""" -from gts.x_gts_ref import XGtsRefValidator +from gts.x_gts_ref import X_GTS_REF_SELF, XGtsRefValidator class TestValidateSchema: @@ -36,53 +36,133 @@ def test_non_string_ref_value(self): def test_invalid_prefix_value(self): errors = XGtsRefValidator().validate_schema({"x-gts-ref": "nope"}) assert len(errors) == 1 - assert "must start with" in errors[0].reason + assert "must be a GTS identifier" in errors[0].reason - def test_relative_pointer_resolves_to_valid_id(self): + def test_selected_type_self_reference_is_valid(self): + validator = XGtsRefValidator() + assert validator.is_self_reference(X_GTS_REF_SELF) + assert validator.validate_schema({"x-gts-ref": X_GTS_REF_SELF}) == [] + + def test_other_pointers_are_invalid(self): + validator = XGtsRefValidator() + for pointer in ("/missing/path", "/properties/id"): + errors = validator.validate_schema({"x-gts-ref": pointer}) + assert len(errors) == 1 + assert "must be a GTS identifier" in errors[0].reason + + def test_recurses_into_nested_structures(self): schema = { - "$id": "gts.x.test._.foo.v1~", "properties": { - "ref_field": {"x-gts-ref": "/$id"}, - }, + "child": {"x-gts-ref": "notgts.*"}, + } } errors = XGtsRefValidator().validate_schema(schema) - assert errors == [] + assert len(errors) == 1 + assert "properties/child/x-gts-ref" in errors[0].field_path - def test_relative_pointer_unresolvable(self): - schema = {"properties": {"ref_field": {"x-gts-ref": "/missing/path"}}} + def test_recurses_into_list_of_dicts(self): + schema = {"allOf": [{"x-gts-ref": "notgts.*"}]} errors = XGtsRefValidator().validate_schema(schema) assert len(errors) == 1 - assert "Cannot resolve reference path" in errors[0].reason + assert "allOf[0]/x-gts-ref" in errors[0].field_path - def test_relative_pointer_resolves_to_invalid_id(self): + def test_ignores_x_gts_ref_in_annotation_data(self): schema = { - "not_gts": "definitely not a gts id !!", - "properties": {"ref_field": {"x-gts-ref": "/not_gts"}}, + "properties": { + "payload": { + "const": {"x-gts-ref": "gts.x.test._.missing.v1~"}, + "default": {"x-gts-ref": "not-a-gts-id"}, + } + } } - errors = XGtsRefValidator().validate_schema(schema) - assert len(errors) == 1 - assert "is not a valid GTS identifier" in errors[0].reason + assert XGtsRefValidator().validate_schema(schema) == [] - def test_recurses_into_nested_structures(self): + def test_property_named_x_gts_ref_is_not_a_keyword(self): schema = { "properties": { - "child": {"x-gts-ref": "notgts.*"}, + "x-gts-ref": {"x-gts-ref": "notgts.*"}, } } errors = XGtsRefValidator().validate_schema(schema) assert len(errors) == 1 - assert "properties/child/x-gts-ref" in errors[0].field_path + assert errors[0].field_path == "properties/x-gts-ref/x-gts-ref" - def test_recurses_into_list_of_dicts(self): - schema = {"allOf": [{"x-gts-ref": "notgts.*"}]} + def test_recurses_into_draft3_schema_forms(self): + schema = { + "$schema": "http://json-schema.org/draft-03/schema#", + "extends": {"x-gts-ref": "invalid-extends"}, + "type": ["object", {"x-gts-ref": "invalid-type"}], + "disallow": ["array", {"x-gts-ref": "invalid-disallow"}], + } errors = XGtsRefValidator().validate_schema(schema) + assert [error.field_path for error in errors] == [ + "extends/x-gts-ref", + "type[1]/x-gts-ref", + "disallow[1]/x-gts-ref", + ] + + +class TestValidateSchemaRefExistence: + def test_missing_concrete_constraint_type_fails(self): + class FakeStore: + def get(self, value): + return None + + errors = XGtsRefValidator(store=FakeStore()).validate_schema_ref_existence( + {"properties": {"ref": {"x-gts-ref": "gts.x.test._.foo.v1~"}}} + ) + assert len(errors) == 1 - assert "allOf[0]/x-gts-ref" in errors[0].field_path + assert errors[0].field_path == "properties/ref/x-gts-ref" + assert ( + "constraint type 'gts.x.test._.foo.v1~' is not registered" + in errors[0].reason + ) + + def test_registered_and_wildcard_constraints_pass(self): + class FakeStore: + def get(self, value): + return object() if value == "gts.x.test._.foo.v1~" else None + + def items(self): + return [("gts.x.test._.foo.v1~", object())] + + errors = XGtsRefValidator(store=FakeStore()).validate_schema_ref_existence( + { + "allOf": [ + {"x-gts-ref": "gts.x.test._.foo.v1~"}, + {"x-gts-ref": "gts.x.test.*"}, + ] + } + ) + + assert errors == [] + + def test_annotation_data_does_not_require_constraint_type(self): + class FakeStore: + def get(self, value): + return None + + errors = XGtsRefValidator(store=FakeStore()).validate_schema_ref_existence( + {"const": {"x-gts-ref": "gts.x.test._.missing.v1~"}} + ) + assert errors == [] + + def test_selected_type_constraint_uses_explicit_leaf(self): + class FakeStore: + def get(self, value): + return object() if value == "gts.x.test._.leaf.v1~" else None + + errors = XGtsRefValidator(store=FakeStore()).validate_schema_ref_existence( + {"x-gts-ref": X_GTS_REF_SELF}, + selected_type_id="gts.x.test._.leaf.v1~", + ) + assert errors == [] class TestValidateInstanceValue: def test_non_string_instance_value_error(self): - error = XGtsRefValidator()._validate_ref_value(123, "gts.*", "ref", {}) + error = XGtsRefValidator()._validate_ref_value(123, "gts.*", "ref", None) assert error is not None assert "Value must be a string" in error.reason @@ -97,17 +177,24 @@ def test_relative_ref_pattern_resolution_on_instance(self): ) assert errors == [] - def test_relative_ref_pattern_resolution_fails_when_not_gts_prefix(self): + def test_self_reference_uses_explicit_selected_leaf(self): schema = { - "other": "not-gts-value", + "$id": "gts.x.test._.base.v1~", "type": "object", - "properties": {"ref": {"x-gts-ref": "/other"}}, + "properties": {"ref": {"x-gts-ref": X_GTS_REF_SELF}}, } + leaf = "gts.x.test._.base.v1~x.test._.leaf.v1~" + assert ( + XGtsRefValidator().validate_instance( + {"ref": leaf}, schema, selected_type_id=leaf + ) + == [] + ) errors = XGtsRefValidator().validate_instance( - {"ref": "gts.x.test._.foo.v1~"}, schema + {"ref": "gts.x.test._.base.v1~"}, schema, selected_type_id=leaf ) assert len(errors) == 1 - assert "is not a GTS pattern" in errors[0].reason + assert "does not match pattern" in errors[0].reason def test_wildcard_pattern_matches_prefix(self): errors = XGtsRefValidator().validate_instance( @@ -155,10 +242,38 @@ def test_array_items_recursion(self): "type": "array", "items": {"x-gts-ref": "gts.x.test.*"}, } + errors = XGtsRefValidator().validate_instance(["gts.x.other.v1~"], schema) + assert len(errors) == 1 + + def test_tuple_additional_items_recursion(self): + schema = { + "type": "array", + "items": [{"type": "string"}], + "additionalItems": { + "type": "string", + "x-gts-ref": "gts.x.test._.target.v1~", + }, + } + errors = XGtsRefValidator().validate_instance( + ["tuple-prefix", "gts.x.other._.target.v1~"], schema + ) + assert len(errors) == 1 + assert errors[0].field_path == "[1]" + + def test_prefix_items_recursion(self): + schema = { + "type": "array", + "prefixItems": [{"type": "string"}], + "items": { + "type": "string", + "x-gts-ref": "gts.x.test._.target.v1~", + }, + } errors = XGtsRefValidator().validate_instance( - ["gts.x.other.v1~"], schema + ["tuple-prefix", "gts.x.other._.target.v1~"], schema ) assert len(errors) == 1 + assert errors[0].field_path == "[1]" def test_object_properties_recursion(self): schema = {