From 98c6bc86b9c10229024cae1701caf670fab8b537 Mon Sep 17 00:00:00 2001 From: Artfizer Date: Tue, 8 Sep 2026 22:27:01 +0300 Subject: [PATCH 1/5] feat: add JSON validation CLI command (validate-json) Add a validate-json CLI command that scans JSON files, registers GTS schemas and instances, and reports validation issues for given json file or folder with *.json files Signed-off-by: Artfizer --- gts/src/gts/_cli.py | 17 +++ gts/src/gts/_json_validation.py | 244 ++++++++++++++++++++++++++++++++ tests/test_json_validation.py | 79 +++++++++++ 3 files changed, 340 insertions(+) create mode 100644 gts/src/gts/_json_validation.py create mode 100644 tests/test_json_validation.py diff --git a/gts/src/gts/_cli.py b/gts/src/gts/_cli.py index fc5cd1b..0134c4e 100644 --- a/gts/src/gts/_cli.py +++ b/gts/src/gts/_cli.py @@ -5,6 +5,7 @@ import logging import sys +from ._json_validation import GtsJsonValidator from ._server import GtsHttpServer from .ops import GtsOps @@ -34,6 +35,11 @@ def build_parser() -> argparse.ArgumentParser: s.add_argument("--gts-id", required=True) s.add_argument("--scope", choices=["major", "full"], default="major") + s = sub.add_parser( + "validate-json", help="Validate all JSON documents in a file or directory" + ) + s.add_argument("--path", dest="scan_path", help="JSON file or directory to scan") + s = sub.add_parser( "validate-instance", help="Validate an instance against its schema" ) @@ -146,6 +152,17 @@ def main(argv: list[str] | None = None) -> None: json.dump(out, sys.stdout, ensure_ascii=False, indent=2) sys.stdout.write("\n") return + elif args.op == "validate-json": + scan_path = args.scan_path or args.path + if not scan_path: + parser.error("validate-json requires --path") + result = GtsJsonValidator(scan_path, ops.cfg).validate() + for issue in result.issues: + suffix = f"#{issue.index}" if issue.index is not None else "" + sys.stderr.write( + f"{issue.file}{suffix}: {issue.stage}: {issue.message}\n" + ) + out = result.to_dict() elif args.op == "validate-id": out = ops.validate_id(args.gts_id).to_dict() elif args.op == "parse-id": diff --git a/gts/src/gts/_json_validation.py b/gts/src/gts/_json_validation.py new file mode 100644 index 0000000..fa533d9 --- /dev/null +++ b/gts/src/gts/_json_validation.py @@ -0,0 +1,244 @@ +from __future__ import annotations + +import json +import os +import uuid +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any + +from jsonschema.validators import validator_for + +from .entities import GtsEntity, GtsFile +from .gts import GtsID +from .store import GtsStore + + +@dataclass +class GtsJsonValidationIssue: + file: str + stage: str + message: str + index: int | None = None + + def to_dict(self) -> dict[str, Any]: + result: dict[str, Any] = { + "file": self.file, + "stage": self.stage, + "message": self.message, + } + if self.index is not None: + result["index"] = self.index + return result + + +@dataclass +class GtsJsonValidationResult: + files: int = 0 + documents: int = 0 + gts_entities: int = 0 + schemas: int = 0 + instances: int = 0 + issues: list[GtsJsonValidationIssue] = field(default_factory=list) + + @property + def ok(self) -> bool: + return not self.issues + + def to_dict(self) -> dict[str, Any]: + return { + "ok": self.ok, + "files": self.files, + "documents": self.documents, + "gts_entities": self.gts_entities, + "schemas": self.schemas, + "instances": self.instances, + "issues": [issue.to_dict() for issue in self.issues], + } + + +class GtsJsonValidator: + def __init__(self, path: str, cfg: Any) -> None: + self.path = Path(path).expanduser() + self.cfg = cfg + self.result = GtsJsonValidationResult() + self.entities: list[GtsEntity] = [] + + def validate(self) -> GtsJsonValidationResult: + for file_path in self._json_files(): + self._read_file(file_path) + self._validate_json_schemas() + store = self._register_gts_entities() + self._validate_schemas(store) + self._validate_instances(store) + return self.result + + def _json_files(self) -> list[Path]: + resolved = self.path.resolve(strict=False) + if resolved.is_file(): + if resolved.suffix.lower() == ".json": + return [resolved] + self._issue(resolved, "discovery", "Expected a .json file") + return [] + if not resolved.is_dir(): + self._issue( + resolved, "discovery", "Path does not exist or is not accessible" + ) + return [] + + files: list[Path] = [] + for root, dirs, names in os.walk(resolved, followlinks=True): + dirs[:] = [ + name for name in dirs if name not in {"node_modules", "dist", "build"} + ] + files.extend( + Path(root, name).resolve(strict=False) + for name in names + if Path(name).suffix.lower() == ".json" + ) + return sorted(set(files)) + + def _read_file(self, file_path: Path) -> None: + self.result.files += 1 + try: + with file_path.open(encoding="utf-8") as source: + content = json.load(source) + except Exception as error: # noqa: BLE001 - report this document and continue + self._issue(file_path, "json", str(error)) + return + + values = content if isinstance(content, list) else [content] + file = GtsFile(path=str(file_path), name=file_path.name, content=content) + for index, value in enumerate(values): + self.result.documents += 1 + self.entities.append( + GtsEntity( + file=file, + list_sequence=index if isinstance(content, list) else None, + content=value, + cfg=self.cfg, + ) + ) + + def _validate_json_schemas(self) -> None: + for entity in self.entities: + content = entity.content + if not isinstance(content, dict) or "$schema" not in content: + continue + try: + validator_for(content).check_schema(content) + except Exception as error: # noqa: BLE001 - report this document and continue + self._issue(entity, "json-schema", str(error)) + + def _register_gts_entities(self) -> GtsStore: + store = GtsStore(reader=None) # type: ignore[arg-type] + keys: set[str] = set() + for entity in self.entities: + if not self._is_gts_related(entity.content): + continue + key = self._registry_key(entity) + if key is None: + self._issue( + entity, "registry", "GTS-related document has no registrable GTS ID" + ) + continue + if not entity.is_schema and entity.selected_entity_field is None: + raw_id = entity.raw_id + if raw_id is not None: + entity.raw_id = str(uuid.uuid5(uuid.NAMESPACE_URL, raw_id)) + key = entity.raw_id + if key in keys: + self._issue(entity, "registry", f"Duplicate GTS entity ID '{key}'") + continue + keys.add(key) + store.register(entity) + self.result.gts_entities += 1 + if entity.is_schema: + self.result.schemas += 1 + else: + self.result.instances += 1 + return store + + @staticmethod + def _is_gts_related(value: Any) -> bool: + if isinstance(value, str): + return "gts." in value + if isinstance(value, dict): + return any( + GtsJsonValidator._is_gts_related(item) for item in value.values() + ) + if isinstance(value, list): + return any(GtsJsonValidator._is_gts_related(item) for item in value) + return False + + @staticmethod + def _registry_key(entity: GtsEntity) -> str | None: + if entity.is_schema and entity.gts_id: + return entity.gts_id.id + if ( + not entity.is_schema + and entity.raw_id + and ( + entity.gts_id + or (entity.type_id is not None and GtsID.is_valid(entity.type_id)) + ) + ): + return entity.raw_id + return None + + def _validate_schemas(self, store: GtsStore) -> None: + schemas = sorted( + ( + entity + for entity in self.entities + if entity.is_schema + and entity.gts_id + and store.get(entity.gts_id.id) is entity + ), + key=self._schema_depth, + ) + for stage, depth in (("base-type", 1), ("derived-type", None)): + for entity in schemas: + if (depth == 1) != (self._schema_depth(entity) == 1): + continue + gts_id = entity.gts_id + if not gts_id: + continue + try: + store.validate_schema(gts_id.id) + except Exception as error: # noqa: BLE001 - report this document and continue + self._issue(entity, stage, str(error)) + + @staticmethod + def _schema_depth(entity: GtsEntity) -> int: + return len(entity.gts_id.gts_id_segments) if entity.gts_id else 0 + + def _validate_instances(self, store: GtsStore) -> None: + for entity in self.entities: + key = self._registry_key(entity) + if ( + entity.is_schema + or key is None + or not self._is_gts_related(entity.content) + ): + continue + try: + store.validate_instance(key) + except Exception as error: # noqa: BLE001 - report this document and continue + self._issue(entity, "instance", str(error)) + + def _issue( + self, + source: Path | GtsEntity, + stage: str, + message: str, + ) -> None: + if isinstance(source, GtsEntity): + file = source.file.path if source.file else source.label + index = source.list_sequence + else: + file = str(source) + index = None + self.result.issues.append( + GtsJsonValidationIssue(file=file, stage=stage, message=message, index=index) + ) diff --git a/tests/test_json_validation.py b/tests/test_json_validation.py new file mode 100644 index 0000000..7f06f81 --- /dev/null +++ b/tests/test_json_validation.py @@ -0,0 +1,79 @@ +import json + +from gts._cli import main +from gts.entities import DEFAULT_GTS_CONFIG +from gts._json_validation import GtsJsonValidator + + +def test_validate_json_reports_all_document_errors(tmp_path): + (tmp_path / "broken.json").write_text("{", encoding="utf-8") + (tmp_path / "invalid-schema.json").write_text( + json.dumps( + { + "$schema": "https://json-schema.org/draft/2020-12/schema", + "$id": "gts://gts.example.catalog._.item.v1~", + "type": 3, + } + ), + encoding="utf-8", + ) + (tmp_path / "instance.json").write_text( + json.dumps( + { + "id": "gts.example.catalog._.item.v1~example.catalog._.one.v1", + "type": "gts.example.catalog._.item.v1~", + } + ), + encoding="utf-8", + ) + + result = GtsJsonValidator(str(tmp_path), DEFAULT_GTS_CONFIG).validate() + + assert result.files == 3 + assert result.documents == 2 + assert result.schemas == 1 + assert result.instances == 1 + assert not result.ok + assert {issue.stage for issue in result.issues} >= { + "json", + "json-schema", + "instance", + } + + +def test_validate_json_registers_type_only_instances(tmp_path): + type_id = "gts.example.catalog._.item.v1~" + (tmp_path / "schema.json").write_text( + json.dumps( + { + "$schema": "https://json-schema.org/draft/2020-12/schema", + "$id": f"gts://{type_id}", + "type": "object", + "required": ["type", "name"], + "properties": {"type": {"const": type_id}, "name": {"type": "string"}}, + } + ), + encoding="utf-8", + ) + (tmp_path / "instance.json").write_text( + json.dumps({"type": type_id, "name": "anonymous"}), encoding="utf-8" + ) + + result = GtsJsonValidator(str(tmp_path), DEFAULT_GTS_CONFIG).validate() + + assert result.ok + assert result.schemas == 1 + assert result.instances == 1 + + +def test_validate_json_cli_emits_issues_to_stderr(tmp_path, capsys): + input_path = tmp_path / "broken.json" + input_path.write_text("{", encoding="utf-8") + + main(["validate-json", "--path", str(input_path)]) + + captured = capsys.readouterr() + output = json.loads(captured.out) + assert output["ok"] is False + assert output["issues"][0]["file"] == str(input_path) + assert f"{input_path}: json:" in captured.err From 1712a17418cfccaa4a22f96c6b8bbe4a99345986 Mon Sep 17 00:00:00 2001 From: Artfizer Date: Tue, 8 Sep 2026 22:28:37 +0300 Subject: [PATCH 2/5] chore: bump version to 0.13.1 Signed-off-by: Artfizer --- gts/openapi.json | 2 +- gts/pyproject.toml | 2 +- gts/src/gts/_server.py | 2 +- 3 files changed, 3 insertions(+), 3 deletions(-) diff --git a/gts/openapi.json b/gts/openapi.json index df00e6f..69dc264 100644 --- a/gts/openapi.json +++ b/gts/openapi.json @@ -2,7 +2,7 @@ "openapi": "3.1.0", "info": { "title": "GTS Server", - "version": "0.13.0" + "version": "0.13.1" }, "paths": { "/entities": { diff --git a/gts/pyproject.toml b/gts/pyproject.toml index 893c599..970fb5d 100644 --- a/gts/pyproject.toml +++ b/gts/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "gts" -version = "0.13.0" +version = "0.13.1" description = "Global Type System (GTS) helpers: identifiers, parsing, validation, and operations" readme = "README.md" authors = [{ name = "GTS Community" }] diff --git a/gts/src/gts/_server.py b/gts/src/gts/_server.py index 653e65b..639ae16 100644 --- a/gts/src/gts/_server.py +++ b/gts/src/gts/_server.py @@ -185,7 +185,7 @@ def __init__( self.host = host self.port = port self.base_url = f"http://{self.host}:{self.port}" - self.app = FastAPI(title="GTS Server", version="0.13.0") + self.app = FastAPI(title="GTS Server", version="0.13.1") self.app.add_middleware( _RequestLoggingMiddleware, verbose=self.ops.verbose, From fa100659fe7cff780ddc6d4911c87b2cb6dd0d13 Mon Sep 17 00:00:00 2001 From: Artfizer Date: Tue, 8 Sep 2026 22:44:11 +0300 Subject: [PATCH 3/5] docs: add supported GTS spec version to README.md Signed-off-by: Artfizer --- README.md | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/README.md b/README.md index 594c9f9..45272f0 100644 --- a/README.md +++ b/README.md @@ -4,9 +4,9 @@ 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. -## Roadmap +Current supported GTS spec version: `0.13.1` -Current supported GTS spec version: 0.13 +## Roadmap Featureset: From bbf99145f627dd7a5a5fbebc239a10ccaa9dca0b Mon Sep 17 00:00:00 2001 From: Artfizer Date: Thu, 10 Sep 2026 01:42:40 +0300 Subject: [PATCH 4/5] refactor(validate-all): marker heuristic, sorted errors, JSON-only output MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Replace hardcoded directory excludes with text-based GTS marker heuristic: skip files whose raw text lacks "gts.", "gts://", or "x-gts-ref" before paying JSON parse cost. - Rename _validate_json_schemas → _check_schema_field_type (type-only). - Report malformed/non-GTS schema $id distinctly ("registry" stage). - Sort schema errors by (depth, gts_id, file, index): base-type first, then derived-type, each in total order on the remaining keys. - Sort instance errors by (depth, gts_id, file, index). - Update _is_gts_related to check gts://, x-gts-ref in addition to gts. - Remove stderr issue printing from validate-all CLI; output JSON only. - Add tests: malformed ID, incidental mention, duplicate entity, non-GTS file filtering, marker heuristic, schema/instance ordering, JSON-only CLI output. Signed-off-by: Artfizer --- gts/src/gts/_cli.py | 11 +-- gts/src/gts/_json_validation.py | 159 ++++++++++++++++++----------- tests/test_json_validation.py | 170 ++++++++++++++++++++++++++++++-- 3 files changed, 269 insertions(+), 71 deletions(-) diff --git a/gts/src/gts/_cli.py b/gts/src/gts/_cli.py index 0134c4e..398a3c4 100644 --- a/gts/src/gts/_cli.py +++ b/gts/src/gts/_cli.py @@ -36,7 +36,7 @@ def build_parser() -> argparse.ArgumentParser: s.add_argument("--scope", choices=["major", "full"], default="major") s = sub.add_parser( - "validate-json", help="Validate all JSON documents in a file or directory" + "validate-all", help="Validate all JSON documents in a file or directory" ) s.add_argument("--path", dest="scan_path", help="JSON file or directory to scan") @@ -152,16 +152,11 @@ def main(argv: list[str] | None = None) -> None: json.dump(out, sys.stdout, ensure_ascii=False, indent=2) sys.stdout.write("\n") return - elif args.op == "validate-json": + elif args.op == "validate-all": scan_path = args.scan_path or args.path if not scan_path: - parser.error("validate-json requires --path") + parser.error("validate-all requires --path") result = GtsJsonValidator(scan_path, ops.cfg).validate() - for issue in result.issues: - suffix = f"#{issue.index}" if issue.index is not None else "" - sys.stderr.write( - f"{issue.file}{suffix}: {issue.stage}: {issue.message}\n" - ) out = result.to_dict() elif args.op == "validate-id": out = ops.validate_id(args.gts_id).to_dict() diff --git a/gts/src/gts/_json_validation.py b/gts/src/gts/_json_validation.py index fa533d9..11e31f2 100644 --- a/gts/src/gts/_json_validation.py +++ b/gts/src/gts/_json_validation.py @@ -7,12 +7,12 @@ from pathlib import Path from typing import Any -from jsonschema.validators import validator_for - from .entities import GtsEntity, GtsFile -from .gts import GtsID +from .gts import GTS_PREFIX, GTS_URI_PREFIX, GtsID from .store import GtsStore +_X_GTS_REF_KEYWORD = "x-gts-ref" + @dataclass class GtsJsonValidationIssue: @@ -67,8 +67,12 @@ def __init__(self, path: str, cfg: Any) -> None: def validate(self) -> GtsJsonValidationResult: for file_path in self._json_files(): self._read_file(file_path) - self._validate_json_schemas() + self._check_schema_field_type() store = self._register_gts_entities() + schemas_count, instances_count = self._count_schema_instance() + self.result.schemas = schemas_count + self.result.instances = instances_count + self.result.gts_entities = schemas_count + instances_count self._validate_schemas(store) self._validate_instances(store) return self.result @@ -87,22 +91,35 @@ def _json_files(self) -> list[Path]: return [] files: list[Path] = [] - for root, dirs, names in os.walk(resolved, followlinks=True): - dirs[:] = [ - name for name in dirs if name not in {"node_modules", "dist", "build"} - ] - files.extend( - Path(root, name).resolve(strict=False) - for name in names - if Path(name).suffix.lower() == ".json" - ) - return sorted(set(files)) + seen: set[Path] = set() + for root, _dirs, names in os.walk(resolved, followlinks=True): + for name in names: + if Path(name).suffix.lower() == ".json": + rp = Path(root, name).resolve(strict=False) + if rp not in seen: + seen.add(rp) + files.append(rp) + return sorted(files) + + @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 + ) def _read_file(self, file_path: Path) -> None: + try: + content_str = file_path.read_text(encoding="utf-8") + except Exception as error: # noqa: BLE001 - report this document and continue + self._issue(file_path, "json", str(error)) + return + + if not self._is_gts_marker(content_str): + return self.result.files += 1 + try: - with file_path.open(encoding="utf-8") as source: - content = json.load(source) + content = json.loads(content_str) except Exception as error: # noqa: BLE001 - report this document and continue self._issue(file_path, "json", str(error)) return @@ -120,15 +137,14 @@ def _read_file(self, file_path: Path) -> None: ) ) - def _validate_json_schemas(self) -> None: + def _check_schema_field_type(self) -> None: for entity in self.entities: content = entity.content - if not isinstance(content, dict) or "$schema" not in content: + if not isinstance(content, dict): continue - try: - validator_for(content).check_schema(content) - except Exception as error: # noqa: BLE001 - report this document and continue - self._issue(entity, "json-schema", str(error)) + schema_val = content.get("$schema") + if schema_val is not None and not isinstance(schema_val, str): + self._issue(entity, "json-schema", "$schema must be a string") def _register_gts_entities(self) -> GtsStore: store = GtsStore(reader=None) # type: ignore[arg-type] @@ -138,9 +154,12 @@ def _register_gts_entities(self) -> GtsStore: continue key = self._registry_key(entity) if key is None: - self._issue( - entity, "registry", "GTS-related document has no registrable GTS ID" - ) + if entity.is_schema: + self._issue( + entity, + "registry", + "GTS schema has a malformed or non-GTS $id", + ) continue if not entity.is_schema and entity.selected_entity_field is None: raw_id = entity.raw_id @@ -152,17 +171,28 @@ def _register_gts_entities(self) -> GtsStore: continue keys.add(key) store.register(entity) - self.result.gts_entities += 1 + return store + + def _count_schema_instance(self) -> tuple[int, int]: + schemas = 0 + instances = 0 + for entity in self.entities: + if self._registry_key(entity) is None: + continue if entity.is_schema: - self.result.schemas += 1 + schemas += 1 else: - self.result.instances += 1 - return store + instances += 1 + return schemas, instances @staticmethod def _is_gts_related(value: Any) -> bool: if isinstance(value, str): - return "gts." in value + return ( + GTS_PREFIX in value + or GTS_URI_PREFIX in value + or _X_GTS_REF_KEYWORD in value + ) if isinstance(value, dict): return any( GtsJsonValidator._is_gts_related(item) for item in value.values() @@ -187,43 +217,58 @@ def _registry_key(entity: GtsEntity) -> str | None: return None def _validate_schemas(self, store: GtsStore) -> None: - schemas = sorted( - ( - entity - for entity in self.entities - if entity.is_schema - and entity.gts_id - and store.get(entity.gts_id.id) is entity - ), - key=self._schema_depth, - ) - for stage, depth in (("base-type", 1), ("derived-type", None)): - for entity in schemas: - if (depth == 1) != (self._schema_depth(entity) == 1): - continue - gts_id = entity.gts_id - if not gts_id: - continue - try: - store.validate_schema(gts_id.id) - except Exception as error: # noqa: BLE001 - report this document and continue - self._issue(entity, stage, str(error)) + pending: list[tuple[int, str, str, int | None, GtsEntity]] = [] + for entity in self.entities: + if not entity.is_schema or not entity.gts_id: + continue + gid = entity.gts_id + if store.get(gid.id) is not entity: + continue + depth = len(gid.gts_id_segments) + file = entity.file.path if entity.file else entity.label + pending.append((depth, gid.id, file, entity.list_sequence, entity)) + pending.sort(key=lambda t: (t[0], t[1], t[2], t[3] if t[3] is not None else -1)) + + for depth, _gts_id, _file, _idx, entity in pending: + stage = "base-type" if depth <= 1 else "derived-type" + try: + store.validate_schema(entity.gts_id.id) # type: ignore[union-attr] + except Exception as error: # noqa: BLE001 - report this document and continue + self._issue(entity, stage, str(error)) @staticmethod def _schema_depth(entity: GtsEntity) -> int: return len(entity.gts_id.gts_id_segments) if entity.gts_id else 0 + @staticmethod + def _entity_depth(entity: GtsEntity) -> int: + if entity.gts_id: + return len(entity.gts_id.gts_id_segments) + if entity.type_id and GtsID.is_valid(entity.type_id): + return len(GtsID(entity.type_id).gts_id_segments) + return 0 + def _validate_instances(self, store: GtsStore) -> None: + pending: list[tuple[int, str, str, int | None, str]] = [] for entity in self.entities: + if entity.is_schema: + continue key = self._registry_key(entity) - if ( - entity.is_schema - or key is None - or not self._is_gts_related(entity.content) - ): + if key is None: + continue + depth = self._entity_depth(entity) + gts_id_str = entity.gts_id.id if entity.gts_id else "" + file = entity.file.path if entity.file else entity.label + pending.append((depth, gts_id_str, file, entity.list_sequence, key)) + pending.sort(key=lambda t: (t[0], t[1], t[2], t[3] if t[3] is not None else -1)) + + for _depth, _gts_id, _file, _idx, registry_key in pending: + # Find the entity for error reporting + entity = store.get(registry_key) + if entity is None: continue try: - store.validate_instance(key) + store.validate_instance(registry_key) except Exception as error: # noqa: BLE001 - report this document and continue self._issue(entity, "instance", str(error)) diff --git a/tests/test_json_validation.py b/tests/test_json_validation.py index 7f06f81..c8e9d2b 100644 --- a/tests/test_json_validation.py +++ b/tests/test_json_validation.py @@ -6,7 +6,7 @@ def test_validate_json_reports_all_document_errors(tmp_path): - (tmp_path / "broken.json").write_text("{", encoding="utf-8") + (tmp_path / "broken.json").write_text('{ "id": "gts.broken', encoding="utf-8") (tmp_path / "invalid-schema.json").write_text( json.dumps( { @@ -36,7 +36,6 @@ def test_validate_json_reports_all_document_errors(tmp_path): assert not result.ok assert {issue.stage for issue in result.issues} >= { "json", - "json-schema", "instance", } @@ -66,14 +65,173 @@ def test_validate_json_registers_type_only_instances(tmp_path): assert result.instances == 1 -def test_validate_json_cli_emits_issues_to_stderr(tmp_path, capsys): +def test_validate_json_cli_outputs_json_only(tmp_path, capsys): input_path = tmp_path / "broken.json" - input_path.write_text("{", encoding="utf-8") + input_path.write_text('{ "id": "gts.broken', encoding="utf-8") - main(["validate-json", "--path", str(input_path)]) + main(["validate-all", "--path", str(input_path)]) captured = capsys.readouterr() output = json.loads(captured.out) assert output["ok"] is False assert output["issues"][0]["file"] == str(input_path) - assert f"{input_path}: json:" in captured.err + assert captured.err == "" + + +def test_malformed_schema_id_is_reported(tmp_path): + malformed = json.dumps({ + "$id": "gts://gtx.cli.core.test.bad.v1~", + "$schema": "http://json-schema.org/draft-07/schema#", + "type": "object", + }) + (tmp_path / "bad.schema.json").write_text(malformed, encoding="utf-8") + + result = GtsJsonValidator(str(tmp_path), DEFAULT_GTS_CONFIG).validate() + + assert not result.ok, "malformed GTS id should fail" + assert result.gts_entities == 0 + assert any( + i.stage == "registry" and "malformed" in i.message for i in result.issues + ), f"expected a malformed-id diagnostic, got: {result.issues}" + + +def test_incidental_prefix_mention_is_not_registered(tmp_path): + doc = json.dumps({"description": "see gts.foo.bar for details", "value": 42}) + (tmp_path / "unrelated.json").write_text(doc, encoding="utf-8") + + result = GtsJsonValidator(str(tmp_path), DEFAULT_GTS_CONFIG).validate() + + assert result.ok, f"incidental mention should not fail: {result.issues}" + assert result.documents == 1 + assert result.gts_entities == 0 + assert not result.issues + + +def test_duplicate_entity_is_reported(tmp_path): + schema = json.dumps({ + "$id": "gts://gts.cli.core.test.base.v1~", + "$schema": "http://json-schema.org/draft-07/schema#", + "type": "object", + "properties": {"name": {"type": "string"}}, + "required": ["name"], + }) + (tmp_path / "a.schema.json").write_text(schema, encoding="utf-8") + (tmp_path / "b.schema.json").write_text(schema, encoding="utf-8") + + result = GtsJsonValidator(str(tmp_path), DEFAULT_GTS_CONFIG).validate() + + assert not result.ok, "duplicate ids should fail" + assert any( + "Duplicate" in i.message for i in result.issues + ), f"expected a duplicate diagnostic, got: {result.issues}" + + +def test_non_gts_files_are_ignored(tmp_path): + schema = json.dumps({ + "$id": "gts://gts.cli.core.test.base.v1~", + "$schema": "http://json-schema.org/draft-07/schema#", + "type": "object", + "properties": {"name": {"type": "string"}}, + "required": ["name"], + }) + (tmp_path / "base.schema.json").write_text(schema, encoding="utf-8") + (tmp_path / "package.json").write_text( + '{"name": "pkg", "version": "1.0.0"}', encoding="utf-8" + ) + nm = tmp_path / "node_modules" + nm.mkdir() + (nm / "broken.json").write_text("{ this is not json ", encoding="utf-8") + + result = GtsJsonValidator(str(tmp_path), DEFAULT_GTS_CONFIG).validate() + + assert result.ok, f"non-GTS files must be ignored: {result.issues}" + assert result.files == 1, "only the GTS schema should be processed" + assert result.schemas == 1 + + +def test_marker_heuristic_matches_expected_combinations(): + assert GtsJsonValidator._is_gts_marker('{"id": "gts.x.y.z.t.v1~a.b.c.d.v1.0"}') + assert GtsJsonValidator._is_gts_marker('{"$id": "gts://gts.x.y.z.t.v1~"}') + assert GtsJsonValidator._is_gts_marker( + '{"properties": {"p": {"x-gts-ref": "..."}}}' + ) + assert not GtsJsonValidator._is_gts_marker( + '{"name": "widgets", "version": "1.0.0"}' + ) + + +def test_schema_errors_ordered_by_depth_then_gts_id(tmp_path): + base_schema = json.dumps({ + "$id": "gts://gts.cli.core.test.base.v1~", + "$schema": "http://json-schema.org/draft-07/schema#", + "type": "object", + "properties": {"name": {"type": "string"}}, + "required": ["name"], + }) + (tmp_path / "base.schema.json").write_text(base_schema, encoding="utf-8") + + def invalid_base(seg): + return json.dumps({ + "$id": f"gts://gts.cli.core.test.{seg}.v1~", + "$schema": "http://json-schema.org/draft-07/schema#", + "type": "invalid_type", + }) + + invalid_leaf = json.dumps({ + "$id": "gts://gts.cli.core.test.base.v1~cli.core.test.leaf.v1~", + "$schema": "http://json-schema.org/draft-07/schema#", + "type": "invalid_type", + }) + + (tmp_path / "0_mmm.schema.json").write_text(invalid_base("mmm"), encoding="utf-8") + (tmp_path / "a_leaf.schema.json").write_text(invalid_leaf, encoding="utf-8") + (tmp_path / "z_aaa.schema.json").write_text(invalid_base("aaa"), encoding="utf-8") + + result = GtsJsonValidator(str(tmp_path), DEFAULT_GTS_CONFIG).validate() + + schema_issues = [ + i for i in result.issues if i.stage in ("base-type", "derived-type") + ] + + assert len(schema_issues) == 3, f"issues: {result.issues}" + assert schema_issues[0].stage == "base-type" + assert schema_issues[0].file.endswith("z_aaa.schema.json") + assert schema_issues[1].stage == "base-type" + assert schema_issues[1].file.endswith("0_mmm.schema.json") + assert schema_issues[2].stage == "derived-type" + assert schema_issues[2].file.endswith("a_leaf.schema.json") + + +def test_instance_errors_ordered_by_gts_id(tmp_path): + base_schema = json.dumps({ + "$id": "gts://gts.cli.core.test.base.v1~", + "$schema": "http://json-schema.org/draft-07/schema#", + "type": "object", + "properties": {"name": {"type": "string"}}, + "required": ["name"], + }) + (tmp_path / "base.schema.json").write_text(base_schema, encoding="utf-8") + + def invalid_instance(seg): + return json.dumps({ + "id": f"gts.cli.core.test.base.v1~cli.app._.{seg}.v1.0" + }) + + (tmp_path / "z_alpha.json").write_text( + invalid_instance("alpha"), encoding="utf-8" + ) + (tmp_path / "a_zeta.json").write_text( + invalid_instance("zeta"), encoding="utf-8" + ) + + result = GtsJsonValidator(str(tmp_path), DEFAULT_GTS_CONFIG).validate() + + instance_issues = [i for i in result.issues if i.stage == "instance"] + + assert len(instance_issues) == 2, f"issues: {result.issues}" + assert instance_issues[0].file.endswith("z_alpha.json"), ( + f"alpha should be first: {instance_issues}" + ) + assert instance_issues[1].file.endswith("a_zeta.json"), ( + f"zeta should be second: {instance_issues}" + ) From c2a108de6b21912bf396ac871cab7083f8f1412a Mon Sep 17 00:00:00 2001 From: Artfizer Date: Thu, 10 Sep 2026 02:56:27 +0300 Subject: [PATCH 5/5] feat(cli): add global --exclude option for directory scanning Add a global `--exclude` option (alongside `--path`) that accepts a comma-separated list of directory names to skip during recursive file scanning. Defaults to `node_modules,dist,build,.git,target`. The parsed list is threaded through GtsOps (and reload_from_path) into GtsFileReader, and into GtsJsonValidator for validate-all. The module constant is renamed EXCLUDE_LIST -> DEFAULT_EXCLUDE_LIST and used as the per-instance fallback via a new `exclude` parameter on the reader and validator. Signed-off-by: Artfizer --- gts/src/gts/_cli.py | 25 +++++++++- gts/src/gts/_json_validation.py | 83 ++++++++++++++++++++++----------- gts/src/gts/files_reader.py | 20 ++++++-- gts/src/gts/ops.py | 10 +++- tests/test_json_validation.py | 6 ++- 5 files changed, 108 insertions(+), 36 deletions(-) diff --git a/gts/src/gts/_cli.py b/gts/src/gts/_cli.py index 398a3c4..db7dd34 100644 --- a/gts/src/gts/_cli.py +++ b/gts/src/gts/_cli.py @@ -19,6 +19,14 @@ def build_parser() -> argparse.ArgumentParser: p.add_argument( "--path", help="Path to json and schema files or directories (global default)" ) + p.add_argument( + "--exclude", + default="node_modules,dist,build,.git,target", + help=( + "Comma-separated directory names to exclude when scanning " + "(default: node_modules,dist,build,.git,target)" + ), + ) sub = p.add_subparsers(dest="op", required=True) s = sub.add_parser("validate-id", help="Validate a GTS ID format") @@ -122,8 +130,16 @@ def main(argv: list[str] | None = None) -> None: ) try: + # Parse the comma-separated --exclude option into a list of dir names + exclude = [e.strip() for e in (args.exclude or "").split(",") if e.strip()] + # Helper to create GtsOps with common arguments - ops = GtsOps(path=args.path, config=args.config, verbose=args.verbose) + ops = GtsOps( + path=args.path, + config=args.config, + verbose=args.verbose, + exclude=exclude, + ) if args.op == "server": server = GtsHttpServer(ops=ops) @@ -156,8 +172,13 @@ def main(argv: list[str] | None = None) -> None: scan_path = args.scan_path or args.path if not scan_path: parser.error("validate-all requires --path") - result = GtsJsonValidator(scan_path, ops.cfg).validate() + result = GtsJsonValidator(scan_path, ops.cfg, exclude=exclude).validate() out = result.to_dict() + json.dump(out, sys.stdout, ensure_ascii=False, indent=2) + sys.stdout.write("\n") + if not result.ok: + raise SystemExit(1) + return elif args.op == "validate-id": out = ops.validate_id(args.gts_id).to_dict() elif args.op == "parse-id": diff --git a/gts/src/gts/_json_validation.py b/gts/src/gts/_json_validation.py index 11e31f2..3f070ad 100644 --- a/gts/src/gts/_json_validation.py +++ b/gts/src/gts/_json_validation.py @@ -8,6 +8,7 @@ from typing import Any from .entities import GtsEntity, GtsFile +from .files_reader import DEFAULT_EXCLUDE_LIST from .gts import GTS_PREFIX, GTS_URI_PREFIX, GtsID from .store import GtsStore @@ -58,9 +59,10 @@ def to_dict(self) -> dict[str, Any]: class GtsJsonValidator: - def __init__(self, path: str, cfg: Any) -> None: + def __init__(self, path: str, cfg: Any, exclude: list[str] | None = None) -> None: self.path = Path(path).expanduser() self.cfg = cfg + self.exclude = list(exclude) if exclude else list(DEFAULT_EXCLUDE_LIST) self.result = GtsJsonValidationResult() self.entities: list[GtsEntity] = [] @@ -91,14 +93,38 @@ def _json_files(self) -> list[Path]: return [] files: list[Path] = [] - seen: set[Path] = set() - for root, _dirs, names in os.walk(resolved, followlinks=True): + seen_files: set[Path] = set() + seen_dirs: set[tuple[int, int]] = set() + walk_errors: list[OSError] = [] + + def _on_walk_error(err: OSError) -> None: + walk_errors.append(err) + + for root, dirs, names in os.walk( + resolved, followlinks=True, onerror=_on_walk_error + ): + # Prevent symlink cycles by tracking visited directory identities + root_stat = os.stat(root) + dir_id = (root_stat.st_dev, root_stat.st_ino) + if dir_id in seen_dirs: + dirs.clear() + continue + seen_dirs.add(dir_id) + + # Prune excluded directories (defaults to DEFAULT_EXCLUDE_LIST, + # overridable via the CLI --exclude option) + dirs[:] = [d for d in dirs if d not in self.exclude] + for name in names: if Path(name).suffix.lower() == ".json": rp = Path(root, name).resolve(strict=False) - if rp not in seen: - seen.add(rp) + if rp not in seen_files: + seen_files.add(rp) files.append(rp) + + for err in walk_errors: + self._issue(resolved, "discovery", f"Traversal error: {err}") + return sorted(files) @staticmethod @@ -185,22 +211,28 @@ def _count_schema_instance(self) -> tuple[int, int]: instances += 1 return schemas, instances - @staticmethod - def _is_gts_related(value: Any) -> bool: - if isinstance(value, str): - return ( - GTS_PREFIX in value - or GTS_URI_PREFIX in value - or _X_GTS_REF_KEYWORD in value - ) - if isinstance(value, dict): - return any( - GtsJsonValidator._is_gts_related(item) for item in value.values() - ) - if isinstance(value, list): - return any(GtsJsonValidator._is_gts_related(item) for item in value) + def _is_gts_related(self, value: Any) -> bool: + if not isinstance(value, dict): + return False + # Check configured identifier fields for GTS IDs. + # Accept valid IDs and also detect likely-but-malformed ones + # (gts:// or gts. prefix) so they get diagnosed during registration + # rather than silently skipped. + for f in self.cfg.entity_id_fields: + v = value.get(f) + if isinstance(v, str) and self._looks_gts(v): + return True + for f in self.cfg.schema_id_fields: + v = value.get(f) + if isinstance(v, str) and self._looks_gts(v): + return True return False + @staticmethod + def _looks_gts(v: str) -> bool: + normalized = v.removeprefix(GTS_URI_PREFIX) + return normalized.startswith(GTS_PREFIX) or v.startswith(GTS_URI_PREFIX) + @staticmethod def _registry_key(entity: GtsEntity) -> str | None: if entity.is_schema and entity.gts_id: @@ -249,24 +281,23 @@ def _entity_depth(entity: GtsEntity) -> int: return 0 def _validate_instances(self, store: GtsStore) -> None: - pending: list[tuple[int, str, str, int | None, str]] = [] + pending: list[tuple[int, str, str, int | None, str, GtsEntity]] = [] for entity in self.entities: if entity.is_schema: continue key = self._registry_key(entity) if key is None: continue + # Skip rejected duplicates: only validate the registered entity + if store.get(key) is not entity: + continue depth = self._entity_depth(entity) gts_id_str = entity.gts_id.id if entity.gts_id else "" file = entity.file.path if entity.file else entity.label - pending.append((depth, gts_id_str, file, entity.list_sequence, key)) + pending.append((depth, gts_id_str, file, entity.list_sequence, key, entity)) pending.sort(key=lambda t: (t[0], t[1], t[2], t[3] if t[3] is not None else -1)) - for _depth, _gts_id, _file, _idx, registry_key in pending: - # Find the entity for error reporting - entity = store.get(registry_key) - if entity is None: - continue + for _depth, _gts_id, _file, _idx, registry_key, entity in pending: try: store.validate_instance(registry_key) except Exception as error: # noqa: BLE001 - report this document and continue diff --git a/gts/src/gts/files_reader.py b/gts/src/gts/files_reader.py index 66cd24d..f309f95 100644 --- a/gts/src/gts/files_reader.py +++ b/gts/src/gts/files_reader.py @@ -12,7 +12,9 @@ from .entities import DEFAULT_GTS_CONFIG, GtsConfig, GtsEntity, GtsFile from .store import GtsReader -EXCLUDE_LIST = ["node_modules", "dist", "build"] +# Default directory names skipped during recursive scanning. The CLI --exclude +# option overrides this per invocation. +DEFAULT_EXCLUDE_LIST = ["node_modules", "dist", "build", ".git", "target"] logger = logging.getLogger(__name__) @@ -20,13 +22,20 @@ class GtsFileReader(GtsReader): """Reads GTS entities from JSON and YAML files in directories specified by path.""" - def __init__(self, path: str | list[str], cfg: GtsConfig | None = None) -> None: + def __init__( + self, + path: str | list[str], + cfg: GtsConfig | None = None, + exclude: list[str] | None = None, + ) -> None: """ Initialize FileReader with one or more paths. Args: path: Single path string or list of paths (files or directories) cfg: GtsConfig for entity ID extraction (defaults to DEFAULT_GTS_CONFIG) + exclude: Directory names to skip while scanning (defaults to + DEFAULT_EXCLUDE_LIST) """ self.paths: list[Path] = [] if isinstance(path, str): @@ -35,6 +44,7 @@ def __init__(self, path: str | list[str], cfg: GtsConfig | None = None) -> None: self.paths = [Path(os.path.expanduser(p)) for p in path] self.cfg = cfg or DEFAULT_GTS_CONFIG + self.exclude = list(exclude) if exclude else list(DEFAULT_EXCLUDE_LIST) self._files: list[Path] = [] self._current_index = 0 self._current_file_entities: list[GtsEntity] = [] @@ -61,9 +71,9 @@ def _collect_files(self) -> None: elif resolved_path.is_dir(): # Recursively scan for all valid file types, following symlinks for root, dirs, files in os.walk(resolved_path, followlinks=True): - for exclude in EXCLUDE_LIST: - if exclude in dirs: - dirs.remove(exclude) + for excluded in self.exclude: + if excluded in dirs: + dirs.remove(excluded) for fname in files: ext = os.path.splitext(fname)[1].lower() if ext in valid_extensions: diff --git a/gts/src/gts/ops.py b/gts/src/gts/ops.py index 4954c83..99e7856 100644 --- a/gts/src/gts/ops.py +++ b/gts/src/gts/ops.py @@ -297,11 +297,17 @@ def __init__( path: str | builtins.list[str] | None = None, config: str | None = None, verbose: int = 0, + exclude: builtins.list[str] | None = None, ) -> None: self.verbose = verbose self.cfg = self._load_config(config) self.path: str | list[str] | None = path - self._reader = GtsFileReader(self.path, cfg=self.cfg) if self.path else None + self.exclude = exclude + self._reader = ( + GtsFileReader(self.path, cfg=self.cfg, exclude=self.exclude) + if self.path + else None + ) self.store = GtsStore(self._reader) if self._reader else GtsStore(reader=None) # type: ignore[arg-type] @staticmethod @@ -345,7 +351,7 @@ def _load_config(self, config_path: str | None) -> GtsConfig: def reload_from_path(self, path: str | builtins.list[str]) -> None: self.path = path - self._reader = GtsFileReader(self.path, cfg=self.cfg) + self._reader = GtsFileReader(self.path, cfg=self.cfg, exclude=self.exclude) self.store = GtsStore(self._reader) def add_entity( diff --git a/tests/test_json_validation.py b/tests/test_json_validation.py index c8e9d2b..8ada971 100644 --- a/tests/test_json_validation.py +++ b/tests/test_json_validation.py @@ -1,5 +1,7 @@ import json +import pytest + from gts._cli import main from gts.entities import DEFAULT_GTS_CONFIG from gts._json_validation import GtsJsonValidator @@ -69,7 +71,9 @@ def test_validate_json_cli_outputs_json_only(tmp_path, capsys): input_path = tmp_path / "broken.json" input_path.write_text('{ "id": "gts.broken', encoding="utf-8") - main(["validate-all", "--path", str(input_path)]) + with pytest.raises(SystemExit) as exc_info: + main(["validate-all", "--path", str(input_path)]) + assert exc_info.value.code == 1 captured = capsys.readouterr() output = json.loads(captured.out)