diff --git a/.gts-spec b/.gts-spec index 2a171df..5b5d578 160000 --- a/.gts-spec +++ b/.gts-spec @@ -1 +1 @@ -Subproject commit 2a171dff7810657de147e3b5f06f89537d7a4293 +Subproject commit 5b5d5786bab1de1192d977a94752e492b66d17a6 diff --git a/Makefile b/Makefile index dc588e2..fcfce56 100644 --- a/Makefile +++ b/Makefile @@ -43,13 +43,15 @@ py-env: $(PY_ENV_STAMP) $(PY_ENV_PYTHON): $(PYTHON_BOOTSTRAP) -m venv --clear $(PY_ENV_DIR) -$(PY_ENV_STAMP): $(PY_ENV_PYTHON) gts/pyproject.toml .gts-spec/tests/requirements.txt Makefile +$(PY_ENV_STAMP): $(PY_ENV_PYTHON) gts/pyproject.toml .gts-spec/tests/requirements.txt requirements.txt Makefile @echo "Creating/updating Python virtual environment in $(PY_ENV_DIR)..." $(PYTHON_BOOTSTRAP) -m venv $(PY_ENV_DIR) $(PYTHON) -m pip install --upgrade pip + # Spec test-client deps, then httprunner (--no-deps: its own pins are + # incompatible with this venv), then local dev tooling + version overrides. $(PYTHON) -m pip install -r .gts-spec/tests/requirements.txt $(PYTHON) -m pip install --no-deps 'httprunner>=4,<5' - $(PYTHON) -m pip install ruff mypy + $(PYTHON) -m pip install -r requirements.txt @touch $@ # Install gts package into the venv (editable, for development) diff --git a/gts/pyproject.toml b/gts/pyproject.toml index bc31a91..da05785 100644 --- a/gts/pyproject.toml +++ b/gts/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "gts" -version = "0.13.1" +version = "0.13.2" description = "Global Type System (GTS) helpers: identifiers, parsing, validation, and operations" readme = "README.md" authors = [{ name = "GTS Community" }] @@ -15,8 +15,12 @@ dependencies = [ "referencing>=0.30,<0.37", "jsonsubschema>=0.0.8,<0.1", "fastapi>=0.110,<1", + # Server models use pydantic v2 APIs (model_validator); declare it directly + # rather than relying on FastAPI to pull a compatible version transitively. + "pydantic>=2,<3", "uvicorn>=0.23,<1", "pyyaml>=6.0,<7", + "regex==2025.11.3", "eval_type_backport>=0.1,<0.3; python_version < '3.10'" ] diff --git a/gts/src/gts/_json_pointer.py b/gts/src/gts/_json_pointer.py new file mode 100644 index 0000000..15aa810 --- /dev/null +++ b/gts/src/gts/_json_pointer.py @@ -0,0 +1,71 @@ +"""RFC 6901 JSON Pointer resolution. + +A JSON Pointer (RFC 6901) addresses a single value inside a JSON document, e.g. +``/properties/type``. Because ``/`` separates reference tokens and ``~`` begins +an escape sequence, those two characters are escaped *inside* a token: + +- ``~1`` denotes a literal ``/`` +- ``~0`` denotes a literal ``~`` + +Unescaping MUST replace ``~1`` before ``~0``; otherwise an encoded ``~01`` would +be corrupted. This module is the single home for that logic, which was +previously duplicated (with the same ``~1``/``~0`` magic) across ``traits.py`` +and ``x_gts_ref.py``. +""" + +from __future__ import annotations + +from typing import Any +from urllib.parse import unquote + +# Sentinel distinguishing "pointer resolved to a real ``None``" from +# "pointer could not be resolved". Callers that care should pass this (or their +# own default) and compare identity against the returned value. +MISSING: Any = object() + + +def unescape_token(token: str) -> str: + """Decode a single RFC 6901 reference token (``~1`` -> ``/``, ``~0`` -> ``~``).""" + return token.replace("~1", "/").replace("~0", "~") + + +def resolve(document: Any, pointer: str, default: Any = None) -> Any: + """Resolve an RFC 6901 JSON Pointer against ``document``. + + ``pointer`` accepts three equivalent spellings: + + - the empty string ``""`` - the whole document; + - a pointer beginning with ``/`` - ``/a/b``; + - a same-document URI fragment - ``#`` or ``#/a/b``. + + Returns ``default`` if any reference token cannot be resolved (missing key, + non-integer/out-of-range array index, or descending into a scalar). + """ + if pointer.startswith("#"): + pointer = unquote(pointer[1:]) + if pointer == "": + return document + if not pointer.startswith("/"): + return default + + current = document + for raw_token in pointer.split("/")[1:]: + token = unescape_token(raw_token) + if isinstance(current, dict): + if token not in current: + return default + current = current[token] + elif isinstance(current, list): + if not ( + token.isascii() + and token.isdecimal() + and (token == "0" or not token.startswith("0")) + ): + return default + try: + current = current[int(token)] + except IndexError: + return default + else: + return default + return current diff --git a/gts/src/gts/_json_validation.py b/gts/src/gts/_json_validation.py index 46104b2..c7e23df 100644 --- a/gts/src/gts/_json_validation.py +++ b/gts/src/gts/_json_validation.py @@ -7,9 +7,10 @@ from pathlib import Path from typing import Any +from ._naming import GTS_PREFIX, GTS_URI_PREFIX, looks_like_gts from .entities import GtsEntity, GtsFile from .files_reader import DEFAULT_EXCLUDE_LIST -from .gts import GTS_PREFIX, GTS_URI_PREFIX, GtsID +from .gts import GtsID from .store import GtsStore _X_GTS_REF_KEYWORD = "x-gts-ref" @@ -232,8 +233,7 @@ def _is_gts_related(self, value: Any) -> bool: @staticmethod def _looks_gts(v: str) -> bool: - normalized = v.removeprefix(GTS_URI_PREFIX) - return normalized.startswith(GTS_PREFIX) or v.startswith(GTS_URI_PREFIX) + return looks_like_gts(v) @staticmethod def _registry_key(entity: GtsEntity) -> str | None: diff --git a/gts/src/gts/_naming.py b/gts/src/gts/_naming.py new file mode 100644 index 0000000..5be4145 --- /dev/null +++ b/gts/src/gts/_naming.py @@ -0,0 +1,62 @@ +"""Internal naming primitives for GTS identifiers. + +This module is the single, **internal** home for the low-level string handling +of GTS identifiers: + +- the ``gts://`` URI scheme, +- the bare ``gts.`` prefix, and +- the ``~`` type marker. + +It is deliberately private (underscore-prefixed and absent from the public +package exports). Consumers of the SDK should never reach for these primitives: +they work with the :class:`~gts.gts.GtsID` value object and high-level +operations (validate, cast, resolve, store lookups, ...), all of which normalize +identifiers internally. Keeping this logic in one private place stops it from +leaking across the library and onto the public API surface. +""" + +from __future__ import annotations + +# Distinguishes GTS identifiers from other strings; also used for URI encoding. +GTS_PREFIX = "gts." +GTS_URI_PREFIX = "gts://" +# Separates the segments of a chained identifier and, at the very end, marks a +# type identifier (e.g. ``gts.acme.pkg._.user.v1~``). +GTS_TYPE_MARKER = "~" + + +def strip_scheme(value: str) -> str: + """Return the canonical bare form, dropping any ``gts://`` scheme. + + Non-GTS strings are returned unchanged, so this is safe to call on arbitrary + registry keys or ``$ref`` targets at a boundary. + """ + return value.removeprefix(GTS_URI_PREFIX) + + +def has_scheme(value: str) -> bool: + """True if ``value`` carries the ``gts://`` URI scheme.""" + return value.startswith(GTS_URI_PREFIX) + + +def with_scheme(value: str) -> str: + """Return the ``gts://`` URI encoding of ``value`` (idempotent).""" + return value if has_scheme(value) else GTS_URI_PREFIX + value + + +def looks_like_gts(value: str) -> bool: + """True if ``value`` looks like a GTS identifier in either encoding. + + A cheap prefix check (bare ``gts.`` or ``gts://``); it does not fully + validate the identifier. + """ + return value.startswith((GTS_URI_PREFIX, GTS_PREFIX)) + + +def is_type_ref(value: str) -> bool: + """True if the (scheme-stripped) identifier denotes a type. + + Type identifiers end with the type marker ``~``; instance identifiers do + not. + """ + return strip_scheme(value).endswith(GTS_TYPE_MARKER) diff --git a/gts/src/gts/compatibility.py b/gts/src/gts/compatibility.py index ce5421e..76a1270 100644 --- a/gts/src/gts/compatibility.py +++ b/gts/src/gts/compatibility.py @@ -132,6 +132,37 @@ def sanitize(schema: Any) -> Any: return schema +def _lower_root_unevaluated_properties(schema: Any) -> Any | None: + if not isinstance(schema, dict) or "unevaluatedProperties" not in schema: + return schema + if any( + key in schema + for key in ( + "$ref", + "$dynamicRef", + "allOf", + "anyOf", + "oneOf", + "if", + "then", + "else", + "not", + "dependentSchemas", + ) + ): + return None + unevaluated = schema["unevaluatedProperties"] + if ( + "additionalProperties" in schema + and schema["additionalProperties"] != unevaluated + ): + return None + result = dict(schema) + result.pop("unevaluatedProperties") + result.setdefault("additionalProperties", unevaluated) + return result + + def _coerce_bool_schema(schema: Any) -> Any: """Turn a top-level boolean schema into its object-equivalent. @@ -147,14 +178,18 @@ def _coerce_bool_schema(schema: Any) -> Any: def _is_subschema(subset: Any, superset: Any) -> bool | None: """``Valid(subset) subset-of Valid(superset)`` or ``None`` when unprovable.""" - finite_result = _finite_subset(subset, superset) + lowered_subset = _lower_root_unevaluated_properties(subset) + lowered_superset = _lower_root_unevaluated_properties(superset) + if lowered_subset is None or lowered_superset is None: + return None + finite_result = _finite_subset(lowered_subset, lowered_superset) if finite_result is not None: return finite_result try: return bool( isSubschema( - _coerce_bool_schema(sanitize(subset)), - _coerce_bool_schema(sanitize(superset)), + _coerce_bool_schema(sanitize(lowered_subset)), + _coerce_bool_schema(sanitize(lowered_superset)), ) ) except Exception: # noqa: BLE001 - intentional broad fallback @@ -172,7 +207,7 @@ def _canonical_dialect(declared: str) -> str: return body.removeprefix("https://").removeprefix("http://") -def _dialect_changed(old_schema: Any, new_schema: Any) -> bool: +def dialects_differ(old_schema: Any, new_schema: Any) -> bool: if not isinstance(old_schema, dict) or not isinstance(new_schema, dict): return False old_dialect = old_schema.get("$schema") @@ -186,14 +221,14 @@ def _dialect_changed(old_schema: Any, new_schema: Any) -> bool: def check_backward_compatibility(old_schema: Any, new_schema: Any) -> str: """new consumers read old data: ``Valid(old) subset-of Valid(new)``.""" - if _dialect_changed(old_schema, new_schema): + if dialects_differ(old_schema, new_schema): return UNKNOWN return _verdict(_is_subschema(old_schema, new_schema)) def check_forward_compatibility(old_schema: Any, new_schema: Any) -> str: """old consumers read new data: ``Valid(new) subset-of Valid(old)``.""" - if _dialect_changed(old_schema, new_schema): + if dialects_differ(old_schema, new_schema): return UNKNOWN return _verdict(_is_subschema(new_schema, old_schema)) diff --git a/gts/src/gts/entities.py b/gts/src/gts/entities.py index 171e68e..94b61c3 100644 --- a/gts/src/gts/entities.py +++ b/gts/src/gts/entities.py @@ -3,7 +3,8 @@ from dataclasses import dataclass, field from typing import TYPE_CHECKING, Any -from .gts import GTS_PREFIX, GTS_URI_PREFIX, GtsID +from ._naming import GTS_PREFIX, has_scheme, strip_scheme +from .gts import GtsID from .schema_cast import GtsEntityCastResult, SchemaCastError if TYPE_CHECKING: @@ -258,8 +259,7 @@ def _extract_gts_ids_with_paths(self) -> list[dict[str, str]]: def gts_id_matcher(node: Any, path: str) -> dict[str, str] | None: """Match GTS ID strings.""" if isinstance(node, str): - val = node - val = val.removeprefix("gts://") + val = strip_scheme(node) if GtsID.is_valid(val): return {"id": val, "sourcePath": path or "root"} return None @@ -274,9 +274,7 @@ def _extract_ref_strings_with_paths(self) -> list[dict[str, str]]: def ref_matcher(node: Any, path: str) -> dict[str, str] | None: """Match $ref properties in dict nodes.""" if isinstance(node, dict) and isinstance(node.get("$ref"), str): - val = node["$ref"] - # Issue #32: handle gts:// prefix - val = val.removeprefix("gts://") + val = strip_scheme(node["$ref"]) ref_path = f"{path}.$ref" if path else "$ref" return {"id": val, "sourcePath": ref_path} return None @@ -290,9 +288,8 @@ def _get_field_value(self, field: str) -> str | None: return None v = self.content.get(field) if isinstance(v, str) and v.strip(): - # Issue #31, #32: Handle gts:// prefix in fields (e.g. $id) - v = v.removeprefix("gts://") - return v + # Normalize the ``gts://`` scheme at this document boundary. + return strip_scheme(v) return None def _schema_id_uses_plain_prefix(self) -> bool: @@ -308,7 +305,7 @@ def _schema_id_uses_plain_prefix(self) -> bool: if not isinstance(raw, str): return False raw = raw.strip() - return raw.startswith(GTS_PREFIX) and not raw.startswith(GTS_URI_PREFIX) + return raw.startswith(GTS_PREFIX) and not has_scheme(raw) def _first_non_empty_field(self, fields: list[str]) -> tuple[str, str] | None: """Find first non-empty field value in order. @@ -348,12 +345,10 @@ def _calc_json_schema_id(self, cfg: GtsConfig) -> str | None: # type_id is the parent (everything up to the second-to-last '~'). # idv ends with '~' for schemas. # Strip trailing '~' to find internal chain boundaries. - inner = idv.removesuffix("~") - last_tilde = inner.rfind("~") - if last_tilde > 0: - # Has at least 2 segments - return parent chain + parent_type_id = GtsID(idv).parent_type_id + if parent_type_id: self.selected_type_id_field = "$id" - return inner[: last_tilde + 1] + return parent_type_id # Base schema (single segment) - no GTS parent type. # The $schema URL is NOT a GTS Type Identifier. return None @@ -368,17 +363,13 @@ def _calc_json_schema_id(self, cfg: GtsConfig) -> str | None: if entity_id_cand[0] == "$id" and not self.is_schema: pass # Skip to PRIORITY 2 else: - idv = entity_id_cand[1] # If already a type id (ends with '~'), use it as-is - if idv.endswith("~"): - self.selected_type_id_field = entity_id_cand[0] - return idv # For chained IDs (well-known instances), extract schema: # everything up to and including last '~' - last_tilde = idv.rfind("~") - if last_tilde > 0: + type_id = GtsID(entity_id_cand[1]).type_id + if type_id: self.selected_type_id_field = entity_id_cand[0] - return idv[: last_tilde + 1] + return type_id # PRIORITY 2: Fall back to explicit schema_id_fields (type, gtsTid, etc.) # Only check these if no chained GTS ID was found in entity_id_fields @@ -389,10 +380,7 @@ def _calc_json_schema_id(self, cfg: GtsConfig) -> str | None: type_id_val = cand[1] # If type_id is a chained GTS ID, extract parent (base type) if GtsID.is_valid(type_id_val): - last_tilde = type_id_val.rfind("~") - if last_tilde > 0 and not type_id_val.endswith("~"): - # It's an instance ID in type field - extract schema part - return type_id_val[: last_tilde + 1] + return GtsID(type_id_val).type_id return type_id_val # No schema reference found for instance diff --git a/gts/src/gts/gts.py b/gts/src/gts/gts.py index eeef6cf..6a21b00 100644 --- a/gts/src/gts/gts.py +++ b/gts/src/gts/gts.py @@ -5,8 +5,23 @@ import uuid from typing import Any -GTS_PREFIX = "gts." -GTS_URI_PREFIX = "gts://" +from ._naming import ( + GTS_PREFIX, + GTS_TYPE_MARKER, +) +from ._naming import ( + has_scheme as _has_scheme, +) +from ._naming import ( + is_type_ref as _is_type_ref, +) +from ._naming import ( + strip_scheme as _strip_scheme, +) +from ._naming import ( + with_scheme as _with_scheme, +) + GTS_NS = uuid.uuid5(uuid.NAMESPACE_URL, "gts") GTS_SEGMENT_TOKEN_REGEX = re.compile(r"^[a-z_][a-z0-9_]*$") UUID_REGEX = re.compile( @@ -74,10 +89,10 @@ def __init__(self, num: int, offset: int, segment: str): self._parse_segment_id(num, offset, segment) def _parse_segment_id(self, num: int, offset: int, segment: str): - if segment.count("~") > 0: - if segment.count("~") > 1: + if segment.count(GTS_TYPE_MARKER) > 0: + if segment.count(GTS_TYPE_MARKER) > 1: raise GtsInvalidSegment(num, offset, segment, "Too many '~' characters") - if segment.endswith("~"): + if segment.endswith(GTS_TYPE_MARKER): self.is_type = True segment = segment[:-1] else: @@ -191,8 +206,8 @@ class GtsID: def __init__(self, id: str): raw = id.strip() - # Strip gts:// URI prefix if present - raw = raw.removeprefix(GTS_URI_PREFIX) + # Normalize to the canonical bare form at this boundary. + raw = _strip_scheme(raw) # Validate it's lower case if raw != raw.lower(): @@ -268,7 +283,7 @@ def __init__(self, id: str): s for s in self.gts_id_segments if not getattr(s, "_is_uuid_tail", False) ] if ( - not self.id.endswith("~") + not self.id.endswith(GTS_TYPE_MARKER) and self.uuid_tail is None and len(non_uuid_segments) == 1 and not any(seg.is_wildcard for seg in self.gts_id_segments) @@ -280,9 +295,32 @@ def __init__(self, id: str): "Instance IDs must be chained (e.g., type~instance).", ) + @property + def uri(self) -> str: + """This identifier rendered in ``gts://`` URI form.""" + return _with_scheme(self.id) + @property def is_type(self) -> bool: - return self.id.endswith("~") + return self.gts_id_segments[-1].is_type + + @property + def is_instance(self) -> bool: + return not self.is_type + + @property + def type_id(self) -> str | None: + return self.id if self.is_type else self.get_type_id() + + @property + def parent_type_id(self) -> str | None: + return self.get_type_id() if self.is_type else None + + @classmethod + def parse_type(cls, value: str) -> GtsID: + if not _is_type_ref(value.strip()): + raise GtsInvalidId(value, "must end with '~'") + return cls(value) def get_type_id(self) -> str | None: if len(self.gts_id_segments) < 2: @@ -297,10 +335,7 @@ def to_uuid(self) -> uuid.UUID: @classmethod def is_valid(cls, s: str) -> bool: - # Strip gts:// URI prefix if present - normalized = s - normalized = normalized.removeprefix(GTS_URI_PREFIX) - if not normalized.startswith(GTS_PREFIX): + if not _strip_scheme(s).startswith(GTS_PREFIX): return False try: _ = cls(s) @@ -451,3 +486,51 @@ def __init__(self, pattern: str): super().__init__(p) except GtsInvalidId as e: raise GtsInvalidWildcard(pattern, str(e)) + + +class GtsRef: + """Classification of a JSON Schema ``$ref`` value used in GTS documents. + + A ``$ref`` is exactly one of three kinds: + + - :attr:`LOCAL` - a same-document JSON Pointer (``#`` or ``#/...``). + - :attr:`GTS` - a reference to a GTS type, either as a ``gts://`` URI or in + the bare ``gts.`` form. + - :attr:`OTHER` - anything else (e.g. an external URL); not resolvable as a + GTS reference. + + Parsing normalizes the target once (see :attr:`target_id`) so callers never + strip the ``gts://`` scheme themselves. This is the single classifier for + ``$ref`` handling shared by the store, entity extraction and validation. + """ + + LOCAL = "local" + GTS = "gts" + OTHER = "other" + + def __init__(self, raw: str, kind: str, target_id: str, has_scheme: bool) -> None: + self.raw = raw + self.kind = kind + # Canonical bare target for non-local refs; empty for local pointers + # (use :attr:`is_local` to distinguish). + self.target_id = target_id + # Whether a GTS ref was written in explicit ``gts://`` URI form. + self.has_scheme = has_scheme + + @classmethod + def parse(cls, raw: str) -> GtsRef: + if raw.startswith("#"): + return cls(raw, cls.LOCAL, "", False) + scheme = _has_scheme(raw) + target = _strip_scheme(raw) + if scheme or target.startswith(GTS_PREFIX): + return cls(raw, cls.GTS, target, scheme) + return cls(raw, cls.OTHER, target, False) + + @property + def is_local(self) -> bool: + return self.kind == self.LOCAL + + @property + def is_gts(self) -> bool: + return self.kind == self.GTS diff --git a/gts/src/gts/ops.py b/gts/src/gts/ops.py index 8ae746c..2883a58 100644 --- a/gts/src/gts/ops.py +++ b/gts/src/gts/ops.py @@ -6,6 +6,7 @@ from pathlib import Path as SysPath from typing import Any +from ._naming import looks_like_gts from .entities import DEFAULT_GTS_CONFIG, GtsConfig, GtsEntity from .files_reader import GtsFileReader from .gts import GtsID, GtsWildcard @@ -557,7 +558,7 @@ def validate_json( try: explicit_type = GtsID(explicit_type_id) except ValueError: - if explicit_type_id.startswith(("gts.", "gts://")): + if looks_like_gts(explicit_type_id): return GtsJsonValidationResult( ok=False, error=f"Explicit type '{explicit_type_id}' must be GTS Type schema", diff --git a/gts/src/gts/schema_cast.py b/gts/src/gts/schema_cast.py index 6b32e90..b197d9c 100644 --- a/gts/src/gts/schema_cast.py +++ b/gts/src/gts/schema_cast.py @@ -6,9 +6,10 @@ from typing import Any from jsonschema import exceptions as js_exceptions -from jsonschema import validate as js_validate +from .compatibility import UNKNOWN, dialects_differ from .gts import GtsID +from .schema_validation import validator_for logger = logging.getLogger(__name__) @@ -25,9 +26,9 @@ class GtsEntityCastResult: added_properties: list[str] = None # type: ignore removed_properties: list[str] = None # type: ignore changed_properties: list[dict[str, str]] = None # type: ignore - is_fully_compatible: bool = False - is_backward_compatible: bool = False - is_forward_compatible: bool = False + is_fully_compatible: bool | None = False + is_backward_compatible: bool | None = False + is_forward_compatible: bool | None = False incompatibility_reasons: list[str] = None # type: ignore backward_errors: list[str] = None # type: ignore forward_errors: list[str] = None # type: ignore @@ -55,7 +56,9 @@ def __post_init__(self): self.forward_errors = [] def to_dict(self) -> dict[str, Any]: - def _compat_str(val: bool) -> str: + def _compat_str(val: bool | None) -> str: + if val is None: + return UNKNOWN return "compatible" if val else "incompatible" backward = self.backward_verdict or _compat_str(self.is_backward_compatible) @@ -120,6 +123,7 @@ def cast( is_forward, forward_errors = cls._check_forward_compatibility( old_schema, new_schema ) + dialect_changed = dialects_differ(old_schema, new_schema) # Apply casting rules to the instance added: list[str] = [] @@ -144,13 +148,16 @@ def cast( added_properties=sorted(dict.fromkeys(added)), removed_properties=sorted(dict.fromkeys(removed)), changed_properties=[], - is_fully_compatible=False, - is_backward_compatible=is_backward, - is_forward_compatible=is_forward, + is_fully_compatible=None if dialect_changed else False, + is_backward_compatible=None if dialect_changed else is_backward, + is_forward_compatible=None if dialect_changed else is_forward, incompatibility_reasons=[str(e)], backward_errors=backward_errors, forward_errors=forward_errors, casted_entity=None, + backward_verdict=UNKNOWN if dialect_changed else None, + forward_verdict=UNKNOWN if dialect_changed else None, + full_verdict=UNKNOWN if dialect_changed else None, ) # Validate the transformed instance against the FULL target schema @@ -172,13 +179,16 @@ def cast( added_properties=sorted(dict.fromkeys(added)), removed_properties=sorted(dict.fromkeys(removed)), changed_properties=[], - is_fully_compatible=is_fully_compatible, - is_backward_compatible=is_backward, - is_forward_compatible=is_forward, + is_fully_compatible=None if dialect_changed else is_fully_compatible, + is_backward_compatible=None if dialect_changed else is_backward, + is_forward_compatible=None if dialect_changed else is_forward, incompatibility_reasons=reasons, backward_errors=backward_errors, forward_errors=forward_errors, casted_entity=casted, + backward_verdict=UNKNOWN if dialect_changed else None, + forward_verdict=UNKNOWN if dialect_changed else None, + full_verdict=UNKNOWN if dialect_changed else None, ) @staticmethod @@ -388,10 +398,12 @@ def _validate_with_gts_id_tolerance( # Create a modified schema that removes const constraints for GTS IDs modified_schema = GtsEntityCastResult._remove_gts_const_constraints(schema) + validator_class = validator_for(modified_schema) if resolver is not None: - js_validate(instance=instance, schema=modified_schema, resolver=resolver) + validator = validator_class(modified_schema, resolver=resolver) else: - js_validate(instance=instance, schema=modified_schema) + validator = validator_class(modified_schema) + validator.validate(instance) @staticmethod def _remove_gts_const_constraints(schema: Any) -> Any: @@ -421,39 +433,141 @@ def _remove_gts_const_constraints(schema: Any) -> Any: return result + @staticmethod + def _merge_constraint_value(key: str, existing: Any, incoming: Any) -> Any | None: + if existing == incoming: + return copy.deepcopy(existing) + if key in { + "minimum", + "exclusiveMinimum", + "minLength", + "minItems", + "minProperties", + }: + return max(existing, incoming) + if key in { + "maximum", + "exclusiveMaximum", + "maxLength", + "maxItems", + "maxProperties", + }: + return min(existing, incoming) + if key == "enum" and isinstance(existing, list) and isinstance(incoming, list): + return [value for value in existing if value in incoming] + if key == "type": + existing_types = existing if isinstance(existing, list) else [existing] + incoming_types = incoming if isinstance(incoming, list) else [incoming] + common_types = [ + value for value in existing_types if value in incoming_types + ] + return common_types[0] if len(common_types) == 1 else common_types + if key == "additionalProperties": + if existing is False or incoming is False: + return False + if existing is True: + return copy.deepcopy(incoming) + if incoming is True: + return copy.deepcopy(existing) + return None + @staticmethod def _flatten_property_schema(schema: dict[str, Any]) -> dict[str, Any]: result: dict[str, Any] = {} + + def merge(key: str, value: Any) -> None: + if key not in result: + result[key] = copy.deepcopy(value) + return + merged = GtsEntityCastResult._merge_constraint_value( + key, result[key], value + ) + if merged is None: + result.setdefault("allOf", []).append({key: copy.deepcopy(value)}) + else: + result[key] = merged + for sub_schema in schema.get("allOf", []): if isinstance(sub_schema, dict): - result.update(GtsEntityCastResult._flatten_property_schema(sub_schema)) - result.update({key: value for key, value in schema.items() if key != "allOf"}) + for key, value in GtsEntityCastResult._flatten_property_schema( + sub_schema + ).items(): + merge(key, value) + for key, value in schema.items(): + if key != "allOf": + merge(key, value) return result + @staticmethod + def _merge_property_schemas(existing: Any, incoming: Any) -> Any: + if existing == incoming: + return copy.deepcopy(existing) + # Intersect the two property sub-schemas so repeated constraints (e.g. + # minLength across allOf members) collapse to their tightest value. + # Non-dict schemas (e.g. boolean) cannot be intersected, so fall back to + # the conservative allOf wrapper. + if not isinstance(existing, dict) or not isinstance(incoming, dict): + return {"allOf": [copy.deepcopy(existing), copy.deepcopy(incoming)]} + return GtsEntityCastResult._flatten_property_schema( + {"allOf": [copy.deepcopy(existing), copy.deepcopy(incoming)]} + ) + @staticmethod def _flatten_schema(schema: dict[str, Any]) -> dict[str, Any]: - """Flatten a schema by merging allOf schemas.""" - result = {"properties": {}, "required": []} - - # Merge allOf schemas - if "allOf" in schema: - for sub_schema in schema["allOf"]: - flattened = GtsEntityCastResult._flatten_schema(sub_schema) - result["properties"].update(flattened.get("properties", {})) - result["required"].extend(flattened.get("required", [])) - # Preserve additionalProperties from sub-schemas (last one wins) - if "additionalProperties" in flattened: - result["additionalProperties"] = flattened["additionalProperties"] - - # Add direct properties and required - if "properties" in schema: - result["properties"].update(schema["properties"]) - if "required" in schema: - result["required"].extend(schema["required"]) - # Preserve additionalProperties from top level (overrides sub-schemas) - if "additionalProperties" in schema: - result["additionalProperties"] = schema["additionalProperties"] + """Flatten object-specific allOf members while retaining schema keywords.""" + result = { + key: copy.deepcopy(value) + for key, value in schema.items() + if key not in {"allOf", "properties", "required"} + } + result["properties"] = {} + result["required"] = [] + for sub_schema in schema.get("allOf", []): + if not isinstance(sub_schema, dict): + continue + flattened = GtsEntityCastResult._flatten_schema(sub_schema) + for key, value in flattened.items(): + if key == "properties": + for name, prop_schema in value.items(): + if name in result["properties"]: + result["properties"][name] = ( + GtsEntityCastResult._merge_property_schemas( + result["properties"][name], prop_schema + ) + ) + else: + result["properties"][name] = copy.deepcopy(prop_schema) + elif key == "required": + result["required"].extend(value) + elif key not in result: + result[key] = copy.deepcopy(value) + else: + merged = GtsEntityCastResult._merge_constraint_value( + key, result[key], value + ) + if merged is None: + result.setdefault("allOf", []).append( + {key: copy.deepcopy(value)} + ) + else: + result[key] = merged + + properties = schema.get("properties") + if isinstance(properties, dict): + for name, prop_schema in properties.items(): + if name in result["properties"]: + result["properties"][name] = ( + GtsEntityCastResult._merge_property_schemas( + result["properties"][name], prop_schema + ) + ) + else: + result["properties"][name] = copy.deepcopy(prop_schema) + required = schema.get("required") + if isinstance(required, list): + result["required"].extend(required) + result["required"] = list(dict.fromkeys(result["required"])) return result @staticmethod @@ -641,29 +755,31 @@ def _check_schema_compatibility( ) # Check enum constraints - old_enum = old_prop_schema.get("enum") - new_enum = new_prop_schema.get("enum") - if old_enum and new_enum: - old_enum_set = set(old_enum) - new_enum_set = set(new_enum) + has_old_enum = "enum" in old_prop_schema + has_new_enum = "enum" in new_prop_schema + old_enum = old_prop_schema.get("enum", []) + new_enum = new_prop_schema.get("enum", []) + if has_old_enum and has_new_enum: if check_backward: - # Backward: cannot add enum values - added_enum_values = new_enum_set - old_enum_set + added_enum_values = [ + value for value in new_enum if value not in old_enum + ] if added_enum_values: errors.append( f"Property '{prop}' added enum values: {added_enum_values}" ) else: - # Forward: cannot remove enum values - removed_enum_values = old_enum_set - new_enum_set + removed_enum_values = [ + value for value in old_enum if value not in new_enum + ] if removed_enum_values: errors.append( f"Property '{prop}' removed enum values: {removed_enum_values}" ) - elif old_enum: + elif has_old_enum: if not check_backward: errors.append(f"Property '{prop}' removed enum constraint") - elif new_enum and check_backward: + elif has_new_enum and check_backward: errors.append(f"Property '{prop}' added enum constraint") # Check constraint compatibility diff --git a/gts/src/gts/schema_validation.py b/gts/src/gts/schema_validation.py new file mode 100644 index 0000000..95b61b7 --- /dev/null +++ b/gts/src/gts/schema_validation.py @@ -0,0 +1,45 @@ +from __future__ import annotations + +from collections.abc import Iterator +from typing import Any + +import regex +from jsonschema import Draft7Validator, FormatChecker, ValidationError, validators +from jsonschema.validators import validator_for as jsonschema_validator_for + +PATTERN_TIMEOUT_SECONDS = 1.0 + +# Shared format checker for instance/trait validation. +# +# A bare ``FormatChecker()`` draws from jsonschema's shared, class-level checker +# registry, in which draft3-only checkers overwrite their draft6/7 successors +# (e.g. "time" ends up as bare ``HH:MM:SS`` and rejects a valid RFC 3339 value +# like "10:30:00Z"). Conversely, ``Draft7Validator.FORMAT_CHECKER`` fixes those +# but lacks "uuid" (added to JSON Schema only in draft 2019-09). +# +# Starting from the bare checker (which provides "uuid") and overlaying the +# draft-07 checkers (which restore correct RFC 3339 "time"/"date-time") yields +# the complete, correct standard-format set. This uses only built-in jsonschema +# checkers -- no custom format functions. +FORMAT_CHECKER = FormatChecker() +FORMAT_CHECKER.checkers.update(Draft7Validator.FORMAT_CHECKER.checkers) + + +def _validate_pattern( + validator: Any, pattern: str, instance: Any, schema: Any +) -> Iterator[ValidationError]: + if not isinstance(instance, str): + return + try: + if regex.search(pattern, instance, timeout=PATTERN_TIMEOUT_SECONDS) is None: + yield ValidationError(f"{instance!r} does not match {pattern!r}") + except TimeoutError: + yield ValidationError("regular expression match timed out") + except regex.error as error: + yield ValidationError(f"invalid regular expression: {error}") + + +def validator_for(schema: Any) -> Any: + return validators.extend( + jsonschema_validator_for(schema), {"pattern": _validate_pattern} + ) diff --git a/gts/src/gts/store.py b/gts/src/gts/store.py index 72edaa3..1a489b4 100644 --- a/gts/src/gts/store.py +++ b/gts/src/gts/store.py @@ -7,19 +7,27 @@ from typing import Any from jsonschema import RefResolver -from jsonschema.validators import validator_for from referencing import Registry, Resource from referencing.jsonschema import DRAFT202012 from . import compatibility, derivation, traits +from ._naming import looks_like_gts, strip_scheme, with_scheme from .entities import GtsEntity -from .gts import GtsID, GtsWildcard +from .gts import GtsID, GtsRef, GtsWildcard from .schema_cast import GtsEntityCastResult +from .schema_validation import FORMAT_CHECKER, validator_for from .x_gts_ref import XGtsRefValidator, _without_x_gts_ref logger = logging.getLogger(__name__) +def _require_schema_id(value: str) -> GtsID: + try: + return GtsID.parse_type(value) + except ValueError as error: + raise ValueError(f"ID '{value}' is not a schema (must end with '~')") from error + + class StoreGtsObjectNotFound(Exception): """Exception raised when a GTS entity is not found in the store.""" @@ -159,19 +167,21 @@ def register_schema(self, type_id: str, schema: dict[str, Any]) -> None: Register a schema (legacy method for backward compatibility). Creates a JsonEntity from the schema dict. """ - if not type_id.endswith("~"): - raise ValueError("Schema type_id must end with '~'") - # parse sanity - gts_id = GtsID(type_id) + gts_id = GtsID.parse_type(type_id) entity = GtsEntity(content=schema, gts_id=gts_id, is_schema=True) - self._by_id[type_id] = entity + self._by_id[gts_id.id] = entity def get(self, entity_id: str) -> GtsEntity | None: """ Get a JsonEntity by its ID. If not found in cache, try to fetch from reader. Returns None if not found. + + Lookups are normalized to the canonical bare form here, so callers may + pass either a bare ``gts.`` id or a ``gts://`` URI without stripping the + scheme themselves. """ + entity_id = strip_scheme(entity_id) # Check cache first if entity_id in self._by_id: return self._by_id[entity_id] @@ -196,13 +206,14 @@ def _create_ref_resolver(self, schema: dict[str, Any]) -> RefResolver: """Create a custom RefResolver that can resolve GTS ID references from the store.""" def resolve_gts_ref(uri: str) -> dict[str, Any]: - """Resolve a GTS ID reference to its schema content.""" - # Issue #32: handle gts:// prefix - uri = uri.removeprefix("gts://") + """Resolve a GTS ID reference to its schema content. + + ``get_schema_content`` normalizes the ``gts://`` scheme internally. + """ try: return self.get_schema_content(uri) except KeyError as e: - raise ValueError(f"Unresolvable: {uri}") from e + raise ValueError(f"Unresolvable: {strip_scheme(uri)}") from e # Create a store dict that maps GTS IDs to their schema content store = {} @@ -224,7 +235,7 @@ def _create_reference_registry(self) -> Registry: _without_x_gts_ref(entity.content), default_specification=DRAFT202012, ) - registry = registry.with_resource(f"gts://{entity_id}", resource) + registry = registry.with_resource(with_scheme(entity_id), resource) return registry def items(self): @@ -254,19 +265,18 @@ def _validate_schema_refs(schema: dict[str, Any], path: str = "") -> None: ref_uri = schema["$ref"] if isinstance(ref_uri, str): current_path = f"{path}.$ref" if path else "$ref" - - # Local refs (JSON Pointer) are always valid - if ref_uri.startswith("#"): - pass # Valid local ref - # GTS refs must use gts:// URI format - elif ref_uri.startswith("gts://"): - gts_id = ref_uri[6:] # Strip prefix - # Validate the GTS ID - if not GtsID.is_valid(gts_id): + ref = GtsRef.parse(ref_uri) + + # Local refs (JSON Pointer) are always valid. + if ref.is_local: + pass + # External GTS refs MUST use the gts:// URI form. + elif ref.is_gts and ref.has_scheme: + if not GtsID.is_valid(ref.target_id): raise ValueError( - f"Invalid $ref at '{current_path}': '{ref_uri}' contains invalid GTS identifier '{gts_id}'" + f"Invalid $ref at '{current_path}': '{ref_uri}' contains invalid GTS identifier '{ref.target_id}'" ) - # Any other external ref is invalid + # Anything else (bare gts., external URL, ...) is invalid. else: raise ValueError( f"Invalid $ref at '{current_path}': '{ref_uri}' must be a local ref (starting with '#') " @@ -292,17 +302,15 @@ def _validate_schema_x_gts_refs(self, gts_id: str) -> None: Args: gts_id: The GTS ID of the schema to validate """ - if not gts_id.endswith("~"): - raise ValueError(f"ID '{gts_id}' is not a schema (must end with '~')") - - schema_entity = self.get(gts_id) + schema_id = _require_schema_id(gts_id) + schema_entity = self.get(schema_id.id) if not schema_entity: - raise StoreGtsSchemaNotFound(gts_id) + raise StoreGtsSchemaNotFound(schema_id.id) if not schema_entity.is_schema: - raise ValueError(f"Entity '{gts_id}' is not a schema") + raise ValueError(f"Entity '{schema_id.id}' is not a schema") - self._validate_schema_x_gts_refs_content(gts_id, schema_entity.content) + self._validate_schema_x_gts_refs_content(schema_id.id, schema_entity.content) def _validate_schema_x_gts_refs_content( self, gts_id: str, schema_content: dict[str, Any] @@ -489,11 +497,10 @@ def _inline_refs( if isinstance(node, dict): ref_uri = node.get("$ref") if isinstance(ref_uri, str): - ref_id: str | None = None - if ref_uri.startswith("gts://"): - ref_id = ref_uri[6:] - elif not ref_uri.startswith("#"): - ref_id = ref_uri + ref = GtsRef.parse(ref_uri) + # Local (#/...) refs are resolved by JSON Schema itself; only + # external targets are inlined from the store. + ref_id = None if ref.is_local else ref.target_id if ref_id is not None: if ref_id in seen: # Cycle detected: leave the $ref unresolved. @@ -613,15 +620,13 @@ def validate_schema_basic(self, gts_id: str) -> None: 3. GTS keyword validation (x-gts-final, x-gts-abstract, placement) 4. JSON Schema meta-schema validation """ - if not gts_id.endswith("~"): - raise ValueError(f"ID '{gts_id}' is not a schema (must end with '~')") - - schema_entity = self.get(gts_id) + schema_id = _require_schema_id(gts_id) + schema_entity = self.get(schema_id.id) if not schema_entity: - raise StoreGtsSchemaNotFound(gts_id) + raise StoreGtsSchemaNotFound(schema_id.id) if not schema_entity.is_schema: - raise ValueError(f"Entity '{gts_id}' is not a schema") + raise ValueError(f"Entity '{schema_id.id}' is not a schema") schema_content = schema_entity.content if not isinstance(schema_content, dict): @@ -633,7 +638,7 @@ def validate_schema_basic(self, gts_id: str) -> None: if ( meta_schema_url and isinstance(meta_schema_url, str) - and meta_schema_url.startswith(("gts.", "gts://")) + and looks_like_gts(meta_schema_url) ): raise ValueError( f"Invalid $schema URL '{meta_schema_url}': must be a standard JSON Schema URL, not a GTS ID" @@ -652,15 +657,13 @@ def validate_schema_content( self, gts_id: str, schema_content: dict[str, Any] ) -> None: """Validate a schema using the registry only for its dependencies.""" - schema_id = GtsID(gts_id) - if not schema_id.is_type: - raise ValueError(f"ID '{gts_id}' is not a schema (must end with '~')") + schema_id = _require_schema_id(gts_id) meta_schema_url = schema_content.get("$schema") if ( meta_schema_url and isinstance(meta_schema_url, str) - and meta_schema_url.startswith(("gts.", "gts://")) + and looks_like_gts(meta_schema_url) ): raise ValueError( f"Invalid $schema URL '{meta_schema_url}': must be a standard JSON Schema URL, not a GTS ID" @@ -674,7 +677,6 @@ def validate_schema_content( try: from jsonschema import Draft7Validator - from jsonschema.validators import validator_for if meta_schema_url: validator_class = validator_for({"$schema": meta_schema_url}) @@ -698,14 +700,7 @@ def validate_schema_content( def validate_schema(self, gts_id: str) -> None: """Validate a registered schema and all of its dependencies.""" - try: - schema_id = GtsID(gts_id) - except ValueError as error: - raise ValueError( - f"ID '{gts_id}' is not a schema (must end with '~')" - ) from error - if not schema_id.is_type: - raise ValueError(f"ID '{gts_id}' is not a schema (must end with '~')") + schema_id = _require_schema_id(gts_id) schema_entity = self.get(schema_id.id) if not schema_entity: @@ -720,9 +715,7 @@ def validate_schema(self, gts_id: str) -> None: def validate_instance_content(self, content: dict[str, Any], type_id: str) -> None: """Validate unregistered instance content against a registered type schema.""" - schema_type = GtsID(type_id) - if not schema_type.is_type: - raise ValueError(f"ID '{type_id}' is not a schema (must end with '~')") + schema_type = _require_schema_id(type_id) try: schema = self.get_schema_content(schema_type.id) except KeyError as error: @@ -736,7 +729,9 @@ def validate_instance_content(self, content: dict[str, Any], type_id: str) -> No schema_for_validation = _without_x_gts_ref(schema) validator_class = validator_for(schema_for_validation) validator = validator_class( - schema_for_validation, registry=self._create_reference_registry() + schema_for_validation, + registry=self._create_reference_registry(), + format_checker=FORMAT_CHECKER, ) validator.validate(content) diff --git a/gts/src/gts/traits.py b/gts/src/gts/traits.py index 2488680..7d7f954 100644 --- a/gts/src/gts/traits.py +++ b/gts/src/gts/traits.py @@ -19,18 +19,16 @@ import copy from typing import Any -from jsonschema import Draft7Validator, FormatChecker -from jsonschema.validators import validator_for - from . import derivation +from ._json_pointer import resolve as resolve_json_pointer +from .schema_validation import FORMAT_CHECKER as _FORMAT_CHECKER +from .schema_validation import validator_for from .x_gts_ref import XGtsRefValidator X_GTS_TRAITS_SCHEMA = "x-gts-traits-schema" X_GTS_TRAITS = "x-gts-traits" MAX_RECURSION_DEPTH = 64 _MISSING = object() -_FORMAT_CHECKER = FormatChecker() -_FORMAT_CHECKER.checkers.update(Draft7Validator.FORMAT_CHECKER.checkers) class EffectiveTraits: @@ -120,7 +118,7 @@ def inline_local_pointers(fragment: Any, root: Any, depth: int = 0) -> Any: if isinstance(fragment, dict): ref = fragment.get("$ref") if isinstance(ref, str) and ref.startswith("#/"): - target = _resolve_json_pointer(root, ref[1:]) + target = resolve_json_pointer(root, ref) if target is not None: resolved = inline_local_pointers(target, root, depth + 1) if len(fragment) > 1 and isinstance(resolved, dict): @@ -136,24 +134,6 @@ def inline_local_pointers(fragment: Any, root: Any, depth: int = 0) -> Any: return copy.deepcopy(fragment) -def _resolve_json_pointer(root: Any, pointer: str) -> Any: - # pointer begins with '/' - parts = [p for p in pointer.split("/") if p != ""] - current = root - for part in parts: - part = part.replace("~1", "/").replace("~0", "~") - if isinstance(current, dict) and part in current: - current = current[part] - elif isinstance(current, list): - try: - current = current[int(part)] - except (ValueError, IndexError): - return None - else: - return None - return current - - # --- RFC 7396 merge -------------------------------------------------------- def merge_rfc7396_into( target: dict[str, Any], patch: dict[str, Any], depth: int = 0 diff --git a/gts/src/gts/x_gts_ref.py b/gts/src/gts/x_gts_ref.py index 726258b..f410273 100644 --- a/gts/src/gts/x_gts_ref.py +++ b/gts/src/gts/x_gts_ref.py @@ -16,7 +16,10 @@ from jsonschema.validators import validator_for -from .gts import GTS_URI_PREFIX, GtsID +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 def _without_x_gts_ref(schema: Any) -> Any: @@ -102,11 +105,23 @@ def validate_instance( """ errors: list[XGtsRefValidationError] = [] - def visit_instance(inst, sch, path, errs): + def resolve_local_ref(ref: str) -> Any | None: + if ref != "#" and not ref.startswith("#/"): + return None + return resolve_json_pointer(schema, ref, default=None) + + def visit_instance(inst, sch, path, errs, refs=None): """Visit instance nodes and validate x-gts-ref constraints.""" if not isinstance(sch, dict): return + refs = refs or set() + ref = sch.get("$ref") + if isinstance(ref, str) and ref not in refs: + target = resolve_local_ref(ref) + if target is not 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) if error: @@ -180,17 +195,14 @@ def visit_instance(inst, sch, path, errs): if _is_structurally_valid(inst, branch): errs.extend(_validate_branch(inst, branch, path)) - if ( - sch.get("type") == "object" - and "properties" in sch - and isinstance(inst, dict) - ): - for prop_name, prop_schema in sch["properties"].items(): + properties = sch.get("properties") + if isinstance(properties, dict) and isinstance(inst, dict): + for prop_name, prop_schema in properties.items(): if prop_name in inst: prop_path = f"{path}.{prop_name}" if path else prop_name visit_instance(inst[prop_name], prop_schema, prop_path, errs) - if sch.get("type") == "array" and "items" in sch and isinstance(inst, list): + 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) @@ -323,7 +335,7 @@ def _validate_ref_pattern( ) # Case 1: Absolute GTS pattern - if ref_pattern.startswith("gts."): + if ref_pattern.startswith(GTS_PREFIX): return self._validate_gts_id_or_pattern(ref_pattern, field_path) # Case 2: Relative reference @@ -362,7 +374,7 @@ def _validate_gts_id_or_pattern( if "*" in pattern: # Wildcard pattern - validate prefix prefix = pattern.rstrip("*") - if not prefix.startswith("gts."): + if not prefix.startswith(GTS_PREFIX): return XGtsRefValidationError( field_path, pattern, @@ -436,41 +448,24 @@ def _validate_gts_pattern( return None - def _normalize_gts_value(self, value: str) -> str: - """Strip gts:// URI prefix if present.""" - if value.startswith(GTS_URI_PREFIX): - return value[len(GTS_URI_PREFIX) :] - return value - def _resolve_pointer(self, schema: dict[str, Any], pointer: str) -> str | None: """ - Resolve a JSON Pointer in the schema. + 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 value or None if not found + The resolved GTS identifier (bare form) or None if not found. """ - - path = pointer.lstrip("/") - if not path: + current = resolve_json_pointer(schema, pointer, default=MISSING) + if current is MISSING or current is None: return None - parts = path.split("/") - current = schema - - for part in parts: - if not isinstance(current, dict): - return None - current = current.get(part) - if current is None: - return None - - # If current is a string, return it (normalizing gts:// prefix) + # If current is a string, return it (normalized to the bare form). if isinstance(current, str): - return self._normalize_gts_value(current) + 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: @@ -478,6 +473,6 @@ def _resolve_pointer(self, schema: dict[str, Any], pointer: str) -> str | None: if isinstance(ref_value, str): if ref_value.startswith("/"): return self._resolve_pointer(schema, ref_value) - return self._normalize_gts_value(ref_value) + return strip_scheme(ref_value) return None diff --git a/requirements.txt b/requirements.txt index b88e957..35f3a12 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,4 +1,16 @@ -jsonschema -dataclasses -fastapi -uvicorn +# Development / test tooling for the local GTS venv (see Makefile `py-env`). +# +# Runtime dependencies live in gts/pyproject.toml. This file holds the extra +# tooling plus version overrides needed by the local workflow, where the GTS +# server and the httprunner-based spec test client share a single virtualenv. +# +# It is installed AFTER .gts-spec/tests/requirements.txt so the pydantic pin +# below overrides the spec client's `pydantic<2` pin: the server uses pydantic +# v2 APIs (e.g. model_validator) and httprunner runs fine on v2. (httprunner +# itself is installed separately with --no-deps in the Makefile because its +# declared pins are incompatible with this venv.) +pydantic>=2,<3 + +# Linters / type checker. +ruff +mypy diff --git a/tests/test_compatibility.py b/tests/test_compatibility.py index 7ba6083..ed0f355 100644 --- a/tests/test_compatibility.py +++ b/tests/test_compatibility.py @@ -74,6 +74,18 @@ def test_keeps_type_when_enum_has_other_types(self): assert result.get("type") == "string" +class TestUnevaluatedPropertiesLowering: + def test_conditional_applicators_are_not_lowered(self): + from gts.compatibility import _lower_root_unevaluated_properties + + assert ( + _lower_root_unevaluated_properties( + {"unevaluatedProperties": False, "if": {"properties": {"a": True}}} + ) + is None + ) + + class TestCompatibilityVerdicts: def test_backward_compatible_widened_enum(self): # old accepted set must be subset of new diff --git a/tests/test_gts_id.py b/tests/test_gts_id.py index f5ba85c..ba613f5 100644 --- a/tests/test_gts_id.py +++ b/tests/test_gts_id.py @@ -1,16 +1,17 @@ """Tests for GTS ID parsing and validation.""" -import pytest import uuid +import pytest + from gts.gts import ( + GTS_PREFIX, GtsID, GtsIdSegment, - GtsWildcard, GtsInvalidId, GtsInvalidSegment, GtsInvalidWildcard, - GTS_PREFIX, + GtsWildcard, ) @@ -140,6 +141,48 @@ def test_get_type_id(self): type_id = gts_id.get_type_id() assert type_id == "gts.vendor.package.namespace.type.v1~" + @pytest.mark.parametrize( + ("value", "is_type", "type_id", "parent_type_id"), + [ + ( + "gts.vendor.package.namespace.type.v1~", + True, + "gts.vendor.package.namespace.type.v1~", + None, + ), + ( + "gts.vendor.package.namespace.type.v1~vendor.package.namespace.child.v1~", + True, + "gts.vendor.package.namespace.type.v1~vendor.package.namespace.child.v1~", + "gts.vendor.package.namespace.type.v1~", + ), + ( + "gts.vendor.package.namespace.type.v1~vendor.package.namespace.instance.v1", + False, + "gts.vendor.package.namespace.type.v1~", + None, + ), + ( + "gts.vendor.package.namespace.type.v1~7a1d2f34-5678-49ab-9012-abcdef123456", + False, + "gts.vendor.package.namespace.type.v1~", + None, + ), + ], + ) + def test_resolves_type_identity(self, value, is_type, type_id, parent_type_id): + gts_id = GtsID(value) + assert gts_id.is_type is is_type + assert gts_id.is_instance is not is_type + assert gts_id.type_id == type_id + assert gts_id.parent_type_id == parent_type_id + + def test_parse_type_rejects_instance(self): + with pytest.raises(GtsInvalidId, match="must end with '~'"): + GtsID.parse_type( + "gts.vendor.package.namespace.type.v1~vendor.package.namespace.instance.v1" + ) + def test_to_uuid(self): """Test UUID generation is deterministic.""" gts_id1 = GtsID("gts.vendor.package.namespace.type.v1~") @@ -244,9 +287,7 @@ def test_underscore_in_tokens_allowed(self): def test_combined_anonymous_id_uses_embedded_uuid(self): embedded_uuid = "7a1d2f34-5678-49ab-9012-abcdef123456" - gts_id = GtsID( - "gts.vendor.package.namespace.type.v1~" + embedded_uuid - ) + gts_id = GtsID("gts.vendor.package.namespace.type.v1~" + embedded_uuid) assert gts_id.uuid_tail == embedded_uuid assert gts_id.to_uuid() == uuid.UUID(embedded_uuid) diff --git a/tests/test_schema_cast.py b/tests/test_schema_cast.py index 454cf08..f0a9451 100644 --- a/tests/test_schema_cast.py +++ b/tests/test_schema_cast.py @@ -69,7 +69,9 @@ def test_none_direction_same_minor(self): ) def test_unknown_on_invalid_id(self): - assert GtsEntityCastResult._infer_direction("not-an-id", "also-not") == "unknown" + assert ( + GtsEntityCastResult._infer_direction("not-an-id", "also-not") == "unknown" + ) def test_combined_anonymous_id_uses_versioned_segment(self): # Regression: the appended UUID-tail segment has ver_minor=None; the @@ -114,13 +116,33 @@ def test_merges_allof(self): assert set(flat["properties"].keys()) == {"a", "b"} assert set(flat["required"]) == {"a", "b"} - def test_additional_properties_top_level_overrides(self): + def test_additional_properties_intersects_allof_members(self): schema = { "allOf": [{"additionalProperties": False}], "additionalProperties": True, } flat = GtsEntityCastResult._flatten_schema(schema) - assert flat["additionalProperties"] is True + assert flat["additionalProperties"] is False + + def test_intersected_additional_properties_removes_extra_values(self): + flat = GtsEntityCastResult._flatten_schema( + {"allOf": [{"additionalProperties": False}], "additionalProperties": True} + ) + casted, _, removed, _ = GtsEntityCastResult._cast_instance_to_schema( + {"extra": "value"}, flat + ) + assert casted == {} + assert removed == ["extra"] + + def test_intersects_repeated_property_constraints(self): + schema = { + "allOf": [ + {"properties": {"name": {"type": "string", "minLength": 1}}}, + {"properties": {"name": {"type": "string", "minLength": 5}}}, + ] + } + flat = GtsEntityCastResult._flatten_schema(schema) + assert flat["properties"]["name"]["minLength"] == 5 class TestCastInstanceToSchema: @@ -293,6 +315,22 @@ def test_string_constraints_checked(self): ) assert errors + def test_allof_tightened_string_constraint_is_checked(self): + old = {"properties": {"name": {"type": "string", "minLength": 1}}} + new = { + "properties": { + "name": { + "allOf": [ + {"type": "string", "minLength": 1}, + {"type": "string", "minLength": 5}, + ] + } + } + } + compatible, errors = GtsEntityCastResult._check_backward_compatibility(old, new) + assert compatible is False + assert any("minLength increased" in error for error in errors) + def test_array_constraints_checked(self): errors = GtsEntityCastResult._check_constraint_compatibility( "p", {"type": "array", "minItems": 1}, {"type": "array", "minItems": 5} @@ -357,8 +395,16 @@ def test_forward_removed_enum_constraint_flagged(self): assert any("removed enum constraint" in e for e in errors) def test_nested_object_errors_prefixed(self): - old = {"properties": {"a": {"type": "object", "properties": {"b": {"type": "string"}}}}} - new = {"properties": {"a": {"type": "object", "properties": {"b": {"type": "integer"}}}}} + old = { + "properties": { + "a": {"type": "object", "properties": {"b": {"type": "string"}}} + } + } + new = { + "properties": { + "a": {"type": "object", "properties": {"b": {"type": "integer"}}} + } + } ok, errors = GtsEntityCastResult._check_backward_compatibility(old, new) assert not ok assert any("Property 'a':" in e for e in errors) @@ -366,7 +412,9 @@ def test_nested_object_errors_prefixed(self): def test_fully_compatible_returns_true(self): old = {"properties": {"a": {"type": "string"}}} new = {"properties": {"a": {"type": "string"}}, "properties2": {}} - ok, errors = GtsEntityCastResult._check_backward_compatibility(old, {"properties": {"a": {"type": "string"}}}) + ok, errors = GtsEntityCastResult._check_backward_compatibility( + old, {"properties": {"a": {"type": "string"}}} + ) assert ok assert errors == [] @@ -492,6 +540,31 @@ def test_cast_fully_compatible(self): assert result.is_fully_compatible is True assert result.casted_entity == {"a": "x"} + def test_cast_across_distinct_dialects_has_unknown_compatibility(self): + result = GtsEntityCastResult.cast( + "gts.x.test._.foo.v1.0~x.test._.bar.v1.0", + "gts.x.test._.foo.v1.1~", + {"status": "active"}, + { + "$schema": "https://json-schema.org/draft-07/schema", + "type": "object", + "properties": {"status": {"type": "string"}}, + }, + { + "$schema": "http://json-schema.org/draft/2020-12/schema#", + "type": "object", + "properties": {"status": {"type": "string"}}, + }, + ) + + assert result.casted_entity == {"status": "active"} + assert result.to_dict()["backward_compatibility"] == "unknown" + assert result.to_dict()["forward_compatibility"] == "unknown" + assert result.to_dict()["full_compatibility"] == "unknown" + assert result.is_backward_compatible is None + assert result.is_forward_compatible is None + assert result.is_fully_compatible is None + def test_cast_with_non_dict_instance_content_defaults_to_empty(self): result = GtsEntityCastResult.cast( "gts.x.test._.foo.v1.0~x.test._.bar.v1.0", diff --git a/tests/test_store.py b/tests/test_store.py index 778ccdc..d8ea42c 100644 --- a/tests/test_store.py +++ b/tests/test_store.py @@ -1,8 +1,10 @@ """Tests for GtsStore and GtsReader.""" -import pytest from typing import Iterator, Optional +import pytest +from jsonschema.exceptions import ValidationError + from gts.store import ( GtsStore, GtsReader, @@ -298,6 +300,38 @@ def test_validate_instance_valid(self): "gts.vendor.package.namespace.type.v1~vendor.package.namespace.inst.v1" ) + def test_validate_instance_content_enforces_standard_formats(self): + store = GtsStore(MockGtsReader([])) + type_id = "gts.vendor.package.namespace.type.v1~" + store.register_schema( + type_id, + { + "$schema": "http://json-schema.org/draft-07/schema#", + "type": "object", + "properties": { + "uuid": {"type": "string", "format": "uuid"}, + "time": {"type": "string", "format": "time"}, + }, + "required": ["uuid", "time"], + }, + ) + + # RFC 3339 "time" (draft-07) requires a timezone offset, so a valid + # value must carry one (e.g. the "Z" UTC designator). + store.validate_instance_content( + {"uuid": "550e8400-e29b-41d4-a716-446655440000", "time": "10:30:00Z"}, + type_id, + ) + with pytest.raises(ValidationError): + store.validate_instance_content( + {"uuid": "not-a-uuid", "time": "10:30:00Z"}, type_id + ) + with pytest.raises(ValidationError): + store.validate_instance_content( + {"uuid": "550e8400-e29b-41d4-a716-446655440000", "time": "25:99:99Z"}, + type_id, + ) + def test_validate_instance_not_found(self): """Test validating non-existent instance.""" reader = MockGtsReader([]) diff --git a/tests/test_store_extra.py b/tests/test_store_extra.py index 4b42ec2..ac90fad 100644 --- a/tests/test_store_extra.py +++ b/tests/test_store_extra.py @@ -3,9 +3,15 @@ import pytest from typing import Iterator, Optional -from gts.store import GtsStore, GtsReader, StoreGtsEntityNotFound, StoreGtsObjectNotFound +from gts.store import ( + GtsStore, + GtsReader, + 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): @@ -47,6 +53,15 @@ def _schema_entity(gts_id: str, content_extra=None): return GtsEntity(content=content, gts_id=GtsID(gts_id), is_schema=True) +class TestBoundedPatternValidation: + def test_catastrophic_pattern_times_out(self): + validator = validator_for({"pattern": "(a+)+$"})({"pattern": "(a+)+$"}) + errors = list(validator.iter_errors("a" * 30_000 + "!")) + assert len(errors) == 1 + assert errors[0].message == "regular expression match timed out" + assert PATTERN_TIMEOUT_SECONDS == 1.0 + + class TestRegisterEdgeCases: def test_register_raises_without_id(self): store = GtsStore(reader=None) @@ -149,9 +164,7 @@ def test_entity_not_schema_raises(self): store._validate_schema_x_gts_refs("gts.x.test._.foo.v1~") def test_invalid_x_gts_ref_raises(self): - schema = _schema_entity( - "gts.x.test._.foo.v1~", {"x-gts-ref": "notgts.*"} - ) + schema = _schema_entity("gts.x.test._.foo.v1~", {"x-gts-ref": "notgts.*"}) store = GtsStore(reader=None) store.register(schema) with pytest.raises(Exception, match="x-gts-ref validation failed"): @@ -211,12 +224,8 @@ def test_unresolvable_ref_left_unresolved(self): assert resolved == schema def test_cyclic_ref_left_unresolved(self): - a = _schema_entity( - "gts.x.test._.a.v1~", {"$ref": "gts://gts.x.test._.b.v1~"} - ) - b = _schema_entity( - "gts.x.test._.b.v1~", {"$ref": "gts://gts.x.test._.a.v1~"} - ) + a = _schema_entity("gts.x.test._.a.v1~", {"$ref": "gts://gts.x.test._.b.v1~"}) + b = _schema_entity("gts.x.test._.b.v1~", {"$ref": "gts://gts.x.test._.a.v1~"}) store = GtsStore(reader=None) store.register(a) store.register(b) @@ -254,7 +263,12 @@ def _build_store(self): old_schema = _schema_entity("gts.x.test._.foo.v1.0~") new_schema = _schema_entity( "gts.x.test._.foo.v1.5~", - {"properties": {"name": {"type": "string"}, "extra": {"type": "string", "default": "d"}}}, + { + "properties": { + "name": {"type": "string"}, + "extra": {"type": "string", "default": "d"}, + } + }, ) instance = GtsEntity( content={ @@ -279,7 +293,9 @@ def test_cast_success(self): def test_cast_from_missing_entity_raises(self): store, *_ = self._build_store() with pytest.raises(StoreGtsEntityNotFound): - store.cast("gts.x.test._.foo.v1.0~x.test._.missing.v1.0", "gts.x.test._.foo.v1.5~") + store.cast( + "gts.x.test._.foo.v1.0~x.test._.missing.v1.0", "gts.x.test._.foo.v1.5~" + ) def test_cast_from_schema_raises(self): store, old_schema, new_schema, instance = self._build_store() @@ -382,9 +398,7 @@ def test_query_result_to_dict_ok(self): class TestValidateSchemaFullFlow: def test_meta_schema_url_rejects_gts_id(self): - schema = _schema_entity( - "gts.x.test._.foo.v1~", {"$schema": "gts.x.other.v1~"} - ) + schema = _schema_entity("gts.x.test._.foo.v1~", {"$schema": "gts.x.other.v1~"}) store = GtsStore(reader=None) store.register(schema) with pytest.raises(ValueError, match="must be a standard JSON Schema URL"): @@ -413,9 +427,7 @@ def test_validate_schema_with_traits_error(self): store.validate_schema("gts.x.test._.foo.v1~") def test_validate_instance_abstract_type_rejected(self): - schema = _schema_entity( - "gts.x.test._.foo.v1~", {"x-gts-abstract": True} - ) + schema = _schema_entity("gts.x.test._.foo.v1~", {"x-gts-abstract": True}) instance = GtsEntity( content={ "$id": "gts.x.test._.foo.v1~x.test._.inst.v1", diff --git a/tests/test_traits.py b/tests/test_traits.py index 72a8342..c75bffe 100644 --- a/tests/test_traits.py +++ b/tests/test_traits.py @@ -1,5 +1,6 @@ """Tests for gts.traits (OP#13 schema traits validation).""" +from gts._json_pointer import resolve from gts.traits import ( build_effective_traits, build_effective_traits_schema, @@ -42,6 +43,20 @@ def test_collect_traits_from_value_merges_allof(self): assert merged == {"b": 2, "a": 1} +class TestJsonPointer: + def test_decodes_uri_fragment_tokens(self): + assert resolve({"a b": "value"}, "#/a%20b") == "value" + + def test_preserves_percent_encoding_in_plain_pointer(self): + assert resolve({"a%20b": "value"}, "/a%20b") == "value" + + def test_rejects_negative_array_index(self): + assert resolve(["value"], "/-1", default="missing") == "missing" + + def test_rejects_leading_zero_array_index(self): + assert resolve(["value"], "/01", default="missing") == "missing" + + class TestInlineLocalPointers: def test_resolves_local_pointer(self): root = {"defs": {"foo": {"type": "string"}}} diff --git a/tests/test_x_gts_ref.py b/tests/test_x_gts_ref.py index 397fad9..ceb78e2 100644 --- a/tests/test_x_gts_ref.py +++ b/tests/test_x_gts_ref.py @@ -170,6 +170,22 @@ def test_object_properties_recursion(self): ) assert len(errors) == 1 + def test_root_ref_traverses_nested_constraints(self): + schema = { + "type": "object", + "properties": { + "link": {"x-gts-ref": "gts.x.test.*"}, + "child": {"$ref": "#"}, + }, + } + + errors = XGtsRefValidator().validate_instance( + {"child": {"link": "gts.x.other.v1~"}}, schema + ) + + assert len(errors) == 1 + assert errors[0].field_path == "child.link" + def test_any_of_no_branch_matched(self): schema = { "anyOf": [