From 739cc55b76755ca621a2d9932c02d39f0b139a4e Mon Sep 17 00:00:00 2001 From: Artifizer Date: Fri, 11 Sep 2026 14:40:39 +0300 Subject: [PATCH 01/17] fix(x-gts-ref): traverse implicit and local reference schemas Apply x-gts-ref validation to type-less object and array schemas, and resolve local JSON Pointer references while guarding against cycles. Signed-off-by: Artifizer --- gts/src/gts/compatibility.py | 30 ++++++++++++++++++++++++--- gts/src/gts/x_gts_ref.py | 39 ++++++++++++++++++++++++++++-------- 2 files changed, 58 insertions(+), 11 deletions(-) diff --git a/gts/src/gts/compatibility.py b/gts/src/gts/compatibility.py index ce5421e..24ccf60 100644 --- a/gts/src/gts/compatibility.py +++ b/gts/src/gts/compatibility.py @@ -132,6 +132,26 @@ 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", "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 +167,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 diff --git a/gts/src/gts/x_gts_ref.py b/gts/src/gts/x_gts_ref.py index 726258b..c5e82cc 100644 --- a/gts/src/gts/x_gts_ref.py +++ b/gts/src/gts/x_gts_ref.py @@ -102,11 +102,37 @@ def validate_instance( """ errors: list[XGtsRefValidationError] = [] - def visit_instance(inst, sch, path, errs): + def resolve_local_ref(ref: str) -> Any | None: + if not ref.startswith("#/"): + return None + current: Any = schema + for part in ref[2:].split("/"): + part = part.replace("~1", "/").replace("~0", "~") + if isinstance(current, dict): + if part not in current: + return None + current = current[part] + elif isinstance(current, list): + try: + current = current[int(part)] + except (ValueError, IndexError): + return None + else: + return None + return current + + 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 +206,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) From 90304b2a57f60840974d4f0ca63d9b4117723d28 Mon Sep 17 00:00:00 2001 From: Artifizer Date: Fri, 11 Sep 2026 22:14:24 +0300 Subject: [PATCH 02/17] fix(validation): enforce standard formats and preserve unknown cast verdicts Enable JSON Schema format checking for instance validation and return unknown compatibility verdicts when casts cross JSON Schema dialects. Signed-off-by: Artifizer --- gts/src/gts/compatibility.py | 8 ++++---- gts/src/gts/schema_cast.py | 8 ++++++++ gts/src/gts/store.py | 6 ++++-- 3 files changed, 16 insertions(+), 6 deletions(-) diff --git a/gts/src/gts/compatibility.py b/gts/src/gts/compatibility.py index 24ccf60..d8171d1 100644 --- a/gts/src/gts/compatibility.py +++ b/gts/src/gts/compatibility.py @@ -137,7 +137,7 @@ def _lower_root_unevaluated_properties(schema: Any) -> Any | None: return schema if any( key in schema - for key in {"$ref", "$dynamicRef", "allOf", "anyOf", "oneOf", "not", "dependentSchemas"} + for key in ("$ref", "$dynamicRef", "allOf", "anyOf", "oneOf", "not", "dependentSchemas") ): return None unevaluated = schema["unevaluatedProperties"] @@ -196,7 +196,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") @@ -210,14 +210,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/schema_cast.py b/gts/src/gts/schema_cast.py index 6b32e90..e9e5faf 100644 --- a/gts/src/gts/schema_cast.py +++ b/gts/src/gts/schema_cast.py @@ -8,6 +8,7 @@ from jsonschema import exceptions as js_exceptions from jsonschema import validate as js_validate +from .compatibility import UNKNOWN, dialects_differ from .gts import GtsID logger = logging.getLogger(__name__) @@ -120,6 +121,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] = [] @@ -151,6 +153,9 @@ def cast( 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 @@ -179,6 +184,9 @@ def cast( 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 diff --git a/gts/src/gts/store.py b/gts/src/gts/store.py index 72edaa3..62bb39d 100644 --- a/gts/src/gts/store.py +++ b/gts/src/gts/store.py @@ -6,7 +6,7 @@ from collections.abc import Iterator from typing import Any -from jsonschema import RefResolver +from jsonschema import FormatChecker, RefResolver from jsonschema.validators import validator_for from referencing import Registry, Resource from referencing.jsonschema import DRAFT202012 @@ -736,7 +736,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=FormatChecker(), ) validator.validate(content) From 8da119a11e34a014ff2a18a301eb8fb4bdd0a706 Mon Sep 17 00:00:00 2001 From: Artifizer Date: Sat, 12 Sep 2026 02:01:21 +0300 Subject: [PATCH 03/17] chore(spec): update gts-spec to v0.13.2 Signed-off-by: Artifizer --- .gts-spec | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.gts-spec b/.gts-spec index 2a171df..0fe1506 160000 --- a/.gts-spec +++ b/.gts-spec @@ -1 +1 @@ -Subproject commit 2a171dff7810657de147e3b5f06f89537d7a4293 +Subproject commit 0fe150696a86a987e95cc2f7775299f04654a57f From 1718708043c7c35ff04c6fcfa5a79b3136a039a3 Mon Sep 17 00:00:00 2001 From: Artifizer Date: Sat, 12 Sep 2026 02:05:03 +0300 Subject: [PATCH 04/17] fix(cast): preserve composed schema constraints Signed-off-by: Artifizer --- gts/src/gts/schema_cast.py | 117 +++++++++++++++++++++++++------------ tests/test_schema_cast.py | 42 +++++++++++-- 2 files changed, 118 insertions(+), 41 deletions(-) diff --git a/gts/src/gts/schema_cast.py b/gts/src/gts/schema_cast.py index e9e5faf..1b80f55 100644 --- a/gts/src/gts/schema_cast.py +++ b/gts/src/gts/schema_cast.py @@ -6,10 +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__) @@ -396,10 +396,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: @@ -434,34 +436,73 @@ def _flatten_property_schema(schema: dict[str, Any]) -> dict[str, Any]: result: dict[str, Any] = {} 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(): + if key in result and result[key] != value: + result = {"allOf": [result, {key: copy.deepcopy(value)}]} + else: + result[key] = copy.deepcopy(value) + for key, value in schema.items(): + if key != "allOf": + if key in result and result[key] != value: + result = {"allOf": [result, {key: copy.deepcopy(value)}]} + else: + result[key] = copy.deepcopy(value) return result + @staticmethod + def _merge_property_schemas(existing: Any, incoming: Any) -> Any: + if existing == incoming: + return copy.deepcopy(existing) + return {"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) + + 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 @@ -649,29 +690,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/tests/test_schema_cast.py b/tests/test_schema_cast.py index 454cf08..a268081 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 @@ -357,8 +359,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 +376,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 +504,28 @@ 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" + 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", From 9ec7f4bea1823b302641717c01e249e337648678 Mon Sep 17 00:00:00 2001 From: Artifizer Date: Sat, 12 Sep 2026 02:05:18 +0300 Subject: [PATCH 05/17] test(validation): cover standard JSON Schema formats Signed-off-by: Artifizer --- tests/test_store.py | 34 +++++++++++++++++++++++++++++++++- 1 file changed, 33 insertions(+), 1 deletion(-) diff --git a/tests/test_store.py b/tests/test_store.py index 778ccdc..4ff4b4e 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,36 @@ 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"], + }, + ) + + store.validate_instance_content( + {"uuid": "550e8400-e29b-41d4-a716-446655440000", "time": "10:30:00"}, + type_id, + ) + with pytest.raises(ValidationError): + store.validate_instance_content( + {"uuid": "not-a-uuid", "time": "10:30:00"}, 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([]) From 7fc865027c7373202b5cdc7db2acbb8091cd0c78 Mon Sep 17 00:00:00 2001 From: Artifizer Date: Sat, 12 Sep 2026 02:06:07 +0300 Subject: [PATCH 06/17] style: format compatibility and store tests Signed-off-by: Artifizer --- gts/src/gts/compatibility.py | 10 +++++++++- 1 file changed, 9 insertions(+), 1 deletion(-) diff --git a/gts/src/gts/compatibility.py b/gts/src/gts/compatibility.py index d8171d1..3b709ae 100644 --- a/gts/src/gts/compatibility.py +++ b/gts/src/gts/compatibility.py @@ -137,7 +137,15 @@ def _lower_root_unevaluated_properties(schema: Any) -> Any | None: return schema if any( key in schema - for key in ("$ref", "$dynamicRef", "allOf", "anyOf", "oneOf", "not", "dependentSchemas") + for key in ( + "$ref", + "$dynamicRef", + "allOf", + "anyOf", + "oneOf", + "not", + "dependentSchemas", + ) ): return None unevaluated = schema["unevaluatedProperties"] From 5ea8bcf16b28a575ee99d303a52553b8a2776d71 Mon Sep 17 00:00:00 2001 From: Artifizer Date: Sat, 12 Sep 2026 02:06:29 +0300 Subject: [PATCH 07/17] fix(validation): bound JSON Schema pattern evaluation Signed-off-by: Artifizer --- gts/pyproject.toml | 1 + gts/src/gts/schema_validation.py | 30 ++++++++++++++++++++ gts/src/gts/store.py | 3 +- gts/src/gts/traits.py | 2 +- tests/test_store_extra.py | 48 ++++++++++++++++++++------------ 5 files changed, 63 insertions(+), 21 deletions(-) create mode 100644 gts/src/gts/schema_validation.py diff --git a/gts/pyproject.toml b/gts/pyproject.toml index bc31a91..8ace9a3 100644 --- a/gts/pyproject.toml +++ b/gts/pyproject.toml @@ -17,6 +17,7 @@ dependencies = [ "fastapi>=0.110,<1", "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/schema_validation.py b/gts/src/gts/schema_validation.py new file mode 100644 index 0000000..f7dece7 --- /dev/null +++ b/gts/src/gts/schema_validation.py @@ -0,0 +1,30 @@ +from __future__ import annotations + +from collections.abc import Iterator +from typing import Any + +import regex +from jsonschema import ValidationError, validators +from jsonschema.validators import validator_for as jsonschema_validator_for + +PATTERN_TIMEOUT_SECONDS = 1.0 + + +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 62bb39d..c1a2f37 100644 --- a/gts/src/gts/store.py +++ b/gts/src/gts/store.py @@ -7,7 +7,6 @@ from typing import Any from jsonschema import FormatChecker, RefResolver -from jsonschema.validators import validator_for from referencing import Registry, Resource from referencing.jsonschema import DRAFT202012 @@ -15,6 +14,7 @@ from .entities import GtsEntity from .gts import GtsID, GtsWildcard from .schema_cast import GtsEntityCastResult +from .schema_validation import validator_for from .x_gts_ref import XGtsRefValidator, _without_x_gts_ref logger = logging.getLogger(__name__) @@ -674,7 +674,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}) diff --git a/gts/src/gts/traits.py b/gts/src/gts/traits.py index 2488680..87b246b 100644 --- a/gts/src/gts/traits.py +++ b/gts/src/gts/traits.py @@ -20,9 +20,9 @@ from typing import Any from jsonschema import Draft7Validator, FormatChecker -from jsonschema.validators import validator_for from . import derivation +from .schema_validation import validator_for from .x_gts_ref import XGtsRefValidator X_GTS_TRAITS_SCHEMA = "x-gts-traits-schema" 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", From 50faf82add8f7718d2a14bd897518b69b410ea55 Mon Sep 17 00:00:00 2001 From: Artifizer Date: Sat, 12 Sep 2026 02:42:28 +0300 Subject: [PATCH 08/17] refactor(ids): centralize type identity parsing Signed-off-by: Artifizer --- gts/src/gts/entities.py | 23 ++++++------------ gts/src/gts/gts.py | 21 +++++++++++++++- gts/src/gts/store.py | 53 ++++++++++++++++------------------------- tests/test_gts_id.py | 53 ++++++++++++++++++++++++++++++++++++----- 4 files changed, 95 insertions(+), 55 deletions(-) diff --git a/gts/src/gts/entities.py b/gts/src/gts/entities.py index 171e68e..e13a761 100644 --- a/gts/src/gts/entities.py +++ b/gts/src/gts/entities.py @@ -348,12 +348,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 +366,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 +383,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..49d87e1 100644 --- a/gts/src/gts/gts.py +++ b/gts/src/gts/gts.py @@ -282,7 +282,26 @@ def __init__(self, id: str): @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: + normalized = value.strip().removeprefix(GTS_URI_PREFIX) + if not normalized.endswith("~"): + raise GtsInvalidId(value, "must end with '~'") + return cls(value) def get_type_id(self) -> str | None: if len(self.gts_id_segments) < 2: diff --git a/gts/src/gts/store.py b/gts/src/gts/store.py index c1a2f37..a6a90dc 100644 --- a/gts/src/gts/store.py +++ b/gts/src/gts/store.py @@ -20,6 +20,13 @@ 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,12 +166,9 @@ 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: """ @@ -292,17 +296,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] @@ -613,15 +615,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): @@ -652,9 +652,7 @@ 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 ( @@ -697,14 +695,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: @@ -719,9 +710,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: 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) From 706366a1bd40f8ad060a724931ecc76413ca87d3 Mon Sep 17 00:00:00 2001 From: Artifizer Date: Sat, 12 Sep 2026 10:30:23 +0300 Subject: [PATCH 09/17] refactor(naming): centralize GTS scheme, marker and JSON-pointer handling GTS identifier plumbing (the gts:// URI scheme, the bare gts. prefix and the ~ type marker) was duplicated across the library: every consumer defensively stripped the scheme, and JSON Pointer resolution (with its ~0/~1 escaping) was reimplemented three times. This spread low-level string logic across modules and, worse, onto the public GtsID surface. Move the primitives into a private _naming module and RFC 6901 pointer resolution into a private _json_pointer module, both internal-only. Introduce an internal GtsRef classifier so $ref handling (local pointer vs gts:// vs other) lives in one place. Normalize the scheme once inside GtsStore.get() so callers pass ids or URIs interchangeably without stripping. Keep GtsID's public surface minimal: only meaningful members (.id, .uri, .is_type, is_valid, parse_type), no scheme string helpers. No behavior change; full test suite green. Signed-off-by: Artifizer --- gts/src/gts/_json_pointer.py | 63 ++++++++++++++++++++++ gts/src/gts/_json_validation.py | 6 +-- gts/src/gts/_naming.py | 62 ++++++++++++++++++++++ gts/src/gts/entities.py | 17 +++--- gts/src/gts/gts.py | 94 ++++++++++++++++++++++++++++----- gts/src/gts/ops.py | 3 +- gts/src/gts/store.py | 53 ++++++++++--------- gts/src/gts/traits.py | 21 +------- gts/src/gts/x_gts_ref.py | 58 ++++++-------------- 9 files changed, 263 insertions(+), 114 deletions(-) create mode 100644 gts/src/gts/_json_pointer.py create mode 100644 gts/src/gts/_naming.py diff --git a/gts/src/gts/_json_pointer.py b/gts/src/gts/_json_pointer.py new file mode 100644 index 0000000..7859b52 --- /dev/null +++ b/gts/src/gts/_json_pointer.py @@ -0,0 +1,63 @@ +"""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 + +# 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). + """ + pointer = pointer.removeprefix("#") + 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): + try: + current = current[int(token)] + except (ValueError, 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/entities.py b/gts/src/gts/entities.py index e13a761..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. diff --git a/gts/src/gts/gts.py b/gts/src/gts/gts.py index 49d87e1..7849bad 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,6 +295,11 @@ 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.gts_id_segments[-1].is_type @@ -298,8 +318,7 @@ def parent_type_id(self) -> str | None: @classmethod def parse_type(cls, value: str) -> GtsID: - normalized = value.strip().removeprefix(GTS_URI_PREFIX) - if not normalized.endswith("~"): + if not _is_type_ref(value.strip()): raise GtsInvalidId(value, "must end with '~'") return cls(value) @@ -316,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) @@ -470,3 +486,53 @@ 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/store.py b/gts/src/gts/store.py index a6a90dc..6eead47 100644 --- a/gts/src/gts/store.py +++ b/gts/src/gts/store.py @@ -11,8 +11,9 @@ 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 validator_for from .x_gts_ref import XGtsRefValidator, _without_x_gts_ref @@ -175,7 +176,12 @@ 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] @@ -200,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 = {} @@ -228,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): @@ -258,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 '#') " @@ -491,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. @@ -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" @@ -658,7 +663,7 @@ def validate_schema_content( 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" diff --git a/gts/src/gts/traits.py b/gts/src/gts/traits.py index 87b246b..506df56 100644 --- a/gts/src/gts/traits.py +++ b/gts/src/gts/traits.py @@ -22,6 +22,7 @@ from jsonschema import Draft7Validator, FormatChecker from . import derivation +from ._json_pointer import resolve as resolve_json_pointer from .schema_validation import validator_for from .x_gts_ref import XGtsRefValidator @@ -120,7 +121,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 +137,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 c5e82cc..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: @@ -103,23 +106,9 @@ def validate_instance( errors: list[XGtsRefValidationError] = [] def resolve_local_ref(ref: str) -> Any | None: - if not ref.startswith("#/"): + if ref != "#" and not ref.startswith("#/"): return None - current: Any = schema - for part in ref[2:].split("/"): - part = part.replace("~1", "/").replace("~0", "~") - if isinstance(current, dict): - if part not in current: - return None - current = current[part] - elif isinstance(current, list): - try: - current = current[int(part)] - except (ValueError, IndexError): - return None - else: - return None - return current + 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.""" @@ -346,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 @@ -385,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, @@ -459,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: @@ -501,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 From 624b86aeb88c7ca007523c636b7551aab7cd2fd7 Mon Sep 17 00:00:00 2001 From: Artifizer Date: Sat, 12 Sep 2026 10:39:16 +0300 Subject: [PATCH 10/17] fix(cast): intersect composed schema constraints and report unknown compat on dialect change When flattening allOf-composed schemas for casting, repeated constraints were wrapped in a nested allOf instead of being intersected, so tightened bounds (minLength, minimum, enum, type, additionalProperties, ...) never surfaced at the top level. Introduce constraint intersection and apply it consistently to top-level keywords, object properties, and repeated same-name property schemas. Also report backward/forward/full compatibility as "unknown" (tri-state) rather than a hard false when the JSON Schema dialect changes across the cast, since the verdict cannot be established. Finally, stop lowering unevaluatedProperties across conditional applicators (if/then/else) in compatibility sanitization, and cover root self-reference ($ref: "#") traversal in x-gts-ref validation. Signed-off-by: Artifizer --- gts/src/gts/compatibility.py | 3 ++ gts/src/gts/schema_cast.py | 87 ++++++++++++++++++++++++++++-------- tests/test_compatibility.py | 12 +++++ tests/test_schema_cast.py | 43 +++++++++++++++++- tests/test_x_gts_ref.py | 16 +++++++ 5 files changed, 140 insertions(+), 21 deletions(-) diff --git a/gts/src/gts/compatibility.py b/gts/src/gts/compatibility.py index 3b709ae..76a1270 100644 --- a/gts/src/gts/compatibility.py +++ b/gts/src/gts/compatibility.py @@ -143,6 +143,9 @@ def _lower_root_unevaluated_properties(schema: Any) -> Any | None: "allOf", "anyOf", "oneOf", + "if", + "then", + "else", "not", "dependentSchemas", ) diff --git a/gts/src/gts/schema_cast.py b/gts/src/gts/schema_cast.py index 1b80f55..b539e90 100644 --- a/gts/src/gts/schema_cast.py +++ b/gts/src/gts/schema_cast.py @@ -26,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 @@ -56,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) @@ -146,9 +148,9 @@ 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, @@ -177,9 +179,9 @@ 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, @@ -431,31 +433,70 @@ 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): for key, value in GtsEntityCastResult._flatten_property_schema( sub_schema ).items(): - if key in result and result[key] != value: - result = {"allOf": [result, {key: copy.deepcopy(value)}]} - else: - result[key] = copy.deepcopy(value) + merge(key, value) for key, value in schema.items(): if key != "allOf": - if key in result and result[key] != value: - result = {"allOf": [result, {key: copy.deepcopy(value)}]} - else: - result[key] = copy.deepcopy(value) + merge(key, value) return result @staticmethod def _merge_property_schemas(existing: Any, incoming: Any) -> Any: if existing == incoming: return copy.deepcopy(existing) - return {"allOf": [copy.deepcopy(existing), copy.deepcopy(incoming)]} + # 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]: @@ -487,6 +528,14 @@ def _flatten_schema(schema: dict[str, Any]) -> dict[str, Any]: 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): 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_schema_cast.py b/tests/test_schema_cast.py index a268081..f0a9451 100644 --- a/tests/test_schema_cast.py +++ b/tests/test_schema_cast.py @@ -116,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: @@ -295,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} @@ -525,6 +561,9 @@ def test_cast_across_distinct_dialects_has_unknown_compatibility(self): 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( 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": [ From ef5ad21e7c2fcdbfb4a4ef4624679204b0375ea3 Mon Sep 17 00:00:00 2001 From: Artifizer Date: Sat, 12 Sep 2026 11:50:35 +0300 Subject: [PATCH 11/17] fix(validation): enforce correct RFC 3339 time/date-time formats Instance and trait validation constructed the jsonschema validator with a bare FormatChecker(), which draws from jsonschema's shared class-level checker registry where the draft-3 "time" checker (HH:MM:SS, no timezone) overwrites the draft6/7 successors. As a result valid RFC 3339 values such as "10:30:00Z" were rejected while invalid UUIDs slipped through. Introduce a shared FORMAT_CHECKER in schema_validation that combines the bare checker (for "uuid", added only in draft 2019-09) with the draft-07 checkers (for correct RFC 3339 "time"/"date-time"), using only built-in jsonschema checkers. Use it in store and reuse it from traits to drop the duplicated construction. Update the store test's baseline to a timezone-bearing time value, since RFC 3339 "time" in draft-07 requires an offset. Signed-off-by: Artifizer --- gts/src/gts/schema_validation.py | 17 ++++++++++++++++- gts/src/gts/store.py | 6 +++--- gts/src/gts/traits.py | 5 +---- tests/test_store.py | 6 ++++-- 4 files changed, 24 insertions(+), 10 deletions(-) diff --git a/gts/src/gts/schema_validation.py b/gts/src/gts/schema_validation.py index f7dece7..95b61b7 100644 --- a/gts/src/gts/schema_validation.py +++ b/gts/src/gts/schema_validation.py @@ -4,11 +4,26 @@ from typing import Any import regex -from jsonschema import ValidationError, validators +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 diff --git a/gts/src/gts/store.py b/gts/src/gts/store.py index 6eead47..1a489b4 100644 --- a/gts/src/gts/store.py +++ b/gts/src/gts/store.py @@ -6,7 +6,7 @@ from collections.abc import Iterator from typing import Any -from jsonschema import FormatChecker, RefResolver +from jsonschema import RefResolver from referencing import Registry, Resource from referencing.jsonschema import DRAFT202012 @@ -15,7 +15,7 @@ from .entities import GtsEntity from .gts import GtsID, GtsRef, GtsWildcard from .schema_cast import GtsEntityCastResult -from .schema_validation import validator_for +from .schema_validation import FORMAT_CHECKER, validator_for from .x_gts_ref import XGtsRefValidator, _without_x_gts_ref logger = logging.getLogger(__name__) @@ -731,7 +731,7 @@ def validate_instance_content(self, content: dict[str, Any], type_id: str) -> No validator = validator_class( schema_for_validation, registry=self._create_reference_registry(), - format_checker=FormatChecker(), + format_checker=FORMAT_CHECKER, ) validator.validate(content) diff --git a/gts/src/gts/traits.py b/gts/src/gts/traits.py index 506df56..7d7f954 100644 --- a/gts/src/gts/traits.py +++ b/gts/src/gts/traits.py @@ -19,10 +19,9 @@ import copy from typing import Any -from jsonschema import Draft7Validator, FormatChecker - 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 @@ -30,8 +29,6 @@ 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: diff --git a/tests/test_store.py b/tests/test_store.py index 4ff4b4e..d8ea42c 100644 --- a/tests/test_store.py +++ b/tests/test_store.py @@ -316,13 +316,15 @@ def test_validate_instance_content_enforces_standard_formats(self): }, ) + # 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:00"}, + {"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:00"}, type_id + {"uuid": "not-a-uuid", "time": "10:30:00Z"}, type_id ) with pytest.raises(ValidationError): store.validate_instance_content( From f4525c9e66f87e1b52305b316c5293596ff63406 Mon Sep 17 00:00:00 2001 From: Artifizer Date: Sat, 12 Sep 2026 12:10:33 +0300 Subject: [PATCH 12/17] chore(release): bump version to 0.13.2 Patch release covering the recent bug fixes and internal refactors since 0.13.1 (RFC 3339 format enforcement, cast constraint intersection, naming and id parsing centralization). No new features or public API changes. Signed-off-by: Artifizer --- gts/pyproject.toml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/gts/pyproject.toml b/gts/pyproject.toml index 8ace9a3..26aa5d7 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" }] From 3d9c6c85c8d8abce3254975ed55c383745aae07a Mon Sep 17 00:00:00 2001 From: Artifizer Date: Sat, 12 Sep 2026 12:21:39 +0300 Subject: [PATCH 13/17] fix(deps): require pydantic v2 for the server venv The GTS server uses pydantic v2 APIs (model_validator) but only pulled pydantic transitively via FastAPI. The spec test-client requirements pin pydantic<2 and share the venv, so refreshing py-env could downgrade pydantic to v1 and break the FastAPI import (IncEx from pydantic.main). Declare pydantic>=2,<3 directly in the package, and repurpose the unused root requirements.txt into the local dev/test tooling file (pydantic override, ruff, mypy) installed after the spec requirements so v2 wins. Wire py-env to install from it instead of hardcoding the packages. Signed-off-by: Artifizer --- Makefile | 6 ++++-- gts/pyproject.toml | 3 +++ requirements.txt | 20 ++++++++++++++++---- 3 files changed, 23 insertions(+), 6 deletions(-) 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 26aa5d7..da05785 100644 --- a/gts/pyproject.toml +++ b/gts/pyproject.toml @@ -15,6 +15,9 @@ 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", 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 From fa749afad060fea1046e938661ca4fa90070b5b9 Mon Sep 17 00:00:00 2001 From: Artifizer Date: Sat, 12 Sep 2026 12:25:24 +0300 Subject: [PATCH 14/17] style: apply ruff formatting to gts and schema_cast Signed-off-by: Artifizer --- gts/src/gts/gts.py | 4 +--- gts/src/gts/schema_cast.py | 24 ++++++++++++++++++++---- 2 files changed, 21 insertions(+), 7 deletions(-) diff --git a/gts/src/gts/gts.py b/gts/src/gts/gts.py index 7849bad..6a21b00 100644 --- a/gts/src/gts/gts.py +++ b/gts/src/gts/gts.py @@ -508,9 +508,7 @@ class GtsRef: GTS = "gts" OTHER = "other" - def __init__( - self, raw: str, kind: str, target_id: str, has_scheme: bool - ) -> None: + 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 diff --git a/gts/src/gts/schema_cast.py b/gts/src/gts/schema_cast.py index b539e90..b197d9c 100644 --- a/gts/src/gts/schema_cast.py +++ b/gts/src/gts/schema_cast.py @@ -437,16 +437,30 @@ def _remove_gts_const_constraints(schema: Any) -> Any: 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"}: + if key in { + "minimum", + "exclusiveMinimum", + "minLength", + "minItems", + "minProperties", + }: return max(existing, incoming) - if key in {"maximum", "exclusiveMaximum", "maxLength", "maxItems", "maxProperties"}: + 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] + 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: @@ -533,7 +547,9 @@ def _flatten_schema(schema: dict[str, Any]) -> dict[str, Any]: key, result[key], value ) if merged is None: - result.setdefault("allOf", []).append({key: copy.deepcopy(value)}) + result.setdefault("allOf", []).append( + {key: copy.deepcopy(value)} + ) else: result[key] = merged From 9ddfcd8e1ca915b867c8e38a6f614d7bf82d7ddb Mon Sep 17 00:00:00 2001 From: Artifizer Date: Sat, 12 Sep 2026 12:42:42 +0300 Subject: [PATCH 15/17] fix: enforce JSON Pointer syntax Signed-off-by: Artifizer --- gts/src/gts/_json_pointer.py | 11 +++++++++-- tests/test_traits.py | 12 ++++++++++++ 2 files changed, 21 insertions(+), 2 deletions(-) diff --git a/gts/src/gts/_json_pointer.py b/gts/src/gts/_json_pointer.py index 7859b52..9ef3e49 100644 --- a/gts/src/gts/_json_pointer.py +++ b/gts/src/gts/_json_pointer.py @@ -16,6 +16,7 @@ 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 @@ -40,7 +41,7 @@ def resolve(document: Any, pointer: str, default: Any = None) -> Any: Returns ``default`` if any reference token cannot be resolved (missing key, non-integer/out-of-range array index, or descending into a scalar). """ - pointer = pointer.removeprefix("#") + pointer = unquote(pointer.removeprefix("#")) if pointer == "": return document if not pointer.startswith("/"): @@ -54,9 +55,15 @@ def resolve(document: Any, pointer: str, default: Any = None) -> Any: 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 (ValueError, IndexError): + except IndexError: return default else: return default diff --git a/tests/test_traits.py b/tests/test_traits.py index 72a8342..778c3ea 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,17 @@ 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_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"}}} From 5b5380bf47c1aebdae5e0dd2d38466f49ef89867 Mon Sep 17 00:00:00 2001 From: Artifizer Date: Sun, 13 Sep 2026 13:58:12 +0300 Subject: [PATCH 16/17] chore: update gts-spec version to 0.13.3 Signed-off-by: Artifizer --- .gts-spec | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.gts-spec b/.gts-spec index 0fe1506..5b5d578 160000 --- a/.gts-spec +++ b/.gts-spec @@ -1 +1 @@ -Subproject commit 0fe150696a86a987e95cc2f7775299f04654a57f +Subproject commit 5b5d5786bab1de1192d977a94752e492b66d17a6 From d001f0fa0dfc6675276b7a6d9de77351aa84c368 Mon Sep 17 00:00:00 2001 From: Artifizer Date: Sun, 13 Sep 2026 14:14:37 +0300 Subject: [PATCH 17/17] fix: preserve plain JSON Pointer escapes Signed-off-by: Artifizer --- gts/src/gts/_json_pointer.py | 3 ++- tests/test_traits.py | 3 +++ 2 files changed, 5 insertions(+), 1 deletion(-) diff --git a/gts/src/gts/_json_pointer.py b/gts/src/gts/_json_pointer.py index 9ef3e49..15aa810 100644 --- a/gts/src/gts/_json_pointer.py +++ b/gts/src/gts/_json_pointer.py @@ -41,7 +41,8 @@ def resolve(document: Any, pointer: str, default: Any = None) -> Any: Returns ``default`` if any reference token cannot be resolved (missing key, non-integer/out-of-range array index, or descending into a scalar). """ - pointer = unquote(pointer.removeprefix("#")) + if pointer.startswith("#"): + pointer = unquote(pointer[1:]) if pointer == "": return document if not pointer.startswith("/"): diff --git a/tests/test_traits.py b/tests/test_traits.py index 778c3ea..c75bffe 100644 --- a/tests/test_traits.py +++ b/tests/test_traits.py @@ -47,6 +47,9 @@ 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"