diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index b239d83..90f807d 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -162,9 +162,15 @@ jobs: - name: Install dependencies run: | python -m pip install --upgrade pip - # Install httprunner first to get pydantic<1.9 constraint - pip install httprunner pytest requests - # Then install gts (will use already-installed pydantic v1) + # The gts-spec test runner requires pytest >= 9 (see + # .gts-spec/tests/pytest.ini). httprunner pins incompatible deps + # (pytest < 8, pydantic < 1.9), so install it with --no-deps and take + # the rest from the spec's requirements, mirroring the Makefile's + # py-env target. Installing gts then pins pydantic to v2. + grep -viE '^[[:space:]]*httprunner([^A-Za-z0-9_.-]|$)' \ + .gts-spec/tests/requirements.txt > /tmp/spec-requirements.txt + pip install -r /tmp/spec-requirements.txt + pip install --no-deps 'httprunner>=4,<5' pip install -e ./gts - name: Run e2e tests diff --git a/.gts-spec b/.gts-spec index 860f6e2..4f0cb12 160000 --- a/.gts-spec +++ b/.gts-spec @@ -1 +1 @@ -Subproject commit 860f6e254c4bf17f9998380532b1a9e3a9d459f9 +Subproject commit 4f0cb121cd1b654c6c4d84a332a4d7fc84fa1202 diff --git a/.gts-spec-version b/.gts-spec-version index 86dd09a..6b60281 100644 --- a/.gts-spec-version +++ b/.gts-spec-version @@ -1 +1 @@ -v0.15.0 +v0.17.0 diff --git a/Makefile b/Makefile index 0c027d7..c1901b7 100644 --- a/Makefile +++ b/Makefile @@ -54,6 +54,21 @@ endif force-py-env: +# Recreate the venv when the bootstrap interpreter's Python version no longer +# matches the existing venv (e.g. the system python3 was upgraded, or the venv +# was created by a different python3). Silently reusing a venv built with an +# incompatible interpreter causes confusing failures: google-re2 ships wheels +# only for 3.10-3.14, so a stale 3.9 venv tries to build it from source and +# fails on missing RE2/Abseil headers. +PY_ENV_VERSION := $(shell "$(PY_ENV_PYTHON)" -c 'import sys;print("%d.%d"%sys.version_info[:2])' 2>/dev/null) +PY_BOOTSTRAP_VERSION := $(shell $(PYTHON_BOOTSTRAP) -c 'import sys;print("%d.%d"%sys.version_info[:2])' 2>/dev/null) +ifneq ($(PY_ENV_VERSION),) +ifneq ($(PY_ENV_VERSION),$(PY_BOOTSTRAP_VERSION)) +$(PY_ENV_PYTHON): force-py-env +$(PY_ENV_STAMP): force-py-env +endif +endif + $(PY_ENV_STAMP): $(PY_ENV_PYTHON) gts/pyproject.toml $(GTS_SPEC_DIR)/tests/requirements.txt requirements.txt Makefile @echo "Creating/updating Python virtual environment in $(PY_ENV_DIR)..." @rm -f "$@" diff --git a/README.md b/README.md index bef6c54..ee58f73 100644 --- a/README.md +++ b/README.md @@ -2,9 +2,9 @@ # GTS Python Library -A minimal, idiomatic Python library for working with **GTS** ([Global Type System](https://github.com/gts-spec/gts-spec)) identifiers and JSON/JSON Schema artifacts. +A minimal, idiomatic Python library for working with **GTS** ([Global Type System](https://github.com/GlobalTypeSystem/gts-spec)) identifiers and JSON/JSON Schema artifacts. -Current supported GTS spec version: `0.15.0`. Conformance CI pins the released gts-spec `v0.15.0` in [`.gts-spec-version`](.gts-spec-version) and the `.gts-spec` submodule; run `make update-spec` to check the pinned release out. The 0.15 breaking change is the [safe regular-expression profile](gts/README.md#regular-expressions). +Targets gts-spec [v0.17.0](https://github.com/GlobalTypeSystem/gts-spec/releases/tag/v0.17.0) — recorded in [`.gts-spec-version`](.gts-spec-version) and pinned by the `.gts-spec` submodule. Run `make update-spec` to check the pinned release out. ## Roadmap @@ -29,14 +29,16 @@ Featureset: See details in [gts/README.md](gts/README.md) -Other GTS spec [Reference Implementation](https://github.com/globaltypesystem/gts-spec/blob/main/README.md#9-reference-implementation-recommendations) recommended features support: +Other GTS spec [Reference Implementation](https://github.com/GlobalTypeSystem/gts-spec/blob/main/README.md#9-reference-implementation-recommendations) recommended features support: - [x] **In-memory entities registry** - simple GTS entities registry with optional GTS references validation on entity registration - [x] **CLI** - command-line interface for all GTS operations - [x] **Web server** - a non-production web-server with REST API for the operations processing and testing -- [x] **x-gts-ref support** - to support special GTS entity reference annotation in schemas +- [x] **x-gts-type-ref / x-gts-instance-ref** - kind-specific GTS entity references in schemas (spec §9.6.1) +- [x] **x-gts-pattern-ref** - pattern-valued string fields: GTS identifier pattern syntax and coverage by an operand, independent of the registry (spec §9.6.2, §10) +- [x] **x-gts-final / x-gts-abstract** - GTS Type Schema modifiers controlling inheritance and instantiation - [x] **YAML support** - to support YAML files (*.yml, *.yaml) as input files -- [x] **UUID for instances** - to support UUID as ID in JSON instances +- [x] **UUID for instances** - to support UUID as ID in JSON instances (anonymous instances) - [ ] **TypeSpec support** - direct support for [typespec.io](https://typespec.io/) files (*.tsp) input files Technical Backlog: @@ -49,7 +51,7 @@ Technical Backlog: ## Installation -GTS requires Python 3.9 or later. +GTS requires Python 3.9 or later. `google-re2` ships wheels for CPython 3.10–3.14; on Python 3.9 pip builds it from source, which needs a C++17 compiler and the RE2 and Abseil development headers. ### Local development @@ -82,39 +84,133 @@ pip install gts ### CLI ```bash -gts - -# see available commands +# Basic operations +gts validate-id --gts-id 'gts.example.demo._.event.v1~' +gts parse-id --gts-id 'gts.example.demo._.event.v1~' +gts match-id-pattern --candidate 'gts.example.demo._.event.v1~' --pattern 'gts.example.*' +gts uuid --gts-id 'gts.example.demo._.event.v1~' + +# Operations with loaded entities. +# Placeholders: point --path at a directory holding your GTS schemas and +# instances, and substitute their IDs (no ./examples directory ships with +# this repository; the gts-spec submodule's .gts-spec/examples has samples). +gts --path ./examples validate-instance --gts-id 'gts.example.demo._.event.v1~example.demo._.created.v1' +gts --path ./examples validate-type-schema --gts-id 'gts.example.demo._.event.v1.1~' +gts --path ./examples validate-entity --gts-id 'gts.example.demo._.event.v1~' +gts --path ./examples resolve-relationships --gts-id 'gts.example.demo._.event.v1~' +gts --path ./examples compatibility --old-schema-id 'gts.example.demo._.event.v1.0~' --new-schema-id 'gts.example.demo._.event.v1.1~' +gts --path ./examples cast --from-id 'gts.example.demo._.event.v1~example.demo._.created.v1' --to-schema-id 'gts.example.demo._.event.v1.1~' +gts --path ./examples query --expr 'gts.example.*' --limit 10 +gts --path ./examples attr --gts-with-path 'gts.example.demo._.event.v1~@properties.name' +gts --path ./examples list --limit 100 + +# See all commands and options gts --help ``` ### Library -See [gts/README.md](gts/README.md) +`GtsOps` provides the high-level in-memory API: -### Web server +```python +from gts.ops import GtsOps + +ops = GtsOps() + +# OP#1 - ID Validation +result = ops.validate_id("gts.example.demo._.event.v1~") +assert result.valid + +# OP#2 - ID Extraction +extracted = ops.extract_id({"id": "gts.example.demo._.event.v1.0", "name": "Event"}) +print(extracted.id) + +# OP#3 - ID Parsing +parsed = ops.parse_id("gts.example.demo._.event.v1~") +for seg in parsed.segments: + print(seg.vendor, seg.package, seg.type, seg.ver_major) + +# OP#4 - Pattern Matching +match = ops.match_id_pattern("gts.example.demo._.event.v1.0", "gts.example.*") +assert match.match + +# OP#5 - ID to UUID Mapping +uuid_res = ops.uuid("gts.example.demo._.event.v1~") +print(uuid_res.uuid) +``` + +See [gts/README.md](gts/README.md) for full library documentation including schema validation, store operations, query execution, traits, and casting. -The web server is a non-production web-server with REST API for the operations processing and testing. It implements reference API for gts-spec [tests](https://github.com/GlobalTypeSystem/gts-spec/tree/main/tests) +### Web server +The web server is a non-production web server with REST API for operations processing and testing. It implements the reference API for [gts-spec tests](https://github.com/GlobalTypeSystem/gts-spec/tree/main/tests). ```bash -# start the web server, default location is http://127.0.0.1:8000 -gts serve +# Start the web server (default: http://127.0.0.1:8000) +gts server -# start the web server on different port +# Start the web server on a different port gts server --port 8081 -# pre-populate server with the JSON instancens and schemas from the gts-spec tests -gts server --path {PATH_TO_JSON_FILES} +# Pre-populate server with entities from files or directories +gts --path {PATH_TO_JSON_FILES} server + +# Allow re-registering an entity with different content (default: reject with 409 Conflict) +gts server --allow-entity-updates -# Generate the OpenAPI schema -curl -s http://127.0.0.1:8000/openapi.json | jq > ./gts/openapi.json +# Generate the OpenAPI specification +gts openapi-spec --out openapi.json -# See the schema +# Or fetch from running server +curl -s http://127.0.0.1:8000/openapi.json | jq > ./openapi.json curl -s http://127.0.0.1:8000/openapi.json | jq | less -S ``` -### Testing +#### API Endpoints + +##### Entity Management +- `GET /entities` - List registered entities (`limit` query parameter, 1–1000) +- `GET /entities/{gts_id}` - Retrieve a specific entity +- `POST /entities` - Register a single entity (optional `?validate=true`) +- `POST /entities/bulk` - Register multiple entities +- `POST /type-schemas` - Register a batch of GTS Type Schemas (each `type_id` derived from its embedded `$id`) + +##### GTS Operations +- `GET /validate-id?gts_id=` - Validate GTS ID (OP#1) +- `POST /extract-id` - Extract GTS ID from JSON (OP#2) +- `GET /parse-id?gts_id=` - Parse GTS ID (OP#3) +- `GET /match-id-pattern?candidate=&pattern=` - Match pattern (OP#4) +- `GET /uuid?gts_id=` - Generate UUID (OP#5) +- `POST /validate-instance` - Validate registered instance (OP#6) +- `POST /validate-json` - Validate transient instance or Type Schema JSON without registering (OP#6) +- `POST /validate-json/{gts_type}` - Validate transient instance JSON against explicit type (OP#6) +- `GET /resolve-relationships?gts_id=` - Resolve relationships and detect broken references (OP#7) +- `GET /compatibility?old_type_id=&new_type_id=` - Check Type Schema evolution compatibility (OP#8) +- `POST /cast` - Cast instance to target schema (OP#9) +- `GET /query?expr=&limit=` - Query entities (OP#10) +- `GET /attr?gts_with_path=` - Retrieve attribute value (OP#11) +- `POST /validate-type-schema` - Validate derived Type Schema against base chain (OP#12) +- `POST /validate-entity` - Validate entity by GTS ID (OP#12/OP#13) + +## Development + +Common development tasks are wrapped in the `Makefile`: + +```bash +make py-env # create/update virtualenv and install dependencies +make install # install gts in editable mode +make build # build source and wheel distributions into dist/ +make check # run all quality checks (fmt, lint, mypy, test, e2e) +make test # run unit tests +make coverage # measure code coverage +make fmt # check code formatting (ruff) +make dev-fmt # auto-format code (ruff) +make lint # run linter (ruff) +make mypy # run type checker +make security # check dependencies for vulnerabilities (pip-audit) +``` + +## Testing You can test the gts-python library by utilizing the shared test suite from the [gts-spec](https://github.com/GlobalTypeSystem/gts-spec) specification and executing the tests against the web server. diff --git a/gts/README.md b/gts/README.md index 9691c58..3b3faa2 100644 --- a/gts/README.md +++ b/gts/README.md @@ -2,7 +2,7 @@ Python helpers and a reference HTTP service for the [Global Type System (GTS)](https://github.com/globaltypesystem/gts-spec). The package supports GTS identifier parsing, JSON Schema-backed validation, schema compatibility and derivation checks, traits, casting, queries, file loading, a CLI, and a FastAPI application. -The package targets GTS specification 0.15.0 and requires Python 3.9 or later. `google-re2` ships wheels for CPython 3.10-3.14; on Python 3.9 pip builds it from source, which needs a C++17 compiler and the RE2 and Abseil development headers. +The package targets GTS specification 0.17.0 and requires Python 3.9 or later. `google-re2` ships wheels for CPython 3.10-3.14; on Python 3.9 pip builds it from source, which needs a C++17 compiler and the RE2 and Abseil development headers. ## Installation @@ -166,8 +166,8 @@ TypeSpec (`.tsp`) inputs must be compiled to JSON Schema before loading. | `get(entity_id)` | Return `GtsEntity` or `None`. | | `items()` | Return an iterator over in-memory `(entity_id, entity)` pairs. | | `get_schema_content(type_id)` | Return a schema dictionary or raise `KeyError`. | -| `validate_schema_basic(type_id)` | Check `$ref` format, `x-gts-ref` declarations, and GTS keyword placement. | -| `validate_schema(type_id)` | Run full JSON Schema, derivation, final/abstract, x-gts-ref, and trait validation. | +| `validate_schema_basic(type_id)` | Check `$ref` format, GTS reference declarations, and GTS keyword placement. | +| `validate_schema(type_id)` | Run full JSON Schema, derivation, final/abstract, reference, and trait validation. | | `validate_instance(gts_id)` | Validate a well-known or UUID-addressed instance against its type schema. | | `is_minor_compatible(old_schema_id, new_schema_id)` | Return compatibility verdicts for two registered schemas. | | `cast(from_id, target_schema_id)` | Cast a registered **instance** to a target schema. | @@ -215,11 +215,15 @@ The library validates schemas with the dialect named by `$schema`; if absent, it During instance validation, GTS references are resolved with `referencing.Registry`. Draft 2019-09 and Draft 2020-12 `$ref` sibling constraints are preserved, so a sibling such as `minLength` is enforced. -### `x-gts-ref` +### GTS references -`x-gts-ref` restricts a string value to a GTS ID or pattern. It accepts an absolute `gts.` pattern or a JSON Pointer beginning with `/` that resolves to one. If a store is present, the referenced entity must be registered. +GTS 0.17 replaces the former `x-gts-ref` with kind-specific entity references `x-gts-type-ref` and `x-gts-instance-ref`, and 0.16 adds `x-gts-pattern-ref`: -`x-gts-ref` can appear in `oneOf`, `anyOf`, and `allOf`. For x-gts-ref-only combinator branches, GTS evaluates the x-gts-ref constraints as the combinator condition. For ordinary or mixed JSON Schema branches, normal JSON Schema structural matching determines which branch’s x-gts-ref constraints are applied. +- `x-gts-type-ref`: restricts a string value to a GTS Type Identifier. +- `x-gts-instance-ref`: restricts a string value to a GTS Instance Identifier. +- `x-gts-pattern-ref`: restricts a string value to a GTS identifier pattern covered by the operand (`M(value) ⊆ M(operand)`), evaluated without consulting the registry. + +The former kind-blind `x-gts-ref` keyword was removed in GTS 0.16 and is rejected on schema validation like any unknown `x-gts-*` keyword. ### `x-gts-final` and `x-gts-abstract` diff --git a/gts/pyproject.toml b/gts/pyproject.toml index 272ed64..8abc75a 100644 --- a/gts/pyproject.toml +++ b/gts/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "gts" -version = "0.16.0" +version = "0.17.0" description = "Global Type System (GTS) helpers: identifiers, parsing, validation, and operations" readme = "README.md" authors = [{ name = "GTS Community" }] diff --git a/gts/src/gts/_json_validation.py b/gts/src/gts/_json_validation.py index b1bb05c..ccc2608 100644 --- a/gts/src/gts/_json_validation.py +++ b/gts/src/gts/_json_validation.py @@ -13,7 +13,9 @@ from .gts import GtsID from .store import GtsStore -_X_GTS_REF_KEYWORD = "x-gts-ref" +# Any GTS extension keyword (x-gts-type-ref, x-gts-pattern-ref, x-gts-final, ...) +# marks a file as GTS-related even when it carries no gts. identifier. +_X_GTS_KEYWORD_PREFIX = "x-gts-" @dataclass @@ -131,7 +133,9 @@ def _on_walk_error(err: OSError) -> None: @staticmethod def _is_gts_marker(text: str) -> bool: return ( - GTS_PREFIX in text or GTS_URI_PREFIX in text or _X_GTS_REF_KEYWORD in text + GTS_PREFIX in text + or GTS_URI_PREFIX in text + or _X_GTS_KEYWORD_PREFIX in text ) def _read_file(self, file_path: Path) -> None: diff --git a/gts/src/gts/compatibility.py b/gts/src/gts/compatibility.py index 5e8d428..2133cdc 100644 --- a/gts/src/gts/compatibility.py +++ b/gts/src/gts/compatibility.py @@ -199,6 +199,17 @@ def _is_subschema(subset: Any, superset: Any) -> bool | None: lowered_superset = _lower_root_unevaluated_properties(superset) if lowered_subset is None or lowered_superset is None: return None + # Equal schemas accept equal instance sets, so inclusion holds. This also + # covers assertions the delegated engine cannot reason about (e.g. nested + # object-valued ``const``/``enum``), where it would otherwise report unknown. + # The GTS reference dimension is compared separately (README §9.6.1), so it + # is sound for this structural primitive to ignore ``x-gts-*`` here. JSON + # spelling distinguishes ``true`` from ``1`` (and ``false`` from ``0``), so + # compare canonical JSON rather than Python equality, which conflates them. + if _json_key(sanitize(_coerce_bool_schema(lowered_subset))) == _json_key( + sanitize(_coerce_bool_schema(lowered_superset)) + ): + return True finite_result = _finite_subset(lowered_subset, lowered_superset) if finite_result is not None: return finite_result @@ -430,18 +441,39 @@ def dialects_differ(old_schema: Any, new_schema: Any) -> bool: ) +def _inclusion_verdict(subset: Any, superset: Any) -> str: + """Structural inclusion folded with the GTS reference-scope dimension. + + The reference scopes are compared separately (README §9.6.1) because the + delegated structural engine does not understand the reference keywords. A + proved reference exclusion is incompatible; an unknown reference relation is + unknown unless the structural dimension already disproves inclusion. + """ + from . import ref_scope_compat + + ref_rel = ref_scope_compat.schema_ref_relation(superset, subset) + if ref_rel == ref_scope_compat.EXCLUDED: + return INCOMPATIBLE + structural = _verdict(_is_subschema(subset, superset)) + if structural == INCOMPATIBLE: + return INCOMPATIBLE + if ref_rel == ref_scope_compat.UNKNOWN: + return UNKNOWN + return structural + + def check_backward_compatibility(old_schema: Any, new_schema: Any) -> str: """new consumers read old data: ``Valid(old) subset-of Valid(new)``.""" if dialects_differ(old_schema, new_schema): return UNKNOWN - return _verdict(_is_subschema(old_schema, new_schema)) + return _inclusion_verdict(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 dialects_differ(old_schema, new_schema): return UNKNOWN - return _verdict(_is_subschema(new_schema, old_schema)) + return _inclusion_verdict(new_schema, old_schema) def full_verdict(backward: str, forward: str) -> str: diff --git a/gts/src/gts/derivation.py b/gts/src/gts/derivation.py index d59b64e..1f90f5e 100644 --- a/gts/src/gts/derivation.py +++ b/gts/src/gts/derivation.py @@ -18,6 +18,7 @@ import copy from typing import Any +from . import ref_scope_compat from .compatibility import boolean_schema_value, check_accepted_set_inclusion MAX_RECURSION_DEPTH = 64 @@ -38,6 +39,19 @@ def validate_derivation_compatibility( base_schema, derived_schema, base_id, derived_id ) ) + # GTS reference scopes (README §9.6.1): the derived scope must be provably + # included in the base's. Admission fails closed, so an unknown relation is + # rejected too. + if ( + ref_scope_compat.schema_ref_relation( + base_schema, derived_schema, base_id, derived_id + ) + != ref_scope_compat.INCLUDED + ): + errors.append( + f"derived schema '{derived_id}' is not included in base '{base_id}': " + "a GTS reference constraint is not included in the base constraint" + ) return errors diff --git a/gts/src/gts/gts.py b/gts/src/gts/gts.py index 9303a4e..8c1040a 100644 --- a/gts/src/gts/gts.py +++ b/gts/src/gts/gts.py @@ -5,9 +5,9 @@ import uuid from typing import Any +from . import gts_id_pattern from ._naming import ( GTS_PREFIX, - GTS_TYPE_MARKER, ) from ._naming import ( has_scheme as _has_scheme, @@ -23,6 +23,8 @@ ) GTS_NS = uuid.uuid5(uuid.NAMESPACE_URL, "gts") +# Major and minor version components are unsigned 32-bit integers (README §2.3). +GTS_MAX_U32_VERSION = gts_id_pattern.MAX_U32 GTS_SEGMENT_TOKEN_REGEX = re.compile(r"^[a-z_][a-z0-9_]*$") UUID_REGEX = re.compile( r"^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$" @@ -65,241 +67,55 @@ def __init__(self, pattern: str, cause: str | None = None): self.cause = cause -class GtsIdSegment: - """Parsed GTS segment. Accepts a segment string in the constructor. +class GtsIdSegment(gts_id_pattern.Segment): + """A parsed GTS segment. - The `segment` may be absolute (starts with 'gts.') or relative (no prefix). - The original string is stored in `segment`. + The grammar, parsing and version/wildcard bounds live in + :mod:`gts.gts_id_pattern` (the single identifier/pattern parser, mirroring + the Go reference's ``gtsid`` package). This thin adapter preserves the + historical ``GtsIdSegment(num, offset, segment)`` construction API and + raises :class:`GtsInvalidSegment`; parsed identifiers store plain + :class:`gts_id_pattern.Segment` instances, which share the same fields. """ - def __init__(self, num: int, offset: int, segment: str): - self.num: int = num - self.offset: int = offset - self.segment: str = segment.strip() - - self.vendor: str = "" - self.package: str = "" - self.namespace: str = "" - self.type: str = "" - self.ver_major: int | None = 0 - self.ver_minor: int | None = None - self.is_type: bool = False - self.is_wildcard: bool = False - # Marks the synthetic UUID tail of a combined anonymous instance id. - self._is_uuid_tail: bool = False - - self._parse_segment_id(num, offset, segment) - - def _parse_segment_id(self, num: int, offset: int, segment: str): - 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(GTS_TYPE_MARKER): - self.is_type = True - segment = segment[:-1] - else: - raise GtsInvalidSegment(num, offset, segment, " '~' must be at the end") - - tokens = segment.split(".") - - if len(tokens) > 6: - raise GtsInvalidSegment(num, offset, segment, "Too many tokens") - - if not segment.endswith("*"): - if len(tokens) < 5: - raise GtsInvalidSegment(num, offset, segment, "Too few tokens") - - for t in range(4): - if not GTS_SEGMENT_TOKEN_REGEX.match(tokens[t]): - raise GtsInvalidSegment( - num, offset, segment, "Invalid segment token: " + tokens[t] - ) - - if len(tokens) > 0: - if tokens[0] == "*": - self.is_wildcard = True - return - self.vendor = tokens[0] - - if len(tokens) > 1: - if tokens[1] == "*": - self.is_wildcard = True - return - self.package = tokens[1] - - if len(tokens) > 2: - if tokens[2] == "*": - self.is_wildcard = True - return - self.namespace = tokens[2] - - if len(tokens) > 3: - if tokens[3] == "*": - self.is_wildcard = True - return - self.type = tokens[3] - - if len(tokens) > 4: - if tokens[4] == "*": - self.is_wildcard = True - return - - if not tokens[4].startswith("v"): - raise GtsInvalidSegment( - num, offset, segment, "Major version must start with 'v'" - ) - try: - self.ver_major = int(tokens[4][1:]) - except ValueError: - raise GtsInvalidSegment( - num, offset, segment, "Major version must be an integer" - ) - - if self.ver_major < 0: - raise GtsInvalidSegment( - num, offset, segment, "Major version must be >= 0" - ) - if str(self.ver_major) != tokens[4][1:]: - raise GtsInvalidSegment( - num, offset, segment, "Major version must be an integer" - ) - - if len(tokens) > 5: - if tokens[5] == "*": - self.is_wildcard = True - return - - try: - self.ver_minor = int(tokens[5]) - except ValueError: - raise GtsInvalidSegment( - num, offset, segment, "Minor version must be an integer" - ) - - if self.ver_minor < 0: - raise GtsInvalidSegment( - num, offset, segment, "Minor version must be >= 0" - ) - if str(self.ver_minor) != tokens[5]: - raise GtsInvalidSegment( - num, offset, segment, "Minor version must be an integer" - ) + __slots__ = () - @classmethod - def _uuid_tail_segment(cls, num: int, offset: int, uuid_str: str) -> GtsIdSegment: - """Create a special UUID tail segment for combined anonymous instances.""" - seg = object.__new__(cls) - seg.num = num - seg.offset = offset - seg.segment = uuid_str - seg.vendor = "" - seg.package = "" - seg.namespace = "" - seg.type = "" - seg.ver_major = None - seg.ver_minor = None - seg.is_type = False - seg.is_wildcard = False - seg._is_uuid_tail = True - return seg + def __init__(self, num: int = 0, offset: int = 0, segment: str = ""): + super().__init__() + try: + parsed = gts_id_pattern.parse_segment(segment) + except gts_id_pattern.InvalidPattern as error: + raise GtsInvalidSegment(num, offset, segment, str(error)) from error + for attr in gts_id_pattern.Segment.__slots__: + setattr(self, attr, getattr(parsed, attr)) class GtsID: - def __init__(self, id: str): - raw = id.strip() - - # A GtsID is always the bare canonical form ("gts.…"). The "gts://" URI - # form is a JSON Schema serialization detail ($id/$ref) and is stripped - # by those URI-specific callers (e.g. entity extraction via - # ``strip_scheme``) before reaching here, mirroring the gts-rust/gts-go - # reference implementations. Accepting it here would let URI-form values - # pass validate-id/parse-id and disagree with the canonical ``id``. - - # Validate it's lower case - if raw != raw.lower(): - raise GtsInvalidId(id, "Must be lower case") - - if not raw.startswith(GTS_PREFIX): - raise GtsInvalidId(id, f"Does not start with '{GTS_PREFIX}'") - if len(raw) > 1024: - raise GtsInvalidId(id, "Too long") - - self.id: str = raw - self.gts_id_segments: list[GtsIdSegment] = [] - self.uuid_tail: str | None = None - - # Detect combined anonymous instance: last tilde-part is a UUID - remainder = raw[len(GTS_PREFIX) :] - tilde_parts = remainder.split("~") - last_part = tilde_parts[-1] if tilde_parts else "" - if UUID_REGEX.match(last_part) and len(tilde_parts) >= 2: - self.uuid_tail = last_part - # Hyphens are only allowed in the UUID tail - segments_portion = raw[: len(raw) - len(last_part) - 1] # strip ~ - if "-" in segments_portion: - raise GtsInvalidId(id, "Must not contain '-'") - else: - if "-" in raw: - raise GtsInvalidId(id, "Must not contain '-'") - - # split preserving empties to detect trailing '~' - _parts = raw[len(GTS_PREFIX) :].split("~") + """A validated, canonical GTS identifier value object. - # If UUID tail, exclude it from segment parsing - if self.uuid_tail: - # All parts before UUID are type segments (end with ~) - seg_count = len(_parts) - 1 # exclude UUID tail - parts = [] - for i in range(seg_count): - if _parts[i] == "": - raise GtsInvalidId(id, f"GTS segment #{i + 1} is empty") - parts.append(_parts[i] + "~") - else: - parts = [] - for i in range(len(_parts)): - if i < len(_parts) - 1: - parts.append(_parts[i] + "~") - if i == len(_parts) - 2 and _parts[i + 1] == "": - break - else: - parts.append(_parts[i]) - - offset = len(GTS_PREFIX) - for i in range(len(parts)): - if parts[i] == "": - raise GtsInvalidId( - id, f"GTS segment #{i + 1} @ offset {offset} is empty" - ) - - self.gts_id_segments.append(GtsIdSegment(i + 1, offset, parts[i])) - offset += len(parts[i]) - - # Add UUID tail as a special segment if present - if self.uuid_tail: - self.gts_id_segments.append( - GtsIdSegment._uuid_tail_segment( - len(self.gts_id_segments) + 1, offset, self.uuid_tail - ) - ) + Parsing, the §2.3/§10 grammar, the unsigned 32-bit version bounds and the + UUID-tail handling are delegated to :mod:`gts.gts_id_pattern`; this class + adds the higher-level value-object API (URI form, type/instance queries, + UUID derivation, pattern matching and attribute-path helpers). + """ - # Issue #37: Single-segment instance IDs are not allowed - # An instance ID (not ending with ~) must be chained (have at least 2 segments) - # UUID tail is exempt - non_uuid_segments = [ - s for s in self.gts_id_segments if not getattr(s, "_is_uuid_tail", False) - ] - if ( - 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) - ): - # Check if it's a wildcard (wildcards are allowed as single segment) - raise GtsInvalidId( - id, - "Single-segment instance IDs are not allowed. " - "Instance IDs must be chained (e.g., type~instance).", - ) + def __init__(self, id: str): + # The bare canonical form is expected; ``gts://`` URI inputs are + # stripped by URI-specific callers before reaching here. Leading and + # trailing whitespace is tolerated for backwards compatibility (the + # stricter pattern parser rejects it for OP#4 and the reference + # keywords). + raw = id.strip() + try: + parsed = gts_id_pattern.parse_concrete(raw) + except gts_id_pattern.InvalidPattern as error: + raise GtsInvalidId(id, str(error)) from error + self.id: str = parsed.id + self.gts_id_segments: list[gts_id_pattern.Segment] = parsed.segments + # Duck-types as gts_id_pattern.ParsedPattern for matching/coverage. + self.segments: list[gts_id_pattern.Segment] = parsed.segments + last = parsed.segments[-1] + self.uuid_tail: str | None = last.segment if last.is_uuid else None @property def uri(self) -> str: @@ -334,7 +150,7 @@ def get_type_id(self) -> str | None: return GTS_PREFIX + "".join([s.segment for s in self.gts_id_segments[:-1]]) def to_uuid(self) -> uuid.UUID: - # For combined anonymous instances, return the embedded UUID directly + # For combined anonymous instances, return the embedded UUID directly. if self.uuid_tail: return uuid.UUID(self.uuid_tail) return uuid.uuid5(GTS_NS, self.id) @@ -342,7 +158,7 @@ def to_uuid(self) -> uuid.UUID: @classmethod def is_valid(cls, s: str) -> bool: # Only the bare canonical form is a valid id; the "gts://" URI form is - # stripped by URI-specific callers before validation (see GtsID.__init__). + # stripped by URI-specific callers before validation. if not s.startswith(GTS_PREFIX): return False try: @@ -352,94 +168,17 @@ def is_valid(cls, s: str) -> bool: return False def wildcard_match(self, pattern: GtsWildcard) -> bool: - p = pattern.id - - # Helper function to match segments with version flexibility - def match_segments( - pattern_segs: list[GtsIdSegment], candidate_segs: list[GtsIdSegment] - ) -> bool: - # Pattern ending with '~*' means "this type and any descendants". - # It should match both: - # - the base type itself (same prefix, no extra segment), and - # - instances/derived ids under that prefix. - if ( - pattern_segs - and pattern_segs[-1].is_wildcard - and len(pattern_segs) == len(candidate_segs) + 1 - ): - return match_segments(pattern_segs[:-1], candidate_segs) - - # If pattern is longer than candidate, no match - if len(pattern_segs) > len(candidate_segs): - return False - - for i, p_seg in enumerate(pattern_segs): - c_seg = candidate_segs[i] - - # If pattern segment is a wildcard, check non-wildcard fields first - if p_seg.is_wildcard: - # Check the fields that are set (non-empty) in the wildcard pattern - if p_seg.vendor and p_seg.vendor != c_seg.vendor: - return False - if p_seg.package and p_seg.package != c_seg.package: - return False - if p_seg.namespace and p_seg.namespace != c_seg.namespace: - return False - if p_seg.type and p_seg.type != c_seg.type: - return False - # Check version fields when version is explicitly present in - # the wildcard segment (including v0.*). - if ".v" in p_seg.segment and p_seg.ver_major != c_seg.ver_major: - return False - if ( - p_seg.ver_minor is not None - and p_seg.ver_minor != c_seg.ver_minor - ): - return False - # Check is_type flag if set; if it doesn't match, wildcard fails, - # otherwise wildcard matches - accept anything after this point - return not (p_seg.is_type and p_seg.is_type != c_seg.is_type) - - # Non-wildcard segment - all fields must match exactly - # Check vendor, package, namespace, type match - if p_seg.vendor != c_seg.vendor: - return False - if p_seg.package != c_seg.package: - return False - if p_seg.namespace != c_seg.namespace: - return False - if p_seg.type != c_seg.type: - return False - - # Check version matching - # Major version must match - if p_seg.ver_major != c_seg.ver_major: - return False - - # Minor version: if pattern has no minor version, accept any minor in candidate - # If pattern has minor version, it must match exactly - if p_seg.ver_minor is not None and p_seg.ver_minor != c_seg.ver_minor: - return False - # else: pattern has no minor version, so any minor version in candidate is OK - - # Check is_type flag matches - if p_seg.is_type != c_seg.is_type: - return False - - # If we've matched all pattern segments, it's a match - return True - - # No wildcard case - need exact match with version flexibility - if "*" not in p: - # Parse both as segments and compare - return match_segments(pattern.gts_id_segments, self.gts_id_segments) - - # Wildcard case - if p.count("*") > 1 or not p.endswith("*"): - return False - - # Use segment matching for wildcard patterns too - return match_segments(pattern.gts_id_segments, self.gts_id_segments) + """Whether this identifier is selected by ``pattern`` (README §10). + + Shares the single matching implementation with OP#4 and the reference + keywords. A concrete identifier is checked by point matching; a wildcard + identifier (candidate) by pattern coverage. + """ + this = gts_id_pattern.ParsedPattern(self.id, self.gts_id_segments) + other = gts_id_pattern.ParsedPattern(pattern.id, pattern.gts_id_segments) + if any(seg.is_wildcard for seg in self.gts_id_segments): + return gts_id_pattern.covers(other, this) + return gts_id_pattern.matches_pattern(this, other) def parse_query(self, expr: str) -> tuple[str, dict[str, str]]: base, _, filt = expr.partition("[") @@ -477,23 +216,23 @@ def split_at_path(cls, gts_with_path: str) -> tuple[str, str | None]: class GtsWildcard(GtsID): + """A validated GTS identifier match pattern (README §10). + + A concrete identifier or one ending in a single trailing ``*`` wildcard + token; parsing is delegated to :mod:`gts.gts_id_pattern`. + """ + def __init__(self, pattern: str): - p = pattern.strip() - if not p.startswith(GTS_PREFIX): - raise GtsInvalidWildcard(pattern, f"Does not start with '{GTS_PREFIX}'") - if p.count("*") > 1: - raise GtsInvalidWildcard( - pattern, "The wildcard '*' token is allowed only once" - ) - if "*" in p and not p.endswith(".*") and not p.endswith("~*"): - raise GtsInvalidWildcard( - pattern, - "The wildcard '*' token is allowed only at the end of the pattern", - ) + raw = pattern.strip() try: - super().__init__(p) - except GtsInvalidId as e: - raise GtsInvalidWildcard(pattern, str(e)) + parsed = gts_id_pattern.parse_pattern(raw) + except gts_id_pattern.InvalidPattern as error: + raise GtsInvalidWildcard(pattern, str(error)) from error + self.id = parsed.id + self.gts_id_segments = parsed.segments + self.segments = parsed.segments + last = parsed.segments[-1] + self.uuid_tail = last.segment if last.is_uuid else None class GtsRef: diff --git a/gts/src/gts/gts_id_pattern.py b/gts/src/gts/gts_id_pattern.py new file mode 100644 index 0000000..59c69d8 --- /dev/null +++ b/gts/src/gts/gts_id_pattern.py @@ -0,0 +1,613 @@ +"""GTS identifier pattern parsing, matching and coverage (README §2.3, §10). + +This module is a self-contained implementation of the GTS identifier *pattern* +grammar and the two relations the specification builds on it: + +- :func:`matches_pattern` — whether a concrete identifier is selected by a + pattern (``candidate ∈ M(pattern)``), used by OP#4 point matching and by the + ``x-gts-type-ref`` / ``x-gts-instance-ref`` candidate checks. +- :func:`covers` — whether every identifier selected by one pattern is also + selected by another (``M(value) ⊆ M(operand)``), used by OP#4 pattern + matching, ``x-gts-pattern-ref`` and the reference derivation/evolution + relations. + +A *pattern* is a validated GTS identifier that may end in a single trailing +``*`` wildcard token but may also be fully concrete. A zero-wildcard pattern is +not an exact match: a base type id used as a pattern also matches everything +derived from it down the chain, with minor-version flexibility. + +It is a faithful port of the Go reference implementation +(``gts-go/gtsid/{id,parse,coverage,match}.go``) and deliberately depends only on +the grammar markers, mirroring that layer's isolation from the rest of the SDK. +""" + +from __future__ import annotations + +import re + +from ._naming import GTS_PREFIX + +PREFIX = GTS_PREFIX +TYPE_MARKER = "~" +WILDCARD_MARKER = "*" +TOKEN_SEP = "." +SEGMENT_SEP = TYPE_MARKER +MAX_ID_LENGTH = 1024 +MAX_U32 = 4294967295 + +_SEGMENT_TOKEN_RE = re.compile(r"^[a-z_][a-z0-9_]*$") +_UUID_RE = re.compile(r"^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$") + + +class InvalidPattern(ValueError): + """A GTS identifier or pattern string is not well formed (§2.3, §10).""" + + +def has_prefix(s: str) -> bool: + return s.startswith(PREFIX) + + +def has_wildcard(s: str) -> bool: + return WILDCARD_MARKER in s + + +def is_type_id(s: str) -> bool: + """Whether a concrete (non-wildcard) id spelling denotes a type.""" + return s.endswith(TYPE_MARKER) + + +class Segment: + """A parsed segment of a GTS identifier or pattern.""" + + __slots__ = ( + "has_version", + "is_type", + "is_uuid", + "is_wildcard", + "namespace", + "package", + "segment", + "type", + "vendor", + "ver_major", + "ver_minor", + ) + + def __init__(self) -> None: + self.segment: str = "" + self.vendor: str = "" + self.package: str = "" + self.namespace: str = "" + self.type: str = "" + self.ver_major: int | None = 0 + self.ver_minor: int | None = None + self.has_version: bool = False + self.is_type: bool = False + self.is_wildcard: bool = False + self.is_uuid: bool = False + + def ver_major_opt(self) -> tuple[int, bool]: + return (self.ver_major or 0), self.has_version + + def uuid_tail(self) -> tuple[str, bool]: + return (self.segment, True) if self.is_uuid else ("", False) + + @property + def _is_uuid_tail(self) -> bool: + # Backwards-compatible alias used by the ``GtsID`` value object and + # schema-cast, which mark the synthetic UUID tail of a combined + # anonymous instance (spec §3.7). + return self.is_uuid + + +class ParsedPattern: + """A validated GTS identifier or match pattern.""" + + __slots__ = ("id", "segments") + + def __init__(self, raw: str, segments: list[Segment]) -> None: + self.id = raw + self.segments = segments + + def is_type(self) -> bool: + return is_type_id(self.id) + + def is_wildcard(self) -> bool: + return any(seg.is_wildcard for seg in self.segments) + + +def _split_preserving_tilde(s: str) -> list[str]: + _parts = s.split(SEGMENT_SEP) + parts: list[str] = [] + for i in range(len(_parts)): + if i < len(_parts) - 1: + parts.append(_parts[i] + SEGMENT_SEP) + if i == len(_parts) - 2 and _parts[i + 1] == "": + break + else: + parts.append(_parts[i]) + return parts + + +def _is_uuid_segment(s: str) -> bool: + return _UUID_RE.match(s) is not None + + +def _parse_segment(segment: str) -> Segment: + seg = Segment() + seg.segment = segment.strip() + working = seg.segment + + if working.count(TYPE_MARKER) > 0: + if working.count(TYPE_MARKER) > 1: + raise InvalidPattern("Too many '~' characters") + if working.endswith(TYPE_MARKER): + seg.is_type = True + working = working[:-1] + else: + raise InvalidPattern(" '~' must be at the end") + + tokens = working.split(TOKEN_SEP) + + if len(tokens) > 6: + raise InvalidPattern("Too many tokens") + + if not working.endswith(WILDCARD_MARKER) and len(tokens) < 5: + raise InvalidPattern("Too few tokens") + + # Validate the name tokens (first up to four). A bare '*' is permitted only + # as the final token; every other name token must be a valid segment token. + name_tokens = min(len(tokens), 4) + for t in range(name_tokens): + if tokens[t] == WILDCARD_MARKER: + if t != len(tokens) - 1: + raise InvalidPattern("Wildcard '*' is only allowed as the final token") + break + if not _SEGMENT_TOKEN_RE.match(tokens[t]): + raise InvalidPattern("Invalid segment token: " + tokens[t]) + + if len(tokens) > 0: + if tokens[0] == WILDCARD_MARKER: + seg.is_wildcard = True + return seg + seg.vendor = tokens[0] + + if len(tokens) > 1: + if tokens[1] == WILDCARD_MARKER: + seg.is_wildcard = True + return seg + seg.package = tokens[1] + + if len(tokens) > 2: + if tokens[2] == WILDCARD_MARKER: + seg.is_wildcard = True + return seg + seg.namespace = tokens[2] + + if len(tokens) > 3: + if tokens[3] == WILDCARD_MARKER: + seg.is_wildcard = True + return seg + seg.type = tokens[3] + + if len(tokens) > 4: + if tokens[4] == WILDCARD_MARKER: + seg.is_wildcard = True + return seg + # Glued version wildcard `v*`: any version (§10); only as final token. + if tokens[4] == "v" + WILDCARD_MARKER: + if len(tokens) != 5: + raise InvalidPattern("Wildcard '*' is only allowed as the final token") + seg.is_wildcard = True + return seg + if not tokens[4].startswith("v"): + raise InvalidPattern("Major version must start with 'v'") + major_str = tokens[4][1:] + seg.ver_major = _parse_version(major_str, "Major") + seg.has_version = True + + if len(tokens) > 5: + if tokens[5] == WILDCARD_MARKER: + seg.is_wildcard = True + return seg + seg.ver_minor = _parse_version(tokens[5], "Minor") + + return seg + + +def parse_segment(segment: str) -> Segment: + """Parse a single GTS segment in isolation (no chain/single-segment rules).""" + return _parse_segment(segment) + + +def _parse_version(text: str, label: str) -> int: + if not text or not text.isdecimal(): + raise InvalidPattern(f"{label} version must be an integer") + value = int(text) + # Reject leading zeros ("01") which isdecimal accepts. + if str(value) != text: + raise InvalidPattern(f"{label} version must be an integer") + if value > MAX_U32: + raise InvalidPattern(f"{label} version must be an unsigned 32-bit integer") + return value + + +def _parse_gts_id(raw: str) -> ParsedPattern: + if raw.strip() != raw: + raise InvalidPattern("Must not contain leading or trailing whitespace") + if raw != raw.lower(): + raise InvalidPattern("Must be lower case") + if not has_prefix(raw): + raise InvalidPattern(f"Does not start with '{PREFIX}'") + if len(raw) > MAX_ID_LENGTH: + raise InvalidPattern("Too long") + + remainder = raw[len(PREFIX) :] + parts = _split_preserving_tilde(remainder) + + has_uuid_tail = len(parts) >= 2 and _is_uuid_segment(parts[-1]) + + gts_parts_to_check = parts[:-1] if has_uuid_tail else parts + for p in gts_parts_to_check: + if "-" in p: + raise InvalidPattern("Must not contain '-'") + + segments: list[Segment] = [] + for i, part in enumerate(parts): + if part == "": + raise InvalidPattern(f"GTS segment #{i + 1} is empty") + if has_uuid_tail and i == len(parts) - 1: + seg = Segment() + seg.segment = part + seg.is_uuid = True + seg.ver_major = None + segments.append(seg) + continue + segments.append(_parse_segment(part)) + + parsed = ParsedPattern(raw, segments) + if len(segments) == 1 and not parsed.is_type() and not segments[0].is_wildcard: + raise InvalidPattern( + "Single-segment instances are prohibited. " + "Well-known instances must be chained with a type segment" + ) + return parsed + + +def _count_wildcards(s: str) -> int: + return s.count(WILDCARD_MARKER) + + +def parse_pattern(pattern: str) -> ParsedPattern: + """Parse a GTS identifier match pattern. + + A concrete GTS identifier or one ending in a single trailing ``*`` wildcard + token. Raises :class:`InvalidPattern` for any malformed input. + """ + if pattern.strip() != pattern: + raise InvalidPattern("must not contain leading or trailing whitespace") + if not has_prefix(pattern): + raise InvalidPattern(f"Does not start with '{PREFIX}'") + count = _count_wildcards(pattern) + if count > 1: + raise InvalidPattern("The wildcard '*' token is allowed only once") + if count == 1 and not pattern.endswith(WILDCARD_MARKER): + raise InvalidPattern( + "The wildcard '*' token is allowed only at the end of the pattern" + ) + return _parse_gts_id(pattern) + + +def parse_concrete(candidate: str) -> ParsedPattern: + """Parse a concrete (non-wildcard) GTS identifier, validated as spelled.""" + return _parse_gts_id(candidate) + + +# -------------------------------------------------------------------------- +# Point matching: candidate ∈ M(pattern) (README §10). +# -------------------------------------------------------------------------- + + +def matches_pattern(candidate: ParsedPattern, pattern: ParsedPattern) -> bool: + return _matches_views(pattern.segments, candidate.segments) + + +def _matches_views(pattern_segs: list[Segment], candidate: list[Segment]) -> bool: + if not pattern_segs: + return False + last = pattern_segs[-1] + matches_empty_suffix = last.is_wildcard and last.segment == WILDCARD_MARKER + required_len = len(pattern_segs) - (1 if matches_empty_suffix else 0) + if required_len > len(candidate): + return False + + for i, p_seg in enumerate(pattern_segs): + if i == len(candidate): + return matches_empty_suffix and i == len(pattern_segs) - 1 + c_seg = candidate[i] + + if p_seg.is_wildcard: + if p_seg.vendor and p_seg.vendor != c_seg.vendor: + return False + if p_seg.package and p_seg.package != c_seg.package: + return False + if p_seg.namespace and p_seg.namespace != c_seg.namespace: + return False + if p_seg.type and p_seg.type != c_seg.type: + return False + p_major, p_has = p_seg.ver_major_opt() + if p_has: + c_major, c_has = c_seg.ver_major_opt() + if not c_has or p_major != c_major: + return False + # `v1.*` requires an explicit minor in the concrete id. + if c_seg.ver_minor is None: + return False + # A pinned minor in the wildcard segment must match exactly. + return p_seg.ver_minor is None or ( + c_seg.ver_minor is not None and p_seg.ver_minor == c_seg.ver_minor + ) + + p_uuid, p_is_uuid = p_seg.uuid_tail() + if p_is_uuid: + c_uuid, c_is_uuid = c_seg.uuid_tail() + if not c_is_uuid or p_uuid != c_uuid: + return False + + if ( + p_seg.vendor != c_seg.vendor + or p_seg.package != c_seg.package + or p_seg.namespace != c_seg.namespace + or p_seg.type != c_seg.type + ): + return False + if not _major_eq(p_seg, c_seg): + return False + if p_seg.ver_minor is not None and ( + c_seg.ver_minor is None or p_seg.ver_minor != c_seg.ver_minor + ): + return False + if p_seg.is_type != c_seg.is_type: + return False + return True + + +def _major_eq(a: Segment, b: Segment) -> bool: + am, aok = a.ver_major_opt() + bm, bok = b.ver_major_opt() + return aok == bok and (not aok or am == bm) + + +def _minor_eq(a: Segment, b: Segment) -> bool: + if a.ver_minor is None: + return b.ver_minor is None + return b.ver_minor is not None and a.ver_minor == b.ver_minor + + +# -------------------------------------------------------------------------- +# Coverage: M(value) ⊆ M(operand) (README §10). +# -------------------------------------------------------------------------- + +_SEL_INSTANCE = 0 +_SEL_PREFIXED = 1 +_MIN_SEGMENT_LEN = 10 # "a.a.a.a.v0" + + +class _Selection: + __slots__ = ("kind", "length", "next", "tail", "types") + + def __init__(self, kind: int) -> None: + self.kind = kind + self.types: list[Segment] = [] + self.tail: Segment | None = None + self.length: int = 0 + self.next: Segment | None = None + + +def _selection_of(pattern: ParsedPattern) -> _Selection: + segs = pattern.segments + end = len(segs) - 1 + last = segs[end] + if last.is_wildcard: + sel = _Selection(_SEL_PREFIXED) + sel.types = segs[:end] + sel.next = None if last.segment == WILDCARD_MARKER else last + return sel + if last.is_type: + sel = _Selection(_SEL_PREFIXED) + sel.types = segs + sel.next = None + return sel + sel = _Selection(_SEL_INSTANCE) + sel.types = segs[:end] + sel.tail = last + sel.length = len(pattern.id) + return sel + + +def _shortest_selected(s: _Selection) -> int: + if s.kind == _SEL_INSTANCE: + return s.length + chain = len(PREFIX) + for segment in s.types: + chain += len(segment.segment) + if s.next is None and len(s.types) > 0: + return chain + segment_len = _MIN_SEGMENT_LEN if s.next is None else _shortest_partial(s.next) + if len(s.types) == 0: + return chain + segment_len + 1 + return chain + segment_len + + +def _shortest_partial(partial: Segment) -> int: + if partial.ver_major_opt()[1]: + # In `v1.*`, replacing `*` with the shortest minor (`0`) keeps length. + return len(partial.segment) + total = 6 + for name in (partial.vendor, partial.package, partial.namespace, partial.type): + total += 1 if len(name) == 0 else len(name) + return total + + +def covers(operand: ParsedPattern, value: ParsedPattern) -> bool: + """Whether every GTS id matched by ``value`` is also matched by ``operand``.""" + v = _selection_of(value) + shortest = _shortest_selected(v) + return shortest > MAX_ID_LENGTH or _covers_nonempty( + _selection_of(operand), v, shortest == MAX_ID_LENGTH + ) + + +def _same_names(a: Segment, b: Segment) -> bool: + return ( + a.vendor == b.vendor + and a.package == b.package + and a.namespace == b.namespace + and a.type == b.type + ) + + +def _segment_covers(a: Segment, b: Segment) -> bool: + return ( + _same_names(a, b) + and _major_eq(a, b) + and (a.ver_minor is None or _minor_eq(a, b)) + ) + + +def _same_chain_covers(a: list[Segment], b: list[Segment]) -> bool: + if len(a) != len(b): + return False + return all(_segment_covers(a[i], b[i]) for i in range(len(a))) + + +def _partial_names_cover(a: Segment, b: Segment) -> bool: + return ( + (a.vendor == "" or a.vendor == b.vendor) + and (a.package == "" or a.package == b.package) + and (a.namespace == "" or a.namespace == b.namespace) + and (a.type == "" or a.type == b.type) + ) + + +def _partial_admits(partial: Segment, segment: Segment) -> bool: + if not _partial_names_cover(partial, segment): + return False + p_major, ok = partial.ver_major_opt() + if not ok: + return True + s_major, s_ok = segment.ver_major_opt() + return s_ok and s_major == p_major and segment.ver_minor is not None + + +def _partial_covers(a: Segment, b: Segment) -> bool: + if not _partial_names_cover(a, b): + return False + a_major, ok = a.ver_major_opt() + if not ok: + return True + b_major, b_ok = b.ver_major_opt() + return b_ok and b_major == a_major + + +def _prefixed_covers( + a: list[Segment], a_next: Segment | None, b: list[Segment], b_next: Segment | None +) -> bool: + if len(a) > len(b): + # A first-segment minor wildcard selects types contained in a bare type + # scope with the same names and major and an omitted minor. + if len(a) == 1 and a_next is None and len(b) == 0 and b_next is not None: + only = a[0] + _, only_has_major = only.ver_major_opt() + _, partial_has_major = b_next.ver_major_opt() + return ( + only.ver_minor is None + and _same_names(only, b_next) + and only_has_major == partial_has_major + and (not only_has_major or only.ver_major == b_next.ver_major) + ) + return False + for i in range(len(a)): + if not _segment_covers(a[i], b[i]): + return False + if a_next is None: + return True + if len(b) > len(a): + return _partial_admits(a_next, b[len(a)]) + return b_next is not None and _partial_covers(a_next, b_next) + + +def _covers_nonempty(operand: _Selection, value: _Selection, at_limit: bool) -> bool: + if operand.kind == _SEL_INSTANCE and value.kind == _SEL_INSTANCE: + # Instance selections always carry a tail segment (see _selection_of). + assert operand.tail is not None and value.tail is not None + if not _same_chain_covers(operand.types, value.types): + return False + a_uuid, aok = operand.tail.uuid_tail() + b_uuid, bok = value.tail.uuid_tail() + if aok and bok: + return a_uuid == b_uuid + if not aok and not bok: + return _segment_covers(operand.tail, value.tail) + return False + if operand.kind == _SEL_INSTANCE and value.kind == _SEL_PREFIXED: + if not at_limit or value.next is None: + return False + assert operand.tail is not None + tail = operand.tail + partial = value.next + _, tail_has_major = tail.ver_major_opt() + _, partial_has_major = partial.ver_major_opt() + _, tail_is_uuid = tail.uuid_tail() + return ( + _same_chain_covers(operand.types, value.types) + and not tail_is_uuid + and tail.ver_minor is None + and _same_names(tail, partial) + and tail_has_major == partial_has_major + and (not tail_has_major or tail.ver_major == partial.ver_major) + ) + if operand.kind == _SEL_PREFIXED and value.kind == _SEL_INSTANCE: + assert value.tail is not None + types = operand.types + ids = value.types + if len(ids) < len(types): + return False + for i in range(len(types)): + if not _segment_covers(types[i], ids[i]): + return False + if operand.next is None: + return True + target = ids[len(types)] if len(ids) > len(types) else value.tail + return _partial_admits(operand.next, target) + return _prefixed_covers(operand.types, operand.next, value.types, value.next) + + +# -------------------------------------------------------------------------- +# OP#4 entry point. +# -------------------------------------------------------------------------- + + +def match(candidate: str, pattern: str) -> tuple[bool, str | None]: + """Match a candidate against a pattern (README §10, OP#4). + + Returns ``(matched, error)``. A concrete candidate matches when it is in + ``M(pattern)``; a wildcard candidate matches when the pattern covers its + whole selection set. An invalid pattern or candidate returns ``(False, + error)``. + """ + try: + pattern_id = parse_pattern(pattern) + except InvalidPattern as e: + return False, str(e) + if has_wildcard(candidate): + try: + candidate_id = parse_pattern(candidate) + except InvalidPattern as e: + return False, str(e) + return covers(pattern_id, candidate_id), None + try: + candidate_id = parse_concrete(candidate) + except InvalidPattern as e: + return False, str(e) + return matches_pattern(candidate_id, pattern_id), None diff --git a/gts/src/gts/ops.py b/gts/src/gts/ops.py index 6d3930a..ea6a4a8 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 . import gts_id_pattern from ._naming import GTS_PREFIX, GTS_URI_PREFIX, looks_like_gts, strip_scheme from .entities import DEFAULT_GTS_CONFIG, GtsConfig, GtsEntity from .files_reader import GtsFileReader @@ -819,20 +820,14 @@ def parse_id(self, gts_id: str) -> GtsIdParseResult: ) def match_id_pattern(self, candidate: str, pattern: str) -> GtsIdMatchResult: - try: - # If candidate contains '*', validate it as a wildcard pattern - # This catches malformed wildcards like 'a*' (wildcard not on token boundary) - if "*" in candidate: - # Validate candidate as a wildcard pattern first - _ = GtsWildcard(candidate) - c = GtsID(candidate) - p = GtsWildcard(pattern) - match = c.wildcard_match(p) - return GtsIdMatchResult(candidate=candidate, pattern=pattern, match=match) - except Exception as e: # noqa: BLE001 - converted to a result object at API boundary - return GtsIdMatchResult( - candidate=candidate, pattern=pattern, match=False, error=str(e) - ) + # README §10 / OP#4: a concrete candidate matches when it is in + # M(pattern); a wildcard candidate matches when the pattern covers its + # whole selection set. Both sides use parsed chain boundaries and the + # minor-version rules (never lexical startsWith). + matched, error = gts_id_pattern.match(candidate, pattern) + return GtsIdMatchResult( + candidate=candidate, pattern=pattern, match=matched, error=error or "" + ) def uuid(self, gts_id: str) -> GtsUuidResult: g = GtsID(gts_id) diff --git a/gts/src/gts/ref_scope_compat.py b/gts/src/gts/ref_scope_compat.py new file mode 100644 index 0000000..c2c8166 --- /dev/null +++ b/gts/src/gts/ref_scope_compat.py @@ -0,0 +1,539 @@ +"""Derivation/evolution comparison of the GTS reference keywords. + +README §9.6.1 "Derivation and evolution" (ADR-0007/0008). A property that +constrains a GTS reference accepts a set of strings; a derived property is +admissible when the strings it accepts are a subset of those the base accepts. +The two entity-reference keywords compare kind-filtered scopes; x-gts-pattern-ref +compares selection-set coverage. Composition (allOf, anyOf, not) combines scopes +by intersection, union and complement. + +The comparison is a sound three-valued decision (:data:`INCLUDED`, +:data:`EXCLUDED`, :data:`UNKNOWN`): it returns included or excluded only when it +can prove the relation, and unknown otherwise. OP#12 admission fails closed +(unknown is rejected); OP#8 reports unknown. + +Ported from the Go reference (``gts-go/gts/ref_scope_compat.go``). +""" + +from __future__ import annotations + +import copy +from typing import Any + +from . import gts_id_pattern as idp +from ._naming import strip_scheme +from .x_gts_ref import ( + REF_KEYWORDS, + SELF_ID_REF, + X_GTS_INSTANCE_REF, + X_GTS_PATTERN_REF, + ref_candidate_violation, +) + +# Three-valued subset result. +INCLUDED = "included" +EXCLUDED = "excluded" +UNKNOWN = "unknown" + +_MAX_DEPTH = 64 +_MIN_INSTANCE_SEGMENT_LEN = 10 # "a.a.a.a.v0" + +# Keywords whose values are literal data, never schemas: a keyword-shaped key +# inside them is data, not a reference keyword (README §9.6.3). +_LITERAL_KEYWORDS = frozenset({"const", "enum", "default", "examples"}) + +# refScope kinds. +_TOP = "top" +_ATOM = "atom" +_UNION = "union" +_INTER = "inter" +_NOT = "not" +_UNDECIDABLE = "undecidable" + + +class _Scope: + __slots__ = ("child", "children", "empty", "kind", "pattern", "ref_keyword") + + def __init__(self, kind: str) -> None: + self.kind = kind + self.ref_keyword: str | None = None + self.pattern: idp.ParsedPattern | None = None + self.empty: bool = False + self.children: list[_Scope] = [] + self.child: _Scope | None = None + + +def is_ref_keyword(key: str) -> bool: + return key in REF_KEYWORDS + + +def _ref_atom(keyword: str, operand: Any) -> _Scope | None: + """Single-keyword scope from an operand, or None when it is not usable (an + invalid operand is a schema error caught separately).""" + if not isinstance(operand, str) or operand == SELF_ID_REF: + return None + try: + pattern = idp.parse_pattern(operand) + except idp.InvalidPattern: + return None + atom = _Scope(_ATOM) + atom.ref_keyword = keyword + atom.pattern = pattern + # An instance reference to a concrete Type ID too long for any instance tail + # selects nothing (README §9.6.1 "Derivation and evolution"). + if ( + keyword == X_GTS_INSTANCE_REF + and not idp.has_wildcard(operand) + and idp.is_type_id(operand) + and len(operand) + _MIN_INSTANCE_SEGMENT_LEN > idp.MAX_ID_LENGTH + ): + atom.empty = True + return atom + + +def build_ref_scope(prop: Any, depth: int = 0) -> tuple[_Scope, bool]: + """The reference scope a property declares, and whether it mentions any + reference keyword at all (directly or through composition). Operands are + expected to be already resolved (no ``/$id``).""" + if not isinstance(prop, dict): + return _Scope(_TOP), False + if depth >= _MAX_DEPTH: + return _Scope(_UNDECIDABLE), True + + parts: list[_Scope] = [] + has = False + + for keyword in REF_KEYWORDS: + if keyword not in prop: + continue + has = True + atom = _ref_atom(keyword, prop[keyword]) + if atom is not None: + parts.append(atom) + + not_value = prop.get("not") + if isinstance(not_value, dict): + child_scope, child_has = build_ref_scope(not_value, depth + 1) + if child_has: + has = True + node = _Scope(_NOT) + node.child = child_scope + parts.append(node) + + for key in ("anyOf", "oneOf"): + branches = prop.get(key) + if not isinstance(branches, list): + continue + union: list[_Scope] = [] + branch_has = False + for branch in branches: + if not isinstance(branch, dict): + union.append(_Scope(_TOP)) + continue + bs, bh = build_ref_scope(branch, depth + 1) + if bh: + branch_has = True + union.append(bs) + if branch_has: + has = True + node = _Scope(_UNION) + node.children = union + parts.append(node) + + all_of = prop.get("allOf") + if isinstance(all_of, list): + for branch in all_of: + if not isinstance(branch, dict): + continue + bs, bh = build_ref_scope(branch, depth + 1) + if bh: + has = True + parts.append(bs) + + if not parts: + return _Scope(_TOP), has + if len(parts) == 1: + return parts[0], has + node = _Scope(_INTER) + node.children = parts + return node, has + + +def relation(base: _Scope, derived: _Scope) -> str: + """Whether Valid(derived) ⊆ Valid(base) for the reference dimension.""" + if base.kind == _UNDECIDABLE or derived.kind == _UNDECIDABLE: + return UNKNOWN + if base.kind == _TOP: + return INCLUDED + if derived.kind == _TOP: + return EXCLUDED + + # Decompose the derived side first. + if derived.kind == _UNION: + result = INCLUDED + for child in derived.children: + r = relation(base, child) + if r == EXCLUDED: + return EXCLUDED + if r == UNKNOWN: + result = UNKNOWN + return result + if derived.kind == _INTER: + for child in derived.children: + if relation(base, child) == INCLUDED: + return INCLUDED + return UNKNOWN + if derived.kind == _NOT: + if base.kind == _NOT: + # complement(D) ⊆ complement(B) ⟺ B ⊆ D. + assert derived.child is not None and base.child is not None + return relation(derived.child, base.child) + return UNKNOWN + + # Derived is an atom; dispatch on the base side. + if base.kind == _UNION: + for child in base.children: + if relation(child, derived) == INCLUDED: + return INCLUDED + return UNKNOWN + if base.kind == _INTER: + result = INCLUDED + for child in base.children: + r = relation(child, derived) + if r == EXCLUDED: + return EXCLUDED + if r == UNKNOWN: + result = UNKNOWN + return result + if base.kind == _NOT: + return UNKNOWN + if base.kind == _ATOM: + return _atom_relation(base, derived) + return UNKNOWN + + +def _atom_relation(base: _Scope, derived: _Scope) -> str: + if derived.empty: + return INCLUDED + base_pattern_kw = base.ref_keyword == X_GTS_PATTERN_REF + derived_pattern_kw = derived.ref_keyword == X_GTS_PATTERN_REF + if base_pattern_kw != derived_pattern_kw: + return UNKNOWN + if base.ref_keyword != derived.ref_keyword: + return EXCLUDED # type vs instance: disjoint nonempty scopes + assert base.pattern is not None and derived.pattern is not None + if idp.covers(base.pattern, derived.pattern): + return INCLUDED + return EXCLUDED + + +def ref_scope_relation(base_prop: Any, derived_prop: Any) -> str: + """Whether the derived property's reference scope is included in the base's. + + Also admits a derived property that narrows with ordinary keywords: a finite + const/enum is included when every value lies in the base scope. + """ + base_scope, base_has = build_ref_scope(base_prop) + if not base_has: + return INCLUDED # base imposes no reference restriction + derived_scope, derived_has = build_ref_scope(derived_prop) + if derived_has: + return relation(base_scope, derived_scope) + values, enumerates = _collect_enumerated_values(derived_prop) + if not enumerates: + return EXCLUDED # a plain string or regex accepts values outside the base scope + for value in values: + if not isinstance(value, str) or not ref_scope_accepts(base_scope, value): + return EXCLUDED + return INCLUDED + + +def _collect_enumerated_values(prop: Any) -> tuple[list[Any], bool]: + if not isinstance(prop, dict): + return [], False + if "const" in prop: + return [prop["const"]], True + enum = prop.get("enum") + if isinstance(enum, list): + return list(enum), True + return [], False + + +def ref_scope_accepts(scope: _Scope, value: str) -> bool: + if scope.kind == _TOP: + return True + if scope.kind == _NOT: + assert scope.child is not None + return not ref_scope_accepts(scope.child, value) + if scope.kind == _UNION: + return any(ref_scope_accepts(c, value) for c in scope.children) + if scope.kind == _INTER: + return all(ref_scope_accepts(c, value) for c in scope.children) + if scope.kind == _ATOM: + if scope.empty: + return False + assert scope.ref_keyword is not None and scope.pattern is not None + return ( + ref_candidate_violation( + scope.ref_keyword, value, scope.pattern, scope.pattern.id + ) + is None + ) + return False + + +# --- effective-property folding (allOf) ------------------------------------ + + +def _narrower_ref_operand(keyword: str, a: Any, b: Any) -> tuple[Any, bool]: + """The operand (of two for the same keyword) selecting the smaller set, with + ok=True; ok=False when neither covers the other or either is uncomparable.""" + atom_a = _ref_atom(keyword, a) + atom_b = _ref_atom(keyword, b) + if atom_a is None or atom_b is None: + return None, False + assert atom_a.pattern is not None and atom_b.pattern is not None + if idp.covers(atom_a.pattern, atom_b.pattern): + return b, True # b ⊆ a + if idp.covers(atom_b.pattern, atom_a.pattern): + return a, True # a ⊆ b + return None, False + + +def merge_effective_property(prev: Any, nxt: Any) -> Any: + """Fold a later allOf-branch declaration of a property onto an earlier one. + + Ordinary keywords follow innermost-wins; a GTS reference keyword conjoins + like allOf (narrower operand survives, incomparable operands are kept as an + allOf conjunction) so a wider overlay never widens the inherited scope. + """ + if prev is None or not isinstance(prev, dict) or not isinstance(nxt, dict): + return nxt + merged = dict(nxt) + # Carry the earlier declaration's allOf conjuncts: a reference constraint + # nested in the inherited ``allOf`` would otherwise be lost when the later + # overlay redeclares the property with ordinary keywords, wrongly widening + # the effective scope. allOf is a conjunction, so order does not matter. + prev_all_of = prev.get("allOf") + if isinstance(prev_all_of, list): + existing = merged.get("allOf") + merged["allOf"] = list(prev_all_of) + ( + list(existing) if isinstance(existing, list) else [] + ) + # Likewise carry the earlier declaration's reference-bearing anyOf/oneOf/not + # as allOf conjuncts: innermost-wins would drop them when the overlay + # redeclares (or omits) the same applicator, widening the inherited scope. + for applicator in ("anyOf", "oneOf", "not"): + prev_value = prev.get(applicator) + if prev_value is None or merged.get(applicator) == prev_value: + continue + if not _mentions_ref_keyword(prev_value): + continue + existing = merged.get("allOf") + merged["allOf"] = [ + *(existing if isinstance(existing, list) else []), + {applicator: prev_value}, + ] + for keyword in REF_KEYWORDS: + if keyword not in prev: + continue + prev_operand = prev[keyword] + if keyword not in merged: + merged[keyword] = prev_operand + continue + narrower, ok = _narrower_ref_operand(keyword, prev_operand, merged[keyword]) + if ok: + merged[keyword] = narrower + continue + # Incomparable operands: keep both as a conjunction. + next_operand = merged.pop(keyword) + existing = merged.get("allOf") + both = list(existing) if isinstance(existing, list) else [] + both.append({keyword: prev_operand}) + both.append({keyword: next_operand}) + merged["allOf"] = both + return merged + + +def _effective_subschemas(schema: Any, depth: int = 0) -> dict[str, Any]: + """Effective child subschemas keyed by location, folding ``allOf`` branches + with reference-aware merging so an inherited narrower scope is not widened. + + Covers the applicators that constrain *every* instance value at their + location — ``properties``, ``patternProperties``, ``items`` (single or + tuple), ``prefixItems``, ``additionalItems`` and ``additionalProperties`` — + so a reference keyword under any of them participates in the subset + comparison. Existential applicators (``contains``) are deliberately omitted: + narrowing them does not narrow the accepted-instance set. + """ + result: dict[str, Any] = {} + if not isinstance(schema, dict) or depth >= _MAX_DEPTH: + return result + + def add(key: str, sub: Any) -> None: + if isinstance(sub, dict): + result[key] = merge_effective_property(result.get(key), sub) + + all_of = schema.get("allOf") + if isinstance(all_of, list): + for branch in all_of: + for key, sub in _effective_subschemas(branch, depth + 1).items(): + add(key, sub) + + properties = schema.get("properties") + if isinstance(properties, dict): + for name, sub in properties.items(): + add(f"properties/{name}", sub) + + pattern_properties = schema.get("patternProperties") + if isinstance(pattern_properties, dict): + for pattern, sub in pattern_properties.items(): + add(f"patternProperties/{pattern}", sub) + + items = schema.get("items") + if isinstance(items, dict): + add("items", items) + elif isinstance(items, list): + for index, sub in enumerate(items): + add(f"items/{index}", sub) + + prefix_items = schema.get("prefixItems") + if isinstance(prefix_items, list): + for index, sub in enumerate(prefix_items): + add(f"prefixItems/{index}", sub) + + for keyword in ("additionalItems", "additionalProperties"): + add(keyword, schema.get(keyword)) + + return result + + +# --- self-reference resolution --------------------------------------------- + + +def resolve_self_ref_operands(node: Any, self_id: str, depth: int = 0) -> None: + """Replace every ``/$id`` reference-keyword operand under ``node`` with + ``self_id`` (the bare top-level identifier of the schema being compared), + in place.""" + if depth >= _MAX_DEPTH: + return + if isinstance(node, dict): + for keyword in REF_KEYWORDS: + if node.get(keyword) == SELF_ID_REF: + node[keyword] = self_id + for value in node.values(): + resolve_self_ref_operands(value, self_id, depth + 1) + elif isinstance(node, list): + for value in node: + resolve_self_ref_operands(value, self_id, depth + 1) + + +def _self_id_of(schema: Any) -> str | None: + if isinstance(schema, dict): + sid = schema.get("$id") + if isinstance(sid, str): + return strip_scheme(sid) + return None + + +def _combine(current: str, child: str) -> str: + if child == EXCLUDED: + return EXCLUDED + if child == UNKNOWN and current != EXCLUDED: + return UNKNOWN + return current + + +def _array_tail(children: dict[str, Any]) -> Any: + """The schema governing array positions past the explicit tuple entries: the + Draft-07 ``additionalItems`` for a tuple ``items`` array, otherwise ``items`` + (a single ``items`` schema, or the 2020-12 tail after ``prefixItems``).""" + if any(key.startswith("items/") for key in children): + return children.get("additionalItems", {}) + return children.get("items", {}) + + +def _child_at(children: dict[str, Any], key: str) -> Any: + """The effective subschema at ``key``. An array location the schema does not + declare explicitly is governed by its tail schema, so each array index is + compared against the constraint that actually applies there. + ``additionalItems`` always resolves to the tail, as it is ignored outside + the tuple form.""" + if key == "additionalItems": + return _array_tail(children) + if key in children: + return children[key] + if key == "items" or key.startswith(("items/", "prefixItems/")): + return _array_tail(children) + return {} + + +def _schema_ref_relation(base: Any, derived: Any, depth: int) -> str: + if depth >= _MAX_DEPTH: + return UNKNOWN + result = ref_scope_relation(base, derived) + if result == EXCLUDED: + return EXCLUDED + base_children = _effective_subschemas(base) + derived_children = _effective_subschemas(derived) + for key in set(base_children) | set(derived_children): + child = _schema_ref_relation( + _child_at(base_children, key), _child_at(derived_children, key), depth + 1 + ) + result = _combine(result, child) + if result == EXCLUDED: + return EXCLUDED + return result + + +def _mentions_ref_keyword(node: Any, want_self_id: bool = False) -> bool: + """Whether ``node`` declares any reference keyword (or, when + ``want_self_id``, specifically a ``/$id`` operand). Keyword-shaped keys in + literal data (``const``/``enum``/``default``/``examples``) are skipped, as + they are not keywords.""" + if isinstance(node, dict): + for keyword in REF_KEYWORDS: + if keyword in node and (not want_self_id or node[keyword] == SELF_ID_REF): + return True + return any( + _mentions_ref_keyword(value, want_self_id) + for key, value in node.items() + if key not in _LITERAL_KEYWORDS + ) + if isinstance(node, list): + return any(_mentions_ref_keyword(item, want_self_id) for item in node) + return False + + +def _with_resolved_self(schema: Any, self_id: str | None) -> Any: + """``schema`` with every ``/$id`` reference operand resolved to ``self_id``. + + Returns the input unchanged (no copy) when there is nothing to resolve, so + the common case avoids a deep copy of the whole schema. + """ + if self_id is None or not _mentions_ref_keyword(schema, want_self_id=True): + return schema + copied = copy.deepcopy(schema) + resolve_self_ref_operands(copied, self_id) + return copied + + +def schema_ref_relation( + base: Any, + derived: Any, + base_self_id: str | None = None, + derived_self_id: str | None = None, +) -> str: + """Three-valued reference-dimension relation ``Valid(derived) ⊆ Valid(base)`` + over two resolved schemas, resolving each schema's ``/$id`` to its own type. + """ + # Only the base's reference keywords restrict: a derived-only constraint can + # only narrow, and a schema with no reference keyword imposes no restriction + # (README §9.6.1). Scanning for a keyword is far cheaper than the deep copy + # and full-tree scope walk the comparison would otherwise do on every OP#8 + # and OP#12 check, so bail out early for the common keyword-free schema. + if not _mentions_ref_keyword(base): + return INCLUDED + base_eval = _with_resolved_self(base, base_self_id or _self_id_of(base)) + derived_eval = _with_resolved_self(derived, derived_self_id or _self_id_of(derived)) + return _schema_ref_relation(base_eval, derived_eval, 0) diff --git a/gts/src/gts/store.py b/gts/src/gts/store.py index d656b79..bfb16bc 100644 --- a/gts/src/gts/store.py +++ b/gts/src/gts/store.py @@ -48,7 +48,12 @@ check_schema_regexes, iter_schema_nodes, ) -from .x_gts_ref import XGtsRefValidator, _without_x_gts_ref, extended_validator_for +from .x_gts_ref import ( + ENTITY_REF_KEYWORDS, + REF_KEYWORDS, + XGtsRefValidator, + extended_validator_for, +) logger = logging.getLogger(__name__) @@ -481,8 +486,12 @@ def _registry_with_schemas( "2019-09": DRAFT201909, "2020-12": DRAFT202012, }[dialect] + # Keep the GTS reference keywords: when a derived type imports a + # base via ``$ref``, the extended validator must still evaluate + # the base's reference constraints (so an inherited ``/$id`` + # resolves to the leaf being validated, ADR-0007/0008). resource = Resource.from_contents( - _without_x_gts_ref(entity.content), + entity.content, default_specification=specification, ) registry = registry.with_resource(with_scheme(entity_id), resource) @@ -690,7 +699,7 @@ def _validate_gts_keywords(content: dict[str, Any]) -> None: "x-gts-traits", "x-gts-traits-schema", } - supported_keywords = top_level_keywords | {"x-gts-ref"} + supported_keywords = top_level_keywords | set(REF_KEYWORDS) # Validate x-gts-final final_val = content.get("x-gts-final") @@ -1268,15 +1277,19 @@ def _schema_dependencies( if not include_gts_refs: continue - x_gts_ref = x_gts_ref_validator.resolve_ref_pattern( - subschema.get("x-gts-ref"), selected_type_id - ) - if ( - isinstance(x_gts_ref, str) - and x_gts_ref.startswith(GTS_PREFIX) - and "*" not in x_gts_ref - ): - yield x_gts_ref, True + # Only the entity-reference keywords establish registry dependencies; + # x-gts-pattern-ref never consults the registry (ADR-0007). A concrete + # operand names a type dependency (the referenced type family root). + for keyword in ENTITY_REF_KEYWORDS: + x_gts_ref = x_gts_ref_validator.resolve_ref_pattern( + subschema.get(keyword), selected_type_id + ) + if ( + isinstance(x_gts_ref, str) + and x_gts_ref.startswith(GTS_PREFIX) + and "*" not in x_gts_ref + ): + yield x_gts_ref, True def _validate_entity_transitive( self, @@ -1713,14 +1726,9 @@ def _matches_id_pattern( True if entity ID matches the pattern """ if is_wildcard and wildcard_pattern: - matched = entity_id.wildcard_match(wildcard_pattern) - if not matched: - return False - if base_pattern.endswith("~*"): - base_depth = max(0, len(wildcard_pattern.gts_id_segments) - 1) - if len(entity_id.gts_id_segments) <= base_depth: - return False - return True + # §10 matching (shared with OP#4) fully decides selection, including + # that a `~*` chain wildcard also selects the base type itself. + return entity_id.wildcard_match(wildcard_pattern) # For non-wildcard patterns, use wildcard_match to support version flexibility # This allows patterns like "gts.x.test.v1~" to match "gts.x.test.v1.0~" diff --git a/gts/src/gts/traits.py b/gts/src/gts/traits.py index f08145b..dca23fd 100644 --- a/gts/src/gts/traits.py +++ b/gts/src/gts/traits.py @@ -438,11 +438,11 @@ def _validate_trait_values( for err in xref.validate_schema_ref_existence( effective_traits_schema, selected_type_id=selected_type_id ): - errors.append(f"trait x-gts-ref: {err.reason}") + errors.append(f"trait reference: {err.reason}") for err in xref.validate_instance( effective_traits, effective_traits_schema, selected_type_id=selected_type_id, ): - errors.append(f"trait x-gts-ref: {err.reason}") + errors.append(f"trait reference: {err.reason}") return errors diff --git a/gts/src/gts/x_gts_ref.py b/gts/src/gts/x_gts_ref.py index fb736bd..cecfa9e 100644 --- a/gts/src/gts/x_gts_ref.py +++ b/gts/src/gts/x_gts_ref.py @@ -1,25 +1,40 @@ -""" -x-gts-ref validation support for GTS schemas (optimized version). - -This module implements validation for the x-gts-ref extension as specified -in the GTS specification section 9.5. - -Key optimizations: -1. Resolve local JSON Schema $ref pointers during traversal -2. Consolidate duplicate validation logic -3. Simplify recursive traversal with a generic walker +"""GTS entity- and pattern-reference keyword validation (README §9.6; ADR-0007/0008). + +This module validates the three GTS reference keywords: + +- ``x-gts-type-ref``: the candidate MUST be a GTS Type Identifier selected by + the operand. +- ``x-gts-instance-ref``: the candidate MUST be a GTS Instance Identifier + selected by the operand. +- ``x-gts-pattern-ref``: the candidate MUST be a GTS identifier *pattern* whose + whole selection set is covered by the operand. + +The former kind-blind ``x-gts-ref`` keyword was removed in GTS 0.16; it is now +handled like any other unknown ``x-gts-*`` keyword (rejected on schema +validation). + +Operand (schema) validity and candidate (value) validity are two concerns: an +invalid operand is a schema error (reported by :meth:`XGtsRefValidator.validate_schema` +and not invertible by ``not``); a value that does not satisfy the constraint is +an ordinary validation failure that composition keywords may invert. Registry +lookup applies only to the two entity-reference keywords; ``x-gts-pattern-ref`` +never consults the registry. + +Matching and coverage come from :mod:`gts.gts_id_pattern` (README §10), so they +agree with OP#4 point matching and pattern coverage. """ from __future__ import annotations +import functools from collections.abc import Mapping from typing import Any from jsonschema.exceptions import ValidationError +from . import gts_id_pattern as idp from ._json_pointer import resolve as resolve_json_pointer from ._naming import GTS_PREFIX, strip_scheme -from .gts import GtsID, GtsWildcard from .gts_ref_validation import GtsRefValidationMode from .schema_dialect import effective_dialect from .schema_validation import ( @@ -31,103 +46,183 @@ validator_for, ) -X_GTS_REF_SELF = "/$id" +X_GTS_TYPE_REF = "x-gts-type-ref" +X_GTS_INSTANCE_REF = "x-gts-instance-ref" +X_GTS_PATTERN_REF = "x-gts-pattern-ref" + +#: The reserved ``/$id`` operand: the leaf GTS Type Schema selected for +#: validation, even where the constraint is inherited (README §9.6.1). +SELF_ID_REF = "/$id" +X_GTS_REF_SELF = SELF_ID_REF # backward-compatible alias + +#: Entity-reference keywords (kind-constrained, registry-aware). +ENTITY_REF_KEYWORDS = (X_GTS_TYPE_REF, X_GTS_INSTANCE_REF) +#: All three GTS reference keywords, in a stable order. +REF_KEYWORDS = (X_GTS_TYPE_REF, X_GTS_INSTANCE_REF, X_GTS_PATTERN_REF) + + +def _noun(keyword: str) -> str: + return ( + "GTS Type Identifier" + if keyword == X_GTS_TYPE_REF + else "GTS Instance Identifier" + ) -def _gts_pattern_violation(value: str, pattern: str) -> str | None: - """Return why the string ``value`` does not satisfy ``pattern``, or None. +def _other_noun(keyword: str) -> str: + return ( + "GTS Instance Identifier" + if keyword == X_GTS_TYPE_REF + else "GTS Type Identifier" + ) - Pure pattern matching (concrete id, type prefix, or single trailing - wildcard). Registry existence and the ``/$id`` self-reference are NOT decided - here — they need the store and the selected-type context, which the - :class:`XGtsRefValidator` walker owns. This helper is shared by that walker - and by the structural ``x-gts-ref`` keyword so both agree on matching. + +class _OperandError(Exception): + """An operand (schema value) of a reference keyword is not usable.""" + + +def resolve_ref_operand( + keyword: str, operand: Any, selected_type_id: str | None +) -> tuple[idp.ParsedPattern, str]: + """Resolve a keyword operand to its pattern and spelling (``/$id`` resolved). + + Raises :class:`_OperandError` when the operand is a schema error. """ - if not GtsID.is_valid(value): - return f"Value '{value}' is not a valid GTS identifier" - if pattern == GTS_PREFIX + "*": - return None - if pattern.endswith("*"): - if not value.startswith(pattern[:-1]): - return f"Value '{value}' does not match pattern '{pattern}'" - return None - # Prefix matching alone ignores segment boundaries: an exact constraint such - # as "gts.a.b.c.d.v1~x.y.z.w.v1" would otherwise also accept "…w.v12"/"…w.v1.5". - # Type patterns (ending with '~') admit derived identifiers; any other (exact) - # pattern requires a full match or a '~' boundary right after the pattern. - if not value.startswith(pattern) or not ( - len(value) == len(pattern) - or pattern.endswith("~") - or value[len(pattern)] == "~" + if not isinstance(operand, str): + raise _OperandError( + f"{keyword} operand must be a string, got {type(operand).__name__}" + ) + + if operand == SELF_ID_REF: + if not selected_type_id: + raise _OperandError( + f"{keyword} operand '{SELF_ID_REF}' has no selected GTS Type " + "Schema '$id' to resolve to" + ) + try: + parsed = idp.parse_concrete(selected_type_id) + except idp.InvalidPattern: + parsed = None # type: ignore[assignment] + if parsed is None or not parsed.is_type(): + raise _OperandError( + f"{keyword} operand '{SELF_ID_REF}' resolves to '{selected_type_id}', " + "which is not a concrete GTS Type Identifier" + ) + return parsed, selected_type_id + + try: + pattern = idp.parse_pattern(operand) + except idp.InvalidPattern as error: + raise _OperandError( + f"{keyword} operand '{operand}': {error}; expected a GTS identifier, " + f"a wildcard pattern or '{SELF_ID_REF}'" + ) from error + + # x-gts-type-ref rejects a concrete Instance ID operand: it selects no types. + if ( + keyword == X_GTS_TYPE_REF + and not idp.has_wildcard(operand) + and not idp.is_type_id(operand) ): - return f"Value '{value}' does not match pattern '{pattern}'" - return None + raise _OperandError( + f"{keyword} operand '{operand}' is a GTS Instance Identifier, " + "which selects no GTS Type Identifiers" + ) + return pattern, operand -def _make_x_gts_ref_keyword(selected_type_id: str | None): - """Build a ``jsonschema`` keyword handler that makes ``x-gts-ref`` a - first-class assertion, so ``oneOf``/``anyOf``/``allOf`` resolve correctly: - two branches that differ only by ``x-gts-ref`` are genuinely different - schemas rather than identical match-all schemas. This is the same design - gts-go and gts-rust use (a registered keyword/vocabulary) and removes the - need to strip x-gts-ref and rewrite ``oneOf``→``anyOf``. +def ref_candidate_violation( + keyword: str, value: Any, operand: idp.ParsedPattern, spelling: str +) -> str | None: + """Return why ``value`` violates the resolved reference constraint, or None. - ``selected_type_id`` (the type being validated) lets the ``/$id`` - self-reference resolve here so a ``/$id`` branch matches only that type - instead of matching unconditionally; without it, ``/$id`` is deferred to - XGtsRefValidator. Registry existence always stays with XGtsRefValidator. + Pure structural check (kind + matching, or pattern coverage); no registry + lookup. """ + if not isinstance(value, str): + return f"Value must be a string, got {type(value).__name__}" + + if keyword == X_GTS_PATTERN_REF: + try: + value_pattern = idp.parse_pattern(value) + except idp.InvalidPattern as error: + return f"'{value}' is not a valid GTS identifier pattern: {error}" + if not idp.covers(operand, value_pattern): + return f"'{value}' selects GTS identifiers outside '{spelling}'" + return None - def _keyword(validator, ref_pattern, instance, schema): - if not isinstance(ref_pattern, str): - return + if idp.has_wildcard(value) or value.strip() != value: + return f"Value '{value}' is not a valid GTS identifier" + try: + candidate = idp.parse_concrete(value) + except idp.InvalidPattern: + return f"Value '{value}' is not a valid GTS identifier" + if (keyword == X_GTS_TYPE_REF) != candidate.is_type(): + return ( + f"Value '{value}' is a {_other_noun(keyword)}, but {keyword} " + f"requires a {_noun(keyword)}" + ) + if not idp.matches_pattern(candidate, operand): + return f"Value '{value}' does not match pattern '{spelling}'" + return None + + +def _make_ref_keyword(keyword: str, selected_type_id: str | None): + """Build a ``jsonschema`` keyword handler so the reference keyword is a + first-class assertion (``oneOf``/``anyOf``/``allOf``/``not`` resolve + correctly). ``selected_type_id`` lets ``/$id`` resolve here; an unresolvable + operand is deferred (no error) and caught by :meth:`XGtsRefValidator.validate_schema`. + """ + + def _keyword(validator, operand, instance, schema): if not isinstance(instance, str): return - pattern = ref_pattern - if ref_pattern == X_GTS_REF_SELF: - if not selected_type_id: - return - pattern = selected_type_id - reason = _gts_pattern_violation(instance, strip_scheme(pattern)) + try: + pattern, spelling = resolve_ref_operand(keyword, operand, selected_type_id) + except _OperandError: + return + reason = ref_candidate_violation(keyword, instance, pattern, spelling) if reason is not None: yield ValidationError(reason) return _keyword -# Backwards-compatible symbol: the /$id-deferring keyword (no selected type). -_x_gts_ref_keyword = _make_x_gts_ref_keyword(None) - +@functools.lru_cache(maxsize=256) +def _ref_keyword_family(selected_type_id: str | None) -> Mapping[str, type]: + """Build (and cache) the safe validator family that evaluates the three GTS + reference keywords for the given selected leaf type. -_EXTENDED_FAMILIES: dict[str | None, Mapping[str, type]] = {} + The ``/$id`` operand is baked into the keyword handlers, so a distinct family + is needed per ``selected_type_id``. An LRU bound keeps the cache from growing + without limit on a long-running server that validates many distinct types. + """ + return validator_family( + { + keyword: _make_ref_keyword(keyword, selected_type_id) + for keyword in REF_KEYWORDS + } + ) def extended_validator_for(schema: Any, selected_type_id: str | None = None) -> type: """Return the ``jsonschema`` validator class for ``schema``'s dialect, - extended so ``x-gts-ref`` is evaluated as a real keyword during structural - validation (including inside combinators). ``selected_type_id`` is threaded - into the keyword so ``/$id`` resolves during combinator resolution. - - The safe family preserves regex and ``x-gts-ref`` keywords across ``$schema`` changes.""" - family = _EXTENDED_FAMILIES.get(selected_type_id) - if family is None: - family = validator_family( - {"x-gts-ref": _make_x_gts_ref_keyword(selected_type_id)} - ) - _EXTENDED_FAMILIES[selected_type_id] = family - return validator_for(schema, family=family) + extended so the three GTS reference keywords are evaluated as real keywords + during structural validation (including inside combinators). + """ + return validator_for(schema, family=_ref_keyword_family(selected_type_id)) def _without_x_gts_ref(schema: Any) -> Any: - # Plain removal of x-gts-ref for the few places that deliberately need the - # bare structural shape: the walker's per-branch structural check - # (_is_structurally_valid) and the $ref reference registry. Structural - # instance validation does NOT use this — it uses extended_validator_for so - # the engine evaluates x-gts-ref natively (see _x_gts_ref_keyword). + """Remove the GTS reference keywords for the few places that need the bare + structural shape (per-branch structural checks and the ``$ref`` registry).""" + def strip(node: Any) -> Any: if not isinstance(node, dict): return node - stripped = {key: value for key, value in node.items() if key != "x-gts-ref"} + stripped = { + key: value for key, value in node.items() if key not in REF_KEYWORDS + } for keyword in ("oneOf", "anyOf", "allOf"): branches = stripped.get(keyword) if ( @@ -165,11 +260,11 @@ def _is_structurally_valid(instance: Any, schema: Any, dialect: str) -> bool: class XGtsRefValidationError(Exception): - """Exception raised when x-gts-ref validation fails.""" + """Exception raised when GTS reference validation fails.""" def __init__(self, field_path: str, value: Any, ref_pattern: str, reason: str): super().__init__( - f"x-gts-ref validation failed for field '{field_path}': {reason}" + f"GTS reference validation failed for field '{field_path}': {reason}" ) self.field_path = field_path self.value = value @@ -178,7 +273,7 @@ def __init__(self, field_path: str, value: Any, ref_pattern: str, reason: str): class XGtsRefValidator: - """Validator for x-gts-ref constraints in GTS schemas.""" + """Validator for the GTS reference keywords in GTS schemas.""" def __init__( self, @@ -206,7 +301,7 @@ def __init__( @staticmethod def is_self_reference(value: Any) -> bool: - return value == X_GTS_REF_SELF + return value == SELF_ID_REF @staticmethod def selected_type_id( @@ -222,17 +317,7 @@ def validate_instance( instance_path: str = "", selected_type_id: str | None = None, ) -> list[XGtsRefValidationError]: - """ - Validate an instance against x-gts-ref constraints in schema. - - Args: - instance: The data instance to validate - schema: The JSON schema with x-gts-ref extensions - instance_path: Current path in instance (for error reporting) - - Returns: - List of validation errors (empty if valid) - """ + """Validate an instance against the reference constraints in ``schema``.""" errors: list[XGtsRefValidationError] = [] dialect = effective_dialect(schema, self.inherited_dialect) selected_type_id = self.selected_type_id(schema, selected_type_id) @@ -243,7 +328,6 @@ def resolve_local_ref(ref: str) -> Any | 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 @@ -254,12 +338,14 @@ def visit_instance(inst, sch, path, errs, refs=None): 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, selected_type_id - ) - if error: - errs.append(error) + if isinstance(inst, str): + for keyword in REF_KEYWORDS: + if keyword in sch: + error = self._validate_ref_value( + keyword, inst, sch[keyword], path, selected_type_id + ) + if error: + errs.append(error) one_of = sch.get("oneOf") if isinstance(one_of, list): @@ -372,15 +458,26 @@ def _validate_branch(inst, branch, path): def validate_schema( self, schema: dict[str, Any], schema_path: str = "" ) -> list[XGtsRefValidationError]: - """Validate x-gts-ref fields in a schema definition.""" + """Validate the reference-keyword operands in a schema definition.""" errors = [] - for subschema, path in iter_schema_nodes(schema, schema_path): - if "x-gts-ref" not in subschema: - continue - ref_path = f"{path}/x-gts-ref" if path else "x-gts-ref" - error = self._validate_ref_pattern(subschema["x-gts-ref"], ref_path) - if error: - errors.append(error) + selected_type_id = self.selected_type_id(schema, None) + for subschema, path in iter_schema_nodes(schema): + for keyword in REF_KEYWORDS: + if keyword not in subschema: + continue + ref_path = f"{path}/{keyword}" if path else keyword + try: + resolve_ref_operand(keyword, subschema[keyword], selected_type_id) + except _OperandError as error: + operand = subschema[keyword] + errors.append( + XGtsRefValidationError( + ref_path, + operand, + operand if isinstance(operand, str) else "", + str(error), + ) + ) return errors def validate_schema_ref_existence( @@ -389,52 +486,54 @@ def validate_schema_ref_existence( schema_path: str = "", selected_type_id: str | None = None, ) -> list[XGtsRefValidationError]: + """Check that each entity-reference operand resolves to a registered + (and, under any-valid, valid) entity. Pattern-reference operands are + never looked up (ADR-0007).""" if self.store is None or self.mode == GtsRefValidationMode.NONE: return [] store = self.store selected_type_id = self.selected_type_id(schema, selected_type_id) - def matches(pattern: str) -> list[str]: - wildcard = GtsWildcard(pattern) - result = [] + def matches(pattern: str) -> bool: for entity_id, _ in store.items(): # noqa: PERF102 - generic store protocol - try: - if GtsID(entity_id).wildcard_match(wildcard): - result.append(entity_id) - except ValueError: - continue - return result + matched, error = idp.match(entity_id, pattern) + if error is None and matched: + return True + return False errors: list[XGtsRefValidationError] = [] for subschema, path in iter_schema_nodes(schema, schema_path): - ref_pattern = subschema.get("x-gts-ref") - resolved = self.resolve_ref_pattern(ref_pattern, selected_type_id) - if not isinstance(resolved, str) or not resolved.startswith(GTS_PREFIX): - continue - ref_path = f"{path}/x-gts-ref" if path else "x-gts-ref" - if "*" in resolved: - if matches(resolved): - self.referenced_wildcard_patterns.add(resolved) - else: + for keyword in ENTITY_REF_KEYWORDS: + operand = subschema.get(keyword) + if not isinstance(operand, str): + continue + resolved = selected_type_id if operand == SELF_ID_REF else operand + if not isinstance(resolved, str) or not resolved.startswith(GTS_PREFIX): + continue + ref_path = f"{path}/{keyword}" if path else keyword + if "*" in resolved: + if matches(resolved): + self.referenced_wildcard_patterns.add(resolved) + else: + errors.append( + XGtsRefValidationError( + ref_path, + operand, + resolved, + f"{keyword} wildcard constraint '{resolved}' has no registered match", + ) + ) + elif store.get(resolved) is None: errors.append( XGtsRefValidationError( ref_path, - ref_pattern, + operand, resolved, - f"x-gts-ref wildcard constraint '{resolved}' has no registered match", + f"{keyword} constraint '{resolved}' is not registered", ) ) - elif store.get(resolved) is None: - errors.append( - XGtsRefValidationError( - ref_path, - ref_pattern, - resolved, - f"x-gts-ref constraint type '{resolved}' is not registered", - ) - ) - else: - self.referenced_ids.add(resolved) + else: + self.referenced_ids.add(resolved) return errors def resolve_ref_pattern( @@ -448,134 +547,41 @@ def resolve_ref_pattern( def _validate_ref_value( self, + keyword: str, value: str, - ref_pattern: str, + operand: Any, field_path: str, selected_type_id: str | None, ) -> XGtsRefValidationError | None: - """ - Validate an instance value against its x-gts-ref constraint. - - Args: - value: The field value to validate - ref_pattern: The x-gts-ref pattern - field_path: Path to the field (for error reporting) - selected_type_id: Canonical identifier of the selected leaf type - - Returns: - XGtsRefValidationError if validation fails, None otherwise - """ - if not isinstance(value, str): + """Validate an instance value against a reference constraint, including + registry existence for the entity-reference keywords.""" + try: + pattern, spelling = resolve_ref_operand(keyword, operand, selected_type_id) + except _OperandError as error: return XGtsRefValidationError( field_path, value, - ref_pattern, - f"Value must be a string, got {type(value).__name__}", + operand if isinstance(operand, str) else "", + str(error), ) - if self.is_self_reference(ref_pattern): - if selected_type_id is None: - return XGtsRefValidationError( - field_path, - value, - ref_pattern, - "Cannot resolve /$id without a selected GTS Type Schema", - ) - ref_pattern = selected_type_id - - # Validate against GTS pattern - return self._validate_gts_pattern(value, ref_pattern, field_path) - - def _validate_ref_pattern( - self, ref_pattern: str, field_path: str - ) -> XGtsRefValidationError | None: - """ - Validate an x-gts-ref pattern in a schema definition. - - Args: - ref_pattern: The x-gts-ref value - field_path: Path to the field (for error reporting) - - Returns: - XGtsRefValidationError if validation fails, None otherwise - """ - if not isinstance(ref_pattern, str): - return XGtsRefValidationError( - field_path, - ref_pattern, - "", - f"x-gts-ref value must be a string, got {type(ref_pattern).__name__}", - ) - - # Case 1: Absolute GTS pattern - if ref_pattern.startswith(GTS_PREFIX): - return self._validate_gts_id_or_pattern(ref_pattern, field_path) - - if self.is_self_reference(ref_pattern): - return None - - return XGtsRefValidationError( - field_path, - ref_pattern, - ref_pattern, - f"Invalid x-gts-ref value: '{ref_pattern}' must be a GTS identifier, wildcard, or '{X_GTS_REF_SELF}'", - ) - - def _validate_gts_id_or_pattern( - self, pattern: str, field_path: str - ) -> XGtsRefValidationError | None: - """Validate a GTS ID or pattern in schema definition.""" - if pattern == GTS_PREFIX + "*": - return None # Valid wildcard - - if "*" in pattern: - # Wildcard pattern - validate prefix - prefix = pattern.rstrip("*") - if not prefix.startswith(GTS_PREFIX): - return XGtsRefValidationError( - field_path, - pattern, - pattern, - f"Invalid GTS wildcard pattern: {pattern}", - ) - return None - - # Specific GTS ID - if not GtsID.is_valid(pattern): - return XGtsRefValidationError( - field_path, pattern, pattern, f"Invalid GTS identifier: {pattern}" - ) - return None - - def _validate_gts_pattern( - self, value: str, pattern: str, field_path: str - ) -> XGtsRefValidationError | None: - """ - Validate value matches a GTS pattern. - - Args: - value: The value to validate - pattern: GTS pattern (e.g., "gts.*", "gts.x.core.modules.*") - field_path: Path to field (for error reporting) - - Returns: - Error if validation fails, None otherwise - """ - # Shared pattern matching (also used by the structural x-gts-ref keyword). - reason = _gts_pattern_violation(value, pattern) + reason = ref_candidate_violation(keyword, value, pattern, spelling) if reason is not None: - return XGtsRefValidationError(field_path, value, pattern, reason) - - # Referenced values use exact registry lookup in presence/full modes. - if self.store and self.mode != GtsRefValidationMode.NONE: - entity = self.store.get(value) - if not entity: + return XGtsRefValidationError(field_path, value, spelling, reason) + + # Entity references may additionally require the candidate to be + # registered. Pattern references never consult the registry. + if ( + keyword in ENTITY_REF_KEYWORDS + and self.store is not None + and self.mode != GtsRefValidationMode.NONE + ): + if self.store.get(value) is None: return XGtsRefValidationError( field_path, value, - pattern, + spelling, f"Referenced entity '{value}' not found in registry", ) self.referenced_ids.add(value) - return None diff --git a/tests/test_ops.py b/tests/test_ops.py index b4b42ca..faae1bd 100644 --- a/tests/test_ops.py +++ b/tests/test_ops.py @@ -147,15 +147,15 @@ def test_add_entities_batch(self, ops): assert result.ok is True assert len(result.results) == 2 - def test_add_entities_rejects_unsupported_x_gts_ref_pointer(self, ops): + def test_add_entities_rejects_unsupported_ref_pointer(self, ops): schema = { "$schema": "http://json-schema.org/draft-07/schema#", "$id": "gts://gts.x.test._.relative.v1~", - "properties": {"ref": {"x-gts-ref": "/missing"}}, + "properties": {"ref": {"x-gts-type-ref": "/missing"}}, } result = ops.add_entities([schema]) assert result.ok is False - assert "must be a GTS identifier" in result.results[0].error + assert "expected a GTS identifier" in result.results[0].error class TestAddSchema: diff --git a/tests/test_ref_scope_compat.py b/tests/test_ref_scope_compat.py new file mode 100644 index 0000000..e97fc4d --- /dev/null +++ b/tests/test_ref_scope_compat.py @@ -0,0 +1,204 @@ +"""Regression tests for gts.ref_scope_compat.schema_ref_relation (README §9.6.1). + +These cover reference-scope folding cases the gts-spec conformance suite does +not exercise (it only places reference keywords on top-level properties). +""" + +from typing import ClassVar + +from gts import ref_scope_compat as rsc + +TYPE_REF = "x-gts-type-ref" + + +def _rel(base, derived): + return rsc.schema_ref_relation(base, derived) + + +class TestInheritedAllOfReference: + """A reference nested in an inherited ``allOf`` must survive a later overlay + that redeclares the property with ordinary keywords.""" + + def test_overlay_does_not_widen_inherited_allof_reference(self): + base = { + "allOf": [ + {"properties": {"link": {"allOf": [{TYPE_REF: "gts.a.*"}]}}}, + {"properties": {"link": {"type": "string"}}}, + ] + } + derived = {"properties": {"link": {"type": "string"}}} + # derived accepts any string; base restricts link to the gts.a.* family. + assert _rel(base, derived) == rsc.EXCLUDED + + def test_narrower_derived_under_inherited_allof_is_included(self): + base = { + "allOf": [ + {"properties": {"link": {"allOf": [{TYPE_REF: "gts.a.*"}]}}}, + {"properties": {"link": {"type": "string"}}}, + ] + } + derived = {"properties": {"link": {TYPE_REF: "gts.a.b.*"}}} + assert _rel(base, derived) == rsc.INCLUDED + + +class TestReferenceUnderItems: + """Reference widening under array ``items`` must be detected.""" + + def test_items_reference_widening_is_excluded(self): + base = { + "type": "array", + "items": {"type": "string", TYPE_REF: "gts.x.test.*"}, + } + derived = {"type": "array", "items": {"type": "string", TYPE_REF: "gts.*"}} + assert _rel(base, derived) == rsc.EXCLUDED + + def test_items_reference_narrowing_is_included(self): + base = {"type": "array", "items": {"type": "string", TYPE_REF: "gts.x.*"}} + derived = { + "type": "array", + "items": {"type": "string", TYPE_REF: "gts.x.test.*"}, + } + assert _rel(base, derived) == rsc.INCLUDED + + def test_items_reference_dropped_by_derived_is_excluded(self): + base = {"type": "array", "items": {"type": "string", TYPE_REF: "gts.x.test.*"}} + derived = {"type": "array", "items": {"type": "string"}} + assert _rel(base, derived) == rsc.EXCLUDED + + def test_tuple_items_reference_widening_is_excluded(self): + base = { + "type": "array", + "items": [{"type": "string", TYPE_REF: "gts.x.test.*"}], + } + derived = {"type": "array", "items": [{"type": "string", TYPE_REF: "gts.*"}]} + assert _rel(base, derived) == rsc.EXCLUDED + + +class TestTupleAdditionalItemsScope: + """A tuple position the base leaves to ``additionalItems`` (or the 2020-12 + ``items`` tail) must be compared against that tail scope.""" + + BASE: ClassVar[dict] = { + "type": "array", + "items": [{"type": "string", TYPE_REF: "gts.x.a.*"}], + "additionalItems": {"type": "string", TYPE_REF: "gts.x.test.*"}, + } + + def test_unrestricted_new_tuple_entry_is_excluded(self): + derived = { + "type": "array", + "items": [ + {"type": "string", TYPE_REF: "gts.x.a.*"}, + {"type": "string"}, + ], + "additionalItems": {"type": "string", TYPE_REF: "gts.x.test.*"}, + } + assert _rel(self.BASE, derived) == rsc.EXCLUDED + + def test_narrower_new_tuple_entry_is_included(self): + derived = { + "type": "array", + "items": [ + {"type": "string", TYPE_REF: "gts.x.a.*"}, + {"type": "string", TYPE_REF: "gts.x.test.b.*"}, + ], + "additionalItems": {"type": "string", TYPE_REF: "gts.x.test.*"}, + } + assert _rel(self.BASE, derived) == rsc.INCLUDED + + def test_shorter_derived_tuple_uses_its_additional_items(self): + derived = { + "type": "array", + "items": [], + "additionalItems": {"type": "string", TYPE_REF: "gts.x.test.*"}, + } + # Index 0 is now governed by derived additionalItems (gts.x.test.*), + # which escapes the base's items[0] scope (gts.x.a.*). + assert _rel(self.BASE, derived) == rsc.EXCLUDED + + def test_prefix_items_new_entry_checked_against_items_tail(self): + base = { + "type": "array", + "prefixItems": [{"type": "string", TYPE_REF: "gts.x.a.*"}], + "items": {"type": "string", TYPE_REF: "gts.x.test.*"}, + } + derived = { + "type": "array", + "prefixItems": [ + {"type": "string", TYPE_REF: "gts.x.a.*"}, + {"type": "string"}, + ], + "items": {"type": "string", TYPE_REF: "gts.x.test.*"}, + } + assert _rel(base, derived) == rsc.EXCLUDED + + def test_single_items_base_against_tuple_derived(self): + base = {"type": "array", "items": {"type": "string", TYPE_REF: "gts.x.*"}} + narrowed = { + "type": "array", + "items": [{"type": "string", TYPE_REF: "gts.x.a.*"}], + "additionalItems": {"type": "string", TYPE_REF: "gts.x.b.*"}, + } + assert _rel(base, narrowed) == rsc.INCLUDED + open_tail = { + "type": "array", + "items": [{"type": "string", TYPE_REF: "gts.x.a.*"}], + } + assert _rel(base, open_tail) == rsc.EXCLUDED + + +class TestInheritedUnionReference: + """A reference-bearing anyOf/oneOf in an earlier allOf branch must survive a + later redeclaration of the property.""" + + def _base(self, applicator): + return { + "allOf": [ + { + "properties": { + "link": { + applicator: [{TYPE_REF: "gts.a.*"}, {TYPE_REF: "gts.b.*"}] + } + } + }, + {"properties": {"link": {"type": "string"}}}, + ] + } + + def test_overlay_does_not_widen_inherited_union(self): + derived = {"properties": {"link": {"type": "string"}}} + for applicator in ("anyOf", "oneOf"): + assert _rel(self._base(applicator), derived) == rsc.EXCLUDED, applicator + + def test_narrower_derived_under_inherited_union_is_included(self): + derived = {"properties": {"link": {TYPE_REF: "gts.b.x.*"}}} + for applicator in ("anyOf", "oneOf"): + assert _rel(self._base(applicator), derived) == rsc.INCLUDED, applicator + + def test_derived_redeclaring_union_keeps_inherited_conjunct(self): + base = {"properties": {"link": {TYPE_REF: "gts.a.*"}}} + derived = { + "allOf": [ + {"properties": {"link": {"anyOf": [{TYPE_REF: "gts.a.x.*"}]}}}, + {"properties": {"link": {"anyOf": [{TYPE_REF: "gts.*"}]}}}, + ] + } + # Effective derived link is gts.a.x.* ∩ gts.*, which lies within gts.a.*. + assert _rel(base, derived) == rsc.INCLUDED + + def test_inherited_not_reference_is_preserved(self): + base = { + "allOf": [ + {"properties": {"link": {"not": {TYPE_REF: "gts.a.*"}}}}, + {"properties": {"link": {"type": "string"}}}, + ] + } + derived = {"properties": {"link": {"type": "string"}}} + assert _rel(base, derived) == rsc.EXCLUDED + + +class TestNoReferenceIsIncluded: + def test_schemas_without_reference_keywords_are_included(self): + base = {"type": "array", "items": {"type": "string"}} + derived = {"type": "array", "items": {"type": "string", "minLength": 1}} + assert _rel(base, derived) == rsc.INCLUDED diff --git a/tests/test_regressions.py b/tests/test_regressions.py index 49757d2..98a87ca 100644 --- a/tests/test_regressions.py +++ b/tests/test_regressions.py @@ -31,21 +31,28 @@ def test_plain_one_of_is_left_to_jsonschema(self): assert errors == [] - def test_x_gts_ref_only_one_of_still_requires_one_match(self): + def test_ref_only_one_of_still_requires_one_match(self): schema = { "oneOf": [ - {"x-gts-ref": "gts.x.test._.first.v1~"}, - {"x-gts-ref": "gts.x.test._.second.v1~"}, + {"x-gts-instance-ref": "gts.x.test._.first.v1~"}, + {"x-gts-instance-ref": "gts.x.test._.second.v1~"}, ] } valid = "gts.x.test._.first.v1~x.test._.item.v1" invalid = "gts.x.test._.other.v1~x.test._.item.v1" - assert XGtsRefValidator(inherited_dialect="draft-07").validate_instance(valid, schema) == [] + assert ( + XGtsRefValidator(inherited_dialect="draft-07").validate_instance( + valid, schema + ) + == [] + ) assert [ error.reason - for error in XGtsRefValidator(inherited_dialect="draft-07").validate_instance(invalid, schema) + for error in XGtsRefValidator( + inherited_dialect="draft-07" + ).validate_instance(invalid, schema) ] == ["oneOf: no branch matched"] diff --git a/tests/test_safe_regex.py b/tests/test_safe_regex.py index 28f6e7e..932a0bd 100644 --- a/tests/test_safe_regex.py +++ b/tests/test_safe_regex.py @@ -907,7 +907,10 @@ def test_operational_failure_in_wildcard_target_fails_validation(monkeypatch): { "type": "object", "properties": { - "ref": {"type": "string", "x-gts-ref": "gts.x.re2._.target.v1~*"} + "ref": { + "type": "string", + "x-gts-instance-ref": "gts.x.re2._.target.v1~*", + } }, }, # Draft-07's meta-schema has no patterns, so the failure first occurs diff --git a/tests/test_store_extra.py b/tests/test_store_extra.py index be557bc..43eddac 100644 --- a/tests/test_store_extra.py +++ b/tests/test_store_extra.py @@ -255,30 +255,34 @@ def test_missing_derived_gts_ref_target_raises(self): class TestSchemaDependencies: - def test_ignores_x_gts_ref_in_annotation_data(self): + def test_ignores_ref_keyword_in_annotation_data(self): store = GtsStore(reader=None) assert ( list( store._schema_dependencies( - {"const": {"x-gts-ref": "gts.x.test._.missing.v1~"}} + {"const": {"x-gts-type-ref": "gts.x.test._.missing.v1~"}} ) ) == [] ) - def test_finds_constraint_under_property_named_x_gts_ref(self): + def test_finds_constraint_under_property_named_like_keyword(self): store = GtsStore(reader=None) assert list( store._schema_dependencies( - {"properties": {"x-gts-ref": {"x-gts-ref": "gts.x.test._.missing.v1~"}}} + { + "properties": { + "x-gts-type-ref": {"x-gts-type-ref": "gts.x.test._.missing.v1~"} + } + } ) ) == [("gts.x.test._.missing.v1~", True)] - def test_unsupported_x_gts_ref_pointer_is_not_a_dependency(self): + def test_unsupported_ref_pointer_is_not_a_dependency(self): store = GtsStore(reader=None) schema = { "target": "gts.x.test._.target.v1~", - "properties": {"ref": {"x-gts-ref": "/target"}}, + "properties": {"ref": {"x-gts-type-ref": "/target"}}, } assert list(store._schema_dependencies(schema)) == [] @@ -342,11 +346,11 @@ def test_entity_not_schema_raises(self): with pytest.raises(ValueError, match="is not a schema"): 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.*"}) + def test_invalid_ref_operand_raises(self): + schema = _schema_entity("gts.x.test._.foo.v1~", {"x-gts-type-ref": "notgts.*"}) store = GtsStore(reader=None) store.register(schema) - with pytest.raises(Exception, match="x-gts-ref validation failed"): + with pytest.raises(Exception, match="validation failed"): store._validate_schema_x_gts_refs("gts.x.test._.foo.v1~") def test_basic_validation_rejects_unsupported_pointer(self): @@ -355,14 +359,14 @@ def test_basic_validation_rejects_unsupported_pointer(self): { "x-gts-traits-schema": { "properties": { - "ref": {"x-gts-ref": "/x-gts-traits-schema/missingTarget"} + "ref": {"x-gts-type-ref": "/x-gts-traits-schema/missingTarget"} } } }, ) store = GtsStore(reader=None) store.register(schema) - with pytest.raises(Exception, match="must be a GTS identifier"): + with pytest.raises(Exception, match="expected a GTS identifier"): store.validate_schema_basic("gts.x.test._.foo.v1~") diff --git a/tests/test_traits.py b/tests/test_traits.py index c783a3b..2f69a49 100644 --- a/tests/test_traits.py +++ b/tests/test_traits.py @@ -278,7 +278,7 @@ def test_abstract_skips_required_in_nested_dialect_schema(self): effective = build_effective_traits([schema], {}, DRAFT7) assert effective.validate(check_unresolved=False) == [] - def test_abstract_checks_x_gts_ref_constraint_type_existence(self): + def test_abstract_checks_ref_constraint_type_existence(self): class FakeStore: def get(self, value): return None @@ -286,15 +286,14 @@ def get(self, value): schema = { "type": "object", "properties": { - "ref": {"type": "string", "x-gts-ref": "gts.x.test._.foo.v1~"} + "ref": {"type": "string", "x-gts-type-ref": "gts.x.test._.foo.v1~"} }, } errors = build_effective_traits([schema], {}, DRAFT7).validate( check_unresolved=False, reference_store=FakeStore() ) assert any( - "constraint type 'gts.x.test._.foo.v1~' is not registered" in e - for e in errors + "constraint 'gts.x.test._.foo.v1~' is not registered" in e for e in errors ) def test_incompatible_trait_schema_chain_flagged(self): @@ -336,11 +335,11 @@ def test_dialect_applied_to_effective_schema(self): == "https://json-schema.org/draft/2020-12/schema" ) - def test_x_gts_ref_errors_prefixed(self): + def test_ref_errors_prefixed(self): schema = { "type": "object", - "properties": {"ref": {"type": "string", "x-gts-ref": "not-valid"}}, + "properties": {"ref": {"type": "string", "x-gts-type-ref": "not-valid"}}, } effective = build_effective_traits([schema], {"ref": "also-not-valid"}, DRAFT7) errors = effective.validate(check_unresolved=True) - assert any(e.startswith("trait x-gts-ref:") for e in errors) + assert any(e.startswith("trait reference:") for e in errors) diff --git a/tests/test_x_gts_ref.py b/tests/test_x_gts_ref.py index 5ff96e6..52271a7 100644 --- a/tests/test_x_gts_ref.py +++ b/tests/test_x_gts_ref.py @@ -1,105 +1,98 @@ -"""Tests for gts.x_gts_ref (x-gts-ref schema & instance validation, spec sec 9.5).""" +"""Tests for gts.x_gts_ref: the entity- and pattern-reference keywords +(x-gts-type-ref, x-gts-instance-ref, x-gts-pattern-ref; README §9.6, ADR-0007/0008).""" -from gts.x_gts_ref import X_GTS_REF_SELF, XGtsRefValidator +from gts.x_gts_ref import ( + SELF_ID_REF, + X_GTS_INSTANCE_REF, + X_GTS_PATTERN_REF, + X_GTS_TYPE_REF, + XGtsRefValidator, +) + +DRAFT7 = "draft-07" + +_TYPE = "gts.x.test._.foo.v1~" +_INSTANCE = "gts.x.test._.foo.v1~x.test._.bar.v1" + + +def _validator(**kwargs): + return XGtsRefValidator(inherited_dialect=DRAFT7, **kwargs) class TestValidateSchema: - def test_valid_absolute_pattern(self): - errors = XGtsRefValidator(inherited_dialect="draft-07").validate_schema( - {"x-gts-ref": "gts.x.test._.foo.v1~"} - ) - assert errors == [] + def test_valid_type_ref_operand(self): + assert _validator().validate_schema({X_GTS_TYPE_REF: _TYPE}) == [] - def test_valid_wildcard_pattern(self): - errors = XGtsRefValidator(inherited_dialect="draft-07").validate_schema({"x-gts-ref": "gts.*"}) - assert errors == [] + def test_valid_instance_ref_operand(self): + assert _validator().validate_schema({X_GTS_INSTANCE_REF: _TYPE}) == [] - def test_valid_prefix_wildcard(self): - errors = XGtsRefValidator(inherited_dialect="draft-07").validate_schema({"x-gts-ref": "gts.x.test.*"}) - assert errors == [] + def test_valid_pattern_ref_wildcard_operand(self): + assert _validator().validate_schema({X_GTS_PATTERN_REF: "gts.*"}) == [] - def test_invalid_wildcard_prefix_direct(self): - error = XGtsRefValidator(inherited_dialect="draft-07")._validate_gts_id_or_pattern("notgts*", "path") - assert error is not None - assert "Invalid GTS wildcard pattern" in error.reason + def test_valid_pattern_ref_family_operand(self): + assert _validator().validate_schema({X_GTS_PATTERN_REF: "gts.x.test.*"}) == [] - def test_invalid_specific_gts_id(self): - errors = XGtsRefValidator(inherited_dialect="draft-07").validate_schema({"x-gts-ref": "gts.bad id"}) + def test_type_ref_rejects_instance_operand(self): + errors = _validator().validate_schema({X_GTS_TYPE_REF: _INSTANCE}) assert len(errors) == 1 - assert "Invalid GTS identifier" in errors[0].reason + assert "selects no GTS Type Identifiers" in errors[0].reason + + def test_instance_ref_allows_instance_operand(self): + assert _validator().validate_schema({X_GTS_INSTANCE_REF: _INSTANCE}) == [] - def test_non_string_ref_value(self): - errors = XGtsRefValidator(inherited_dialect="draft-07").validate_schema({"x-gts-ref": 123}) + def test_non_string_operand(self): + errors = _validator().validate_schema({X_GTS_TYPE_REF: 123}) assert len(errors) == 1 assert "must be a string" in errors[0].reason - def test_invalid_prefix_value(self): - errors = XGtsRefValidator(inherited_dialect="draft-07").validate_schema({"x-gts-ref": "nope"}) + def test_invalid_operand_value(self): + errors = _validator().validate_schema({X_GTS_PATTERN_REF: "nope"}) assert len(errors) == 1 - assert "must be a GTS identifier" in errors[0].reason + assert "expected a GTS identifier, a wildcard pattern" in errors[0].reason - def test_selected_type_self_reference_is_valid(self): - validator = XGtsRefValidator(inherited_dialect="draft-07") - assert validator.is_self_reference(X_GTS_REF_SELF) - assert validator.validate_schema({"x-gts-ref": X_GTS_REF_SELF}) == [] + def test_self_reference_requires_selected_type(self): + # No $id to resolve /$id against: a schema error. + errors = _validator().validate_schema({X_GTS_TYPE_REF: SELF_ID_REF}) + assert len(errors) == 1 + assert "has no selected GTS Type Schema" in errors[0].reason + + def test_self_reference_valid_with_id(self): + schema = {"$id": "gts://" + _TYPE, X_GTS_TYPE_REF: SELF_ID_REF} + assert _validator().validate_schema(schema) == [] def test_other_pointers_are_invalid(self): - validator = XGtsRefValidator(inherited_dialect="draft-07") for pointer in ("/missing/path", "/properties/id"): - errors = validator.validate_schema({"x-gts-ref": pointer}) + errors = _validator().validate_schema({X_GTS_PATTERN_REF: pointer}) assert len(errors) == 1 - assert "must be a GTS identifier" in errors[0].reason def test_recurses_into_nested_structures(self): - schema = { - "properties": { - "child": {"x-gts-ref": "notgts.*"}, - } - } - errors = XGtsRefValidator(inherited_dialect="draft-07").validate_schema(schema) + schema = {"properties": {"child": {X_GTS_PATTERN_REF: "notgts.*"}}} + errors = _validator().validate_schema(schema) assert len(errors) == 1 - assert "properties/child/x-gts-ref" in errors[0].field_path + assert "properties/child/x-gts-pattern-ref" in errors[0].field_path def test_recurses_into_list_of_dicts(self): - schema = {"allOf": [{"x-gts-ref": "notgts.*"}]} - errors = XGtsRefValidator(inherited_dialect="draft-07").validate_schema(schema) + schema = {"allOf": [{X_GTS_TYPE_REF: "notgts.*"}]} + errors = _validator().validate_schema(schema) assert len(errors) == 1 - assert "allOf[0]/x-gts-ref" in errors[0].field_path + assert "allOf[0]/x-gts-type-ref" in errors[0].field_path - def test_ignores_x_gts_ref_in_annotation_data(self): + def test_ignores_keyword_shaped_literal_data(self): schema = { "properties": { "payload": { - "const": {"x-gts-ref": "gts.x.test._.missing.v1~"}, - "default": {"x-gts-ref": "not-a-gts-id"}, + "const": {X_GTS_TYPE_REF: "gts.x.test._.missing.v1~"}, + "default": {X_GTS_PATTERN_REF: "not-a-gts-id"}, } } } - assert XGtsRefValidator(inherited_dialect="draft-07").validate_schema(schema) == [] + assert _validator().validate_schema(schema) == [] - def test_property_named_x_gts_ref_is_not_a_keyword(self): - schema = { - "properties": { - "x-gts-ref": {"x-gts-ref": "notgts.*"}, - } - } - errors = XGtsRefValidator(inherited_dialect="draft-07").validate_schema(schema) + def test_property_named_like_keyword_is_not_a_keyword(self): + schema = {"properties": {X_GTS_TYPE_REF: {X_GTS_PATTERN_REF: "notgts.*"}}} + errors = _validator().validate_schema(schema) assert len(errors) == 1 - assert errors[0].field_path == "properties/x-gts-ref/x-gts-ref" - - def test_recurses_into_draft3_schema_forms(self): - schema = { - "$schema": "http://json-schema.org/draft-03/schema#", - "extends": {"x-gts-ref": "invalid-extends"}, - "type": ["object", {"x-gts-ref": "invalid-type"}], - "disallow": ["array", {"x-gts-ref": "invalid-disallow"}], - } - errors = XGtsRefValidator(inherited_dialect="draft-07").validate_schema(schema) - assert [error.field_path for error in errors] == [ - "extends/x-gts-ref", - "type[1]/x-gts-ref", - "disallow[1]/x-gts-ref", - ] + assert errors[0].field_path == "properties/x-gts-type-ref/x-gts-pattern-ref" class TestValidateSchemaRefExistence: @@ -108,110 +101,127 @@ class FakeStore: def get(self, value): return None - errors = XGtsRefValidator(store=FakeStore(), inherited_dialect="draft-07").validate_schema_ref_existence( - {"properties": {"ref": {"x-gts-ref": "gts.x.test._.foo.v1~"}}} + errors = _validator(store=FakeStore()).validate_schema_ref_existence( + {"properties": {"ref": {X_GTS_TYPE_REF: _TYPE}}} ) - assert len(errors) == 1 - assert errors[0].field_path == "properties/ref/x-gts-ref" - assert ( - "constraint type 'gts.x.test._.foo.v1~' is not registered" - in errors[0].reason - ) + assert errors[0].field_path == "properties/ref/x-gts-type-ref" + assert "is not registered" in errors[0].reason def test_registered_and_wildcard_constraints_pass(self): class FakeStore: def get(self, value): - return object() if value == "gts.x.test._.foo.v1~" else None + return object() if value == _TYPE else None def items(self): - return [("gts.x.test._.foo.v1~", object())] + return [(_TYPE, object())] - errors = XGtsRefValidator(store=FakeStore(), inherited_dialect="draft-07").validate_schema_ref_existence( + errors = _validator(store=FakeStore()).validate_schema_ref_existence( { "allOf": [ - {"x-gts-ref": "gts.x.test._.foo.v1~"}, - {"x-gts-ref": "gts.x.test.*"}, + {X_GTS_TYPE_REF: _TYPE}, + {X_GTS_INSTANCE_REF: "gts.x.test.*"}, ] } ) + assert errors == [] + + def test_pattern_ref_operand_is_not_looked_up(self): + class FakeStore: + def get(self, value): + return None + + def items(self): + return [] + # x-gts-pattern-ref never consults the registry (ADR-0007). + errors = _validator(store=FakeStore()).validate_schema_ref_existence( + {X_GTS_PATTERN_REF: "gts.x.test._.missing.v1~"} + ) assert errors == [] - def test_annotation_data_does_not_require_constraint_type(self): + def test_keyword_shaped_literal_data_is_not_looked_up(self): class FakeStore: def get(self, value): return None - errors = XGtsRefValidator(store=FakeStore(), inherited_dialect="draft-07").validate_schema_ref_existence( - {"const": {"x-gts-ref": "gts.x.test._.missing.v1~"}} + errors = _validator(store=FakeStore()).validate_schema_ref_existence( + {"const": {X_GTS_TYPE_REF: "gts.x.test._.missing.v1~"}} ) assert errors == [] - def test_selected_type_constraint_uses_explicit_leaf(self): + def test_self_reference_uses_explicit_leaf(self): + leaf = "gts.x.test._.leaf.v1~" + class FakeStore: def get(self, value): - return object() if value == "gts.x.test._.leaf.v1~" else None + return object() if value == leaf else None - errors = XGtsRefValidator(store=FakeStore(), inherited_dialect="draft-07").validate_schema_ref_existence( - {"x-gts-ref": X_GTS_REF_SELF}, - selected_type_id="gts.x.test._.leaf.v1~", + errors = _validator(store=FakeStore()).validate_schema_ref_existence( + {X_GTS_TYPE_REF: SELF_ID_REF}, selected_type_id=leaf ) assert errors == [] class TestValidateInstanceValue: - def test_non_string_instance_value_error(self): - error = XGtsRefValidator(inherited_dialect="draft-07")._validate_ref_value(123, "gts.*", "ref", None) - assert error is not None - assert "Value must be a string" in error.reason + def test_type_ref_accepts_type_candidate(self): + errors = _validator().validate_instance(_TYPE, {X_GTS_TYPE_REF: "gts.x.test.*"}) + assert errors == [] - def test_relative_ref_pattern_resolution_on_instance(self): - schema = { - "$id": "gts.x.test._.foo.v1~", - "type": "object", - "properties": {"ref": {"x-gts-ref": "/$id"}}, - } - errors = XGtsRefValidator(inherited_dialect="draft-07").validate_instance( - {"ref": "gts.x.test._.foo.v1~x.test._.bar.v1"}, schema + def test_type_ref_rejects_instance_candidate(self): + errors = _validator().validate_instance( + _INSTANCE, {X_GTS_TYPE_REF: "gts.x.test.*"} + ) + assert len(errors) == 1 + assert "requires a GTS Type Identifier" in errors[0].reason + + def test_instance_ref_rejects_type_candidate(self): + errors = _validator().validate_instance( + _TYPE, {X_GTS_INSTANCE_REF: "gts.x.test.*"} + ) + assert len(errors) == 1 + assert "requires a GTS Instance Identifier" in errors[0].reason + + def test_pattern_ref_accepts_covered_pattern(self): + errors = _validator().validate_instance( + "gts.x.test._.foo.v1~*", {X_GTS_PATTERN_REF: "gts.x.test.*"} ) assert errors == [] + def test_pattern_ref_rejects_uncovered_pattern(self): + errors = _validator().validate_instance( + "gts.x.other.*", {X_GTS_PATTERN_REF: "gts.x.test.*"} + ) + assert len(errors) == 1 + assert "selects GTS identifiers outside" in errors[0].reason + + def test_pattern_ref_rejects_malformed_value(self): + errors = _validator().validate_instance( + "not a pattern", {X_GTS_PATTERN_REF: "gts.*"} + ) + assert len(errors) == 1 + assert "not a valid GTS identifier pattern" in errors[0].reason + def test_self_reference_uses_explicit_selected_leaf(self): schema = { "$id": "gts.x.test._.base.v1~", "type": "object", - "properties": {"ref": {"x-gts-ref": X_GTS_REF_SELF}}, + "properties": {"ref": {X_GTS_TYPE_REF: SELF_ID_REF}}, } leaf = "gts.x.test._.base.v1~x.test._.leaf.v1~" assert ( - XGtsRefValidator(inherited_dialect="draft-07").validate_instance( - {"ref": leaf}, schema, selected_type_id=leaf - ) + _validator().validate_instance({"ref": leaf}, schema, selected_type_id=leaf) == [] ) - errors = XGtsRefValidator(inherited_dialect="draft-07").validate_instance( + errors = _validator().validate_instance( {"ref": "gts.x.test._.base.v1~"}, schema, selected_type_id=leaf ) assert len(errors) == 1 assert "does not match pattern" in errors[0].reason - def test_wildcard_pattern_matches_prefix(self): - errors = XGtsRefValidator(inherited_dialect="draft-07").validate_instance( - "gts.x.test._.foo.v1~", {"x-gts-ref": "gts.x.test.*"} - ) - assert errors == [] - - def test_wildcard_pattern_mismatch(self): - errors = XGtsRefValidator(inherited_dialect="draft-07").validate_instance( - "gts.x.other._.foo.v1~", {"x-gts-ref": "gts.x.test.*"} - ) - assert len(errors) == 1 - assert "does not match pattern" in errors[0].reason - def test_exact_pattern_mismatch(self): - errors = XGtsRefValidator(inherited_dialect="draft-07").validate_instance( - "gts.x.other._.foo.v1~", {"x-gts-ref": "gts.x.test._.foo.v1~"} + errors = _validator().validate_instance( + "gts.x.other._.foo.v1~", {X_GTS_TYPE_REF: _TYPE} ) assert len(errors) == 1 assert "does not match pattern" in errors[0].reason @@ -221,40 +231,35 @@ class FakeStore: def get(self, value): return None - errors = XGtsRefValidator(store=FakeStore(), inherited_dialect="draft-07").validate_instance( - "gts.x.test._.foo.v1~", {"x-gts-ref": "gts.*"} + errors = _validator(store=FakeStore()).validate_instance( + _TYPE, {X_GTS_TYPE_REF: "gts.*"} ) assert len(errors) == 1 assert "not found in registry" in errors[0].reason - def test_store_lookup_found_entity(self): + def test_pattern_ref_does_not_consult_store(self): class FakeStore: def get(self, value): - return object() + return None - errors = XGtsRefValidator(store=FakeStore(), inherited_dialect="draft-07").validate_instance( - "gts.x.test._.foo.v1~", {"x-gts-ref": "gts.*"} + # A covered pattern value is accepted even though nothing is registered. + errors = _validator(store=FakeStore()).validate_instance( + "gts.x.test._.foo.v1~", {X_GTS_PATTERN_REF: "gts.*"} ) assert errors == [] def test_array_items_recursion(self): - schema = { - "type": "array", - "items": {"x-gts-ref": "gts.x.test.*"}, - } - errors = XGtsRefValidator(inherited_dialect="draft-07").validate_instance(["gts.x.other.v1~"], schema) + schema = {"type": "array", "items": {X_GTS_TYPE_REF: "gts.x.test.*"}} + errors = _validator().validate_instance(["gts.x.other.v1~"], schema) assert len(errors) == 1 def test_tuple_additional_items_recursion(self): schema = { "type": "array", "items": [{"type": "string"}], - "additionalItems": { - "type": "string", - "x-gts-ref": "gts.x.test._.target.v1~", - }, + "additionalItems": {"type": "string", X_GTS_TYPE_REF: "gts.x.test.*"}, } - errors = XGtsRefValidator(inherited_dialect="draft-07").validate_instance( + errors = _validator().validate_instance( ["tuple-prefix", "gts.x.other._.target.v1~"], schema ) assert len(errors) == 1 @@ -264,12 +269,9 @@ def test_prefix_items_recursion(self): schema = { "type": "array", "prefixItems": [{"type": "string"}], - "items": { - "type": "string", - "x-gts-ref": "gts.x.test._.target.v1~", - }, + "items": {"type": "string", X_GTS_TYPE_REF: "gts.x.test.*"}, } - errors = XGtsRefValidator(inherited_dialect="draft-07").validate_instance( + errors = _validator().validate_instance( ["tuple-prefix", "gts.x.other._.target.v1~"], schema ) assert len(errors) == 1 @@ -278,58 +280,46 @@ def test_prefix_items_recursion(self): def test_object_properties_recursion(self): schema = { "type": "object", - "properties": {"ref": {"x-gts-ref": "gts.x.test.*"}}, + "properties": {"ref": {X_GTS_TYPE_REF: "gts.x.test.*"}}, } - errors = XGtsRefValidator(inherited_dialect="draft-07").validate_instance( - {"ref": "gts.x.other.v1~"}, schema - ) + errors = _validator().validate_instance({"ref": "gts.x.other.v1~"}, schema) assert len(errors) == 1 def test_root_ref_traverses_nested_constraints(self): schema = { "type": "object", "properties": { - "link": {"x-gts-ref": "gts.x.test.*"}, + "link": {X_GTS_TYPE_REF: "gts.x.test.*"}, "child": {"$ref": "#"}, }, } - - errors = XGtsRefValidator(inherited_dialect="draft-07").validate_instance( + errors = _validator().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": [ - {"x-gts-ref": "gts.x.test._.a.v1~"}, - {"x-gts-ref": "gts.x.test._.b.v1~"}, + {X_GTS_TYPE_REF: "gts.x.test._.a.v1~"}, + {X_GTS_TYPE_REF: "gts.x.test._.b.v1~"}, ] } - errors = XGtsRefValidator(inherited_dialect="draft-07").validate_instance( - "gts.x.test._.c.v1~x.test._.item.v1", schema - ) + errors = _validator().validate_instance("gts.x.test._.c.v1~", schema) assert any("anyOf: no branch matched" in e.reason for e in errors) def test_any_of_matches_one_branch(self): schema = { "anyOf": [ - {"x-gts-ref": "gts.x.test._.a.v1~"}, - {"x-gts-ref": "gts.x.test._.b.v1~"}, + {X_GTS_TYPE_REF: "gts.x.test._.a.v1~"}, + {X_GTS_TYPE_REF: "gts.x.test._.b.v1~"}, ] } - errors = XGtsRefValidator(inherited_dialect="draft-07").validate_instance( - "gts.x.test._.a.v1~x.test._.item.v1", schema - ) + errors = _validator().validate_instance("gts.x.test._.a.v1~", schema) assert errors == [] def test_all_of_validates_each_matching_branch(self): - schema = { - "allOf": [ - {"type": "string", "x-gts-ref": "gts.x.test.*"}, - ] - } - errors = XGtsRefValidator(inherited_dialect="draft-07").validate_instance("gts.x.other.v1~", schema) + schema = {"allOf": [{"type": "string", X_GTS_TYPE_REF: "gts.x.test.*"}]} + errors = _validator().validate_instance("gts.x.other.v1~", schema) assert len(errors) == 1