[Git][debian-gis-team/stac-validator][master] 5 commits: New upstream version 4.6.1
Antonio Valentino (@antonio.valentino)
gitlab at salsa.debian.org
Sat Sep 5 10:05:26 BST 2026
Antonio Valentino pushed to branch master at Debian GIS Project / stac-validator
Commits:
2cebc8d5 by Antonio Valentino at 2026-09-05T08:35:44+00:00
New upstream version 4.6.1
- - - - -
5d0d1a9a by Antonio Valentino at 2026-09-05T08:36:06+00:00
Update upstream source from tag 'upstream/4.6.1'
Update to upstream version '4.6.1'
with Debian dir 974663c2076746e7666857fdc57679e66afaf0ed
- - - - -
0f0401d3 by Antonio Valentino at 2026-09-05T08:43:20+00:00
New usptream release
- - - - -
f3ecb473 by Antonio Valentino at 2026-09-05T08:52:09+00:00
Refresh all patches
- - - - -
e1febfc9 by Antonio Valentino at 2026-09-05T08:54:36+00:00
Set distribution to unstable
- - - - -
8 changed files:
- CHANGELOG.md
- debian/changelog
- debian/patches/0001-No-network.patch
- pyproject.toml
- stac_validator/fast_validator.py
- + tests/test_error_reporting.py
- + tests/test_fast_validator_enhancements.py
- tests/test_validate_item_collection.py
Changes:
=====================================
CHANGELOG.md
=====================================
@@ -6,13 +6,31 @@ The format is (loosely) based on [Keep a Changelog](http://keepachangelog.com/)
## [Unreleased]
+## [v4.6.1] - 2026-09-04
+
+### Fixed
+
+- **Multi-Extension Composition Errors**: Fixed false-positive validation errors when multiple extensions are composed together by unconditionally stripping `unevaluatedProperties: false` from extension schemas. Modern STAC extensions (e.g., `projection/v2.0.0`, `eo/v2.0.0`) use Draft 2020-12 `unevaluatedProperties: false` which causes sibling extension fields to be incorrectly flagged as unevaluated when multiple extensions are active.
+- **Branch Unroller for allOf-Nested oneOf**: Enhanced `compile_unrolled_schema()` to detect and unroll `oneOf`/`anyOf` branches nested inside `allOf` arrays (e.g., Projection v2.0.0 schema structure), preventing generic "must be valid exactly by one definition (0 matches found)" errors.
+- **Branch Validator Fallback**: Updated extension validation logic to use branch validator when primary validator fails, allowing accurate error reporting for composition-related failures while accepting valid items that pass branch validation.
+
+## [v4.6.0] - 2026-09-03
+
### Added
+- **Multi-Error Pass**: Accumulated error reporting now captures and reports all validation failures across all active extensions on a single item in one pass.
+- **RFC 6901 JSON Pointers**: Added JSON Pointer path translation (`parse_json_pointer()`) to format field failures into standard JSON paths (e.g., `$.properties.eo:cloud_cover`).
+- **Resilient Extension Compilation**: Introduced multi-tier schema patching to compile 100% of community extension schemas (including `file/v2.1.0`, `product/v1.0.0`, and `storage/v2.0.0`) without requiring compiler skips.
+
### Changed
+- **Clean Error Attribution**: Refactored error formatting to explicitly cite the failing schema source and field location (e.g., `[Extension: eo/v2.0.0] Field '$.properties.eo:cloud_cover': must be number`).
+- **Composite Schema Unrolling**: Pre-compiles `oneOf`/`anyOf` subschema branches independently to preserve C-speed execution while surfacing exact field failures.
+
### Fixed
-### Removed
+- **Swallowed Error Paths**: Fixed generic `$` root errors caused by `fastjsonschema`'s internal branch handling in composite extension schemas.
+- **Single-Error Short-Circuiting**: Fixed batch execution halting on the first field failure per item. [#308](https://github.com/stac-utils/stac-validator/pull/308)
## [v4.5.2] - 2026-08-05
@@ -490,7 +508,8 @@ The format is (loosely) based on [Keep a Changelog](http://keepachangelog.com/)
- With the newest version - 1.0.0-beta.2 - items will run through jsonchema validation before the PySTAC validation. The reason for this is that jsonschema will give more informative error messages. This should be addressed better in the future. This is not the case with the --recursive option as time can be a concern here with larger collections.
- Logging. Various additions were made here depending on the options selected. This was done to help assist people to update their STAC collections.
-[Unreleased]: https://github.com/sparkgeo/stac-validator/compare/v4.5.2..main
+[Unreleased]: https://github.com/sparkgeo/stac-validator/compare/v4.6.0..main
+[v4.6.0]: https://github.com/sparkgeo/stac-validator/compare/v4.5.2..v4.6.0
[v4.5.2]: https://github.com/sparkgeo/stac-validator/compare/v4.5.1..v4.5.2
[v4.5.1]: https://github.com/sparkgeo/stac-validator/compare/v4.5.0..v4.5.1
[v4.5.0]: https://github.com/sparkgeo/stac-validator/compare/v4.4.0..v4.5.0
=====================================
debian/changelog
=====================================
@@ -1,3 +1,11 @@
+stac-validator (4.6.1-1) unstable; urgency=medium
+
+ * New upstream release.
+ * debian/patches:
+ - Refresh all patches.
+
+ -- Antonio Valentino <antonio.valentino at tiscali.it> Sat, 05 Sep 2026 08:54:16 +0000
+
stac-validator (4.5.2-1) unstable; urgency=medium
* New upstream release.
=====================================
debian/patches/0001-No-network.patch
=====================================
@@ -26,7 +26,7 @@ Forwarded: not-needed
19 files changed, 135 insertions(+), 1 deletion(-)
diff --git a/pyproject.toml b/pyproject.toml
-index 2e37a0f..f9e1cdd 100644
+index a903785..5baf098 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -73,4 +73,9 @@ stac-validator = "stac_validator.stac_validator:cli"
@@ -970,7 +970,7 @@ index f02f3ba..08fb9b8 100644
stac = stac_validator.StacValidate()
with open("tests/test_data/1rc2/extensions-collection/collection.json", "r") as f:
diff --git a/tests/test_validate_item_collection.py b/tests/test_validate_item_collection.py
-index 2b413d8..6340c0c 100644
+index 20fa885..00e1fc7 100644
--- a/tests/test_validate_item_collection.py
+++ b/tests/test_validate_item_collection.py
@@ -3,9 +3,12 @@ Description: Test stac-validator on item-collection validation.
@@ -1010,11 +1010,8 @@ index 2b413d8..6340c0c 100644
def test_validate_item_collection_remote_pages_1_v110():
# Use a permanent historical collection (sentinel-2-l2a) instead of NRT
# NRT collections are dynamic and may have fewer items at test time
-@@ -588,6 +594,7 @@ def test_validate_item_collection_remote_pages_1_v110():
+@@ -586,3 +592,4 @@ def test_validate_item_collection_remote_pages_1_v110():
+ schema in msg["schema"] for schema in expected_schemas
+ ), f"Missing expected schemas in {msg['schema']}"
assert len(stac.message) == 10
-
-
-+ at pytest.mark.network
- def test_validate_item_collection_remote_pages_3_v110():
- # Fix: Point to a massive, permanent historical collection, NOT a temporary Near Real-Time (nrt) one.
- stac_file = (
++
=====================================
pyproject.toml
=====================================
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
[project]
name = "stac_valid"
-version = "4.5.2"
+version = "4.6.1"
description = "A package to validate STAC files"
authors = [
{name = "Jonathan Healy", email = "jon at healy-hyperspatial.dev"},
=====================================
stac_validator/fast_validator.py
=====================================
@@ -2,7 +2,9 @@ import io
import json
import logging
import os
-import sys
+import re
+import tempfile
+import threading
import time
from concurrent.futures import Future, ThreadPoolExecutor
from contextlib import redirect_stderr, redirect_stdout
@@ -19,16 +21,122 @@ from .utilities import validate_with_ref_resolver
# Standard Python logger for FastAPI/Uvicorn integration
logger = logging.getLogger(__name__)
-# --- Caches & Config ---
+
+def parse_json_pointer(expr_name: Optional[str]) -> str:
+ """Converts fastjsonschema variable names into clean JSON Pointers.
+
+ Handles both bracket notation (data['properties']['eo:cloud_cover'])
+ and dot notation (data.properties.eo:cloud_cover).
+ """
+ if not expr_name or not isinstance(expr_name, str) or expr_name == "data":
+ return "$"
+ # Bracket notation: data['properties']['eo:cloud_cover']
+ keys = re.findall(r"['\"]([^'\"]*)['\"]", expr_name)
+ if keys:
+ return "$." + ".".join(keys)
+ # Dot notation: data.properties.eo:cloud_cover
+ if expr_name.startswith("data."):
+ return "$." + expr_name[5:]
+ return expr_name
+
+
+class FastSTACValidationError(fastjsonschema.JsonSchemaValueException):
+ """Custom validation exception inheriting from fastjsonschema.JsonSchemaValueException.
+
+ Guarantees backward compatibility for legacy callers catching
+ fastjsonschema.JsonSchemaValueException.
+ """
+
+ def __init__(self, source: str, field_path: str, raw_message: str):
+ self.source = source
+ self.field_path = field_path
+ self.raw_message = raw_message
+ formatted_msg = f"[{source}] Field '{field_path}': {raw_message}"
+
+ # Populate fastjsonschema.JsonSchemaValueException superclass attributes:
+ # self.message -> formatted_msg
+ # self.name -> field_path
+ super().__init__(
+ message=formatted_msg,
+ value=None,
+ name=field_path,
+ definition=None,
+ )
+
+ def __str__(self) -> str:
+ return f"[{self.source}] Field '{self.field_path}': {self.raw_message}"
+
+
+class FastSTACMultiValidationError(FastSTACValidationError):
+ """Container for all validation errors on a single STAC object.
+
+ Inherits from FastSTACValidationError (and transitively JsonSchemaValueException),
+ allowing legacy exception handlers to catch multi-error failures seamlessly.
+ """
+
+ def __init__(self, errors: List[FastSTACValidationError]):
+ self.errors = errors
+ first_err = (
+ errors[0]
+ if errors
+ else FastSTACValidationError("Base STAC", "$", "Unknown validation error")
+ )
+ super().__init__(first_err.source, first_err.field_path, first_err.raw_message)
+
+ def __str__(self) -> str:
+ err_list_str = "; ".join(str(err) for err in self.errors)
+ return f"Found {len(self.errors)} validation error(s): {err_list_str}"
+
+
+def get_cache_directory() -> str:
+ r"""Determines a writable disk cache directory across environments (Docker, Lambda, CLI).
+
+ Priority:
+ 1. STAC_VALIDATOR_CACHE_DIR environment variable (explicit override)
+ 2. ~/.cache/stac_validator (standard user cache on Linux/macOS)
+ 3. %LOCALAPPDATA%\stac_validator (Windows user cache)
+ 4. tempfile.gettempdir()/stac_validator_cache (Docker/Lambda /tmp)
+
+ Returns:
+ Path to writable cache directory (guaranteed to exist or be creatable)
+ """
+ # 1. Respect explicit environment variable if set
+ env_dir = os.environ.get("STAC_VALIDATOR_CACHE_DIR")
+ if env_dir:
+ try:
+ os.makedirs(env_dir, exist_ok=True)
+ logger.debug(f"Using STAC_VALIDATOR_CACHE_DIR: {env_dir}")
+ return env_dir
+ except (OSError, PermissionError) as e:
+ logger.warning(f"Cannot write to STAC_VALIDATOR_CACHE_DIR ({env_dir}): {e}")
+
+ # 2. Try standard user cache directory
+ try:
+ user_cache = os.path.join(os.path.expanduser("~"), ".cache", "stac_validator")
+ os.makedirs(user_cache, exist_ok=True)
+ logger.debug(f"Using user cache directory: {user_cache}")
+ return user_cache
+ except (OSError, PermissionError) as e:
+ logger.debug(f"Cannot write to user cache ({user_cache}): {e}")
+
+ # 3. Fallback to system temp directory
+ temp_cache = os.path.join(tempfile.gettempdir(), "stac_validator_cache")
+ try:
+ os.makedirs(temp_cache, exist_ok=True)
+ logger.debug(f"Falling back to temp cache: {temp_cache}")
+ return temp_cache
+ except (OSError, PermissionError) as e:
+ logger.warning(f"Cannot write to temp cache ({temp_cache}): {e}")
+ # Return temp_cache anyway - fetch_schema will handle write failures gracefully
+ return temp_cache
+
+
+# --- Thread-Safe Caches & Lock ---
SCHEMA_CACHE: Dict[str, Any] = {}
VALIDATOR_CACHE: Dict[Any, Any] = {}
-QUIET_MODE: bool = False
-# Store cached schemas inside the repository under local_schemas/.schemas (project-root relative)
-LOCAL_SCHEMA_DIR = os.path.join(
- os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
- "local_schemas",
- ".schemas",
-)
+CACHE_LOCK = threading.Lock()
+# Dynamically determine writable cache directory across environments
+LOCAL_SCHEMA_DIR = get_cache_directory()
# Shared HTTP session with keep-alive connection pooling and retries for crawler workloads.
HTTP_SESSION = requests.Session()
@@ -53,12 +161,19 @@ def get_local_path_for_uri(uri: str) -> str:
return os.path.join(LOCAL_SCHEMA_DIR, safe_filename)
-def fetch_schema(uri: str) -> Dict[str, Any]:
- """The Ultimate Handler: RAM -> Disk -> Network -> Disk -> RAM"""
+def fetch_schema(uri: str, quiet: bool = False) -> Dict[str, Any]:
+ """The Ultimate Handler: RAM -> Disk -> Network -> Disk -> RAM
- # 1. RAM Cache
- if uri in SCHEMA_CACHE:
- return SCHEMA_CACHE[uri]
+ Thread-safe schema fetching with three-tier caching (RAM -> Disk -> Network).
+
+ Args:
+ uri: Schema URI to fetch
+ quiet: If True, suppress network fetch messages
+ """
+ # 1. RAM Cache (Thread-Safe Check)
+ with CACHE_LOCK:
+ if uri in SCHEMA_CACHE:
+ return SCHEMA_CACHE[uri]
local_path = get_local_path_for_uri(uri)
@@ -67,13 +182,14 @@ def fetch_schema(uri: str) -> Dict[str, Any]:
try:
with open(local_path, "r") as f:
schema_dict = json.load(f)
- SCHEMA_CACHE[uri] = schema_dict
+ with CACHE_LOCK:
+ SCHEMA_CACHE[uri] = schema_dict
return schema_dict
except Exception:
pass # If corrupted, fallback to network
# 3. Network Fetch
- if not QUIET_MODE:
+ if not quiet:
click.secho(f" [Network] Fetching: {uri}", fg="yellow", dim=True)
logger.debug(f"Network cache miss. Fetching schema: {uri}")
try:
@@ -83,40 +199,183 @@ def fetch_schema(uri: str) -> Dict[str, Any]:
except requests.RequestException as e:
raise RuntimeError(f"Could not resolve schema: {uri}. Reason: {e}")
- # 4. Save to Disk Cache
- os.makedirs(os.path.dirname(local_path), exist_ok=True)
+ # 4. Save to Disk Cache (Safeguarded)
+ # Fail gracefully if cache directory is unwritable (e.g., Docker/Lambda)
+ # RAM cache (SCHEMA_CACHE) still functions normally even if disk write fails
try:
+ os.makedirs(os.path.dirname(local_path), exist_ok=True)
with open(local_path, "w") as f:
json.dump(schema_dict, f)
- except IOError:
- pass # If we can't write to disk, no big deal, keep going
+ except (IOError, OSError, PermissionError):
+ # If we can't write to disk, no big deal - keep going with RAM cache
+ pass
- # 5. Save to RAM Cache
- SCHEMA_CACHE[uri] = schema_dict
+ # 5. Save to RAM Cache (Thread-Safe Store)
+ with CACHE_LOCK:
+ SCHEMA_CACHE[uri] = schema_dict
return schema_dict
-def optimize_schema_for_compiler(schema: Any, remove_allof: bool = False) -> Any:
+def compile_unrolled_schema(schema_dict: Dict[str, Any], quiet: bool = False) -> Any:
+ """Unrolls top-level or allOf-nested oneOf/anyOf branches into separate compiled
+ fastjsonschema functions so field-level errors are never swallowed by branch handling.
+
+ Handles two patterns:
+ 1. Direct top-level oneOf/anyOf
+ 2. oneOf/anyOf nested inside allOf (e.g., Projection v2.0.0)
+
+ Returns a validator function that tries each branch independently and reports
+ the deepest error found (most specific field path).
+
+ Args:
+ schema_dict: The schema to compile
+ quiet: If True, suppress network fetch messages
"""
- Recursively patches STAC schemas in-memory to bypass fastjsonschema code generation bugs.
- Strips problematic constructs that cause IndentationError when compiling complex schemas.
+
+ def handler(u: str) -> Dict[str, Any]:
+ return fetch_schema(u, quiet=quiet)
+
+ handlers_dict = {"http": handler, "https": handler}
+
+ target_keyword = None
+ branches = []
+ base_meta = {}
+
+ # Case 1: Direct top-level oneOf/anyOf
+ if "oneOf" in schema_dict or "anyOf" in schema_dict:
+ target_keyword = "oneOf" if "oneOf" in schema_dict else "anyOf"
+ branches = schema_dict[target_keyword]
+ base_meta = {
+ k: v for k, v in schema_dict.items() if k not in ("oneOf", "anyOf")
+ }
+
+ # Case 2: oneOf/anyOf nested inside allOf (e.g., Projection v2.0.0)
+ elif "allOf" in schema_dict and isinstance(schema_dict["allOf"], list):
+ other_allOf = []
+ for elem in schema_dict["allOf"]:
+ if isinstance(elem, dict) and ("oneOf" in elem or "anyOf" in elem):
+ target_keyword = "oneOf" if "oneOf" in elem else "anyOf"
+ branches = elem[target_keyword]
+ remainder = {
+ k: v for k, v in elem.items() if k not in ("oneOf", "anyOf")
+ }
+ if remainder:
+ other_allOf.append(remainder)
+ else:
+ other_allOf.append(elem)
+
+ if branches:
+ base_meta = {k: v for k, v in schema_dict.items() if k != "allOf"}
+ if other_allOf:
+ base_meta["allOf"] = other_allOf
+
+ # Compile each branch independently
+ if branches:
+ compiled_branches = []
+ for branch in branches:
+ if "allOf" in base_meta:
+ merged = {**base_meta, "allOf": base_meta["allOf"] + [branch]}
+ else:
+ merged = {**base_meta, **branch}
+
+ try:
+ # Tier 1: Standard optimization (keeps oneOf/allOf)
+ opt = optimize_schema_for_compiler(merged, remove_allof=False)
+ val = fastjsonschema.compile(opt, handlers=handlers_dict)
+ compiled_branches.append(val)
+ except Exception:
+ try:
+ # Tier 2: Aggressive optimization (strips oneOf/allOf)
+ opt = optimize_schema_for_compiler(merged, remove_allof=True)
+ val = fastjsonschema.compile(opt, handlers=handlers_dict)
+ compiled_branches.append(val)
+ except Exception:
+ pass
+
+ if compiled_branches:
+
+ def branch_validator(data: Dict[str, Any]) -> None:
+ best_err = None
+ max_depth = -1
+
+ for val in compiled_branches:
+ try:
+ val(data)
+ return
+ except fastjsonschema.JsonSchemaValueException as err:
+ clean_p = parse_json_pointer(err.name)
+ # Root '$' errors score 0; deeper paths score higher
+ depth = (
+ 0 if clean_p == "$" else len(re.split(r"[\[\.]", err.name))
+ )
+ if depth > max_depth:
+ max_depth = depth
+ best_err = err
+
+ if best_err:
+ raise best_err
+
+ return branch_validator
+
+ opt = optimize_schema_for_compiler(schema_dict, remove_allof=False)
+ return fastjsonschema.compile(opt, handlers=handlers_dict)
+
+
+def optimize_schema_for_compiler(
+ schema: Any,
+ remove_allof: bool = False,
+ in_props_branch: bool = False,
+ in_shared_props: bool = False,
+) -> Any:
+ r"""Recursively patches STAC schemas in-memory for fastjsonschema code generation.
+
+ Strips problematic constructs (like duration formats or dangling conditionals) and prunes
+ empty subschemas ({}) that cause CPython IndentationErrors during compilation.
+
+ Strips restrictive additionalProperties/unevaluatedProperties flags when traversing
+ shared STAC Item properties so active extensions don't reject sibling fields,
+ while preserving additionalProperties: false inside nested sub-objects (assets, links).
Args:
schema: The JSON schema dictionary to optimize
- remove_allof: If True, also remove allOf/oneOf/anyOf (used for all schemas)
+ remove_allof: If True, also remove allOf/oneOf/anyOf (used for aggressive patching)
+ in_props_branch: True if we are inside a properties block (first level)
+ in_shared_props: True if we are inside shared Item/Collection properties
"""
if isinstance(schema, list):
- return [optimize_schema_for_compiler(item, remove_allof) for item in schema]
+ cleaned_list = []
+ for item in schema:
+ opt_item = optimize_schema_for_compiler(
+ item, remove_allof, in_props_branch, in_shared_props
+ )
+ # Omit empty dictionaries inside composition lists (allOf, oneOf, anyOf)
+ if isinstance(opt_item, dict) and not opt_item:
+ continue
+ cleaned_list.append(opt_item)
+ return cleaned_list
if isinstance(schema, dict):
cleaned = {}
for k, v in schema.items():
- # BUG FIX 1: fastjsonschema crashes on the 'duration' format (fixes product extension)
+ # BUG FIX 1: fastjsonschema crashes on the 'duration' format
if k == "format" and v == "duration":
continue
- # BUG FIX 2: fastjsonschema writes invalid Python code (empty for/else blocks)
- # when translating complex JSON Schema conditionals (fixes file & storage extensions)
+ # BUG FIX 2: Strip unevaluatedProperties: false EVERYWHERE (Draft 2020-12 issue)
+ # Modern extension schemas use unevaluatedProperties: false which causes false positives
+ # when multiple extensions are composed. Each oneOf branch sees sibling extension fields
+ # (eo:cloud_cover, datetime, etc.) and rejects them as unevaluated, collapsing to
+ # "must be valid exactly by one definition (0 matches found)". Strip unconditionally.
+ if k == "unevaluatedProperties" and v is False:
+ continue
+
+ # BUG FIX 3: Strip additionalProperties: false ONLY in shared Item properties
+ # Preserve it in nested objects like assets, links, and sub-definitions
+ if k == "additionalProperties" and v is False:
+ if in_shared_props:
+ continue
+
+ # BUG FIX 4: Conditionals & dependencies that produce empty Python code blocks
if k in (
"if",
"then",
@@ -124,28 +383,91 @@ def optimize_schema_for_compiler(schema: Any, remove_allof: bool = False) -> Any
"dependencies",
"dependentRequired",
"dependentSchemas",
+ "$comment",
):
continue
- # BUG FIX 3: Remove allOf/oneOf/anyOf at top level when requested
- # These cause IndentationError in fastjsonschema's code generator
+ # BUG FIX 5: Remove allOf/oneOf/anyOf at top level when requested
if remove_allof and k in ("allOf", "oneOf", "anyOf") and len(schema) > 1:
- # Only skip if there are other validation keywords
continue
- cleaned[k] = optimize_schema_for_compiler(v, remove_allof)
+ # Context tracking across schema hierarchy
+ next_in_props_branch = in_props_branch
+ next_in_shared_props = in_shared_props
+
+ if k == "properties":
+ # Entering a properties block
+ if in_props_branch or in_shared_props:
+ # Already in properties context, mark as shared
+ next_in_shared_props = True
+ else:
+ # First properties block encountered
+ next_in_props_branch = True
+ elif k in ("assets", "links"):
+ # Nested objects - exit shared properties context
+ next_in_shared_props = False
+ elif k in ("$defs", "definitions"):
+ # Definitions are part of shared properties context
+ next_in_shared_props = True
+
+ opt_v = optimize_schema_for_compiler(
+ v, remove_allof, next_in_props_branch, next_in_shared_props
+ )
+
+ # BUG FIX 6: Prune empty subschemas ({}) in properties & patternProperties
+ # BUG FIX 5: Prune empty subschemas ({}) in properties & patternProperties
+ # to prevent fastjsonschema from generating empty for/else blocks
+ if k in ("patternProperties", "properties", "dependentSchemas"):
+ if isinstance(opt_v, dict):
+ non_empty_props = {
+ pk: pv
+ for pk, pv in opt_v.items()
+ if not (isinstance(pv, dict) and not pv)
+ }
+ if not non_empty_props:
+ continue
+ opt_v = non_empty_props
+
+ if k in ("items", "additionalProperties", "unevaluatedProperties"):
+ if isinstance(opt_v, dict) and not opt_v:
+ continue
+
+ cleaned[k] = opt_v
+
+ # Clean up empty composition keyword arrays
+ for comp_key in ("allOf", "oneOf", "anyOf"):
+ if (
+ comp_key in cleaned
+ and isinstance(cleaned[comp_key], list)
+ and not cleaned[comp_key]
+ ):
+ del cleaned[comp_key]
+
return cleaned
return schema
-def get_validator(stac_type: str, stac_version: str, extensions: List[str]):
- """Builds and caches a validator based on Object Type, Version, and Extensions."""
+def get_validator(
+ stac_type: str, stac_version: str, extensions: List[str], quiet: bool = False
+):
+ """Builds and caches a validator based on Object Type, Version, and Extensions.
+
+ Thread-safe validator compilation and caching.
+
+ Args:
+ stac_type: STAC object type (item, collection, catalog)
+ stac_version: STAC version (e.g., "1.0.0")
+ extensions: List of extension URIs
+ quiet: If True, suppress network fetch and compilation messages
+ """
ext_key = tuple(sorted(extensions))
cache_key = (stac_type, stac_version, ext_key)
- if cache_key in VALIDATOR_CACHE:
- return VALIDATOR_CACHE[cache_key], True
+ # Thread-safe Cache Read
+ with CACHE_LOCK:
+ if cache_key in VALIDATOR_CACHE:
+ return VALIDATOR_CACHE[cache_key], True
# Determine base schema URI
stac_type_lower = stac_type.lower()
@@ -159,16 +481,18 @@ def get_validator(stac_type: str, stac_version: str, extensions: List[str]):
raise ValueError(f"Unknown STAC type for validation: {stac_type}")
# Fetch the raw Base Schema directly
- raw_base_schema = fetch_schema(base_uri)
+ raw_base_schema = fetch_schema(base_uri, quiet=quiet)
+
+ def handler(u: str) -> Dict[str, Any]:
+ return fetch_schema(u, quiet=quiet)
+
+ handlers_dict = {"http": handler, "https": handler}
try:
- # Tier 1: Try to compile with standard patching first
- optimized_base = optimize_schema_for_compiler(raw_base_schema)
- base_validator = fastjsonschema.compile(
- optimized_base, handlers={"http": fetch_schema, "https": fetch_schema}
- )
+ # Tier 1: Try to compile with unrolled oneOf/anyOf branches first
+ base_validator = compile_unrolled_schema(raw_base_schema, quiet=quiet)
logger.debug(
- f"Base schema {stac_type} {stac_version} compiled with fastjsonschema"
+ f"Base schema {stac_type} {stac_version} compiled with unrolled branch compilation"
)
except Exception:
try:
@@ -177,7 +501,7 @@ def get_validator(stac_type: str, stac_version: str, extensions: List[str]):
raw_base_schema, remove_allof=True
)
base_validator = fastjsonschema.compile(
- optimized_base, handlers={"http": fetch_schema, "https": fetch_schema}
+ optimized_base, handlers=handlers_dict
)
logger.debug(
f"Base schema {stac_type} {stac_version} compiled with fastjsonschema (aggressive patching)"
@@ -191,7 +515,7 @@ def get_validator(stac_type: str, stac_version: str, extensions: List[str]):
resolver = jsonschema.RefResolver(
base_uri=base_uri,
referrer=raw_base_schema,
- handlers={"http": fetch_schema, "https": fetch_schema},
+ handlers=handlers_dict,
)
ValidatorClass = jsonschema.validators.validator_for(raw_base_schema)
@@ -205,14 +529,14 @@ def get_validator(stac_type: str, stac_version: str, extensions: List[str]):
f"Base schema {stac_type} {stac_version} compiled with cached jsonschema fallback"
)
- ext_validators = []
+ ext_validators: List[Tuple[str, Any, Any]] = []
skipped_extensions = []
if extensions:
logger.info(
f"Warming STAC Validator Cache: Compiling {len(extensions)} extension(s) for {stac_type} {stac_version}..."
)
- if not QUIET_MODE:
+ if not quiet:
click.secho(
f" [Extensions] Compiling {len(extensions)} extension(s):",
fg="cyan",
@@ -221,44 +545,70 @@ def get_validator(stac_type: str, stac_version: str, extensions: List[str]):
for ext in extensions:
try:
- # 1. Fetch the raw dictionary
- raw_ext_schema = fetch_schema(ext)
+ raw_ext_schema = fetch_schema(ext, quiet=quiet)
- # 2. Try to compile without patching first
+ # 1. Primary validator compilation with graceful fallback
+ ext_val = None
try:
+ # Tier 1a: Raw unpatched schema
ext_val = fastjsonschema.compile(
raw_ext_schema,
- handlers={"http": fetch_schema, "https": fetch_schema},
+ handlers=handlers_dict,
)
except Exception:
- # If compilation fails, try with aggressive patching (remove allOf/oneOf/anyOf)
- optimized_schema = optimize_schema_for_compiler(
- raw_ext_schema, remove_allof=True
- )
- ext_val = fastjsonschema.compile(
- optimized_schema,
- handlers={"http": fetch_schema, "https": fetch_schema},
- )
+ try:
+ # Tier 1b: Standard optimization (strips duration format, if/then/else, keeps oneOf/allOf)
+ opt_schema = optimize_schema_for_compiler(
+ raw_ext_schema, remove_allof=False
+ )
+ ext_val = fastjsonschema.compile(
+ opt_schema,
+ handlers=handlers_dict,
+ )
+ except Exception:
+ # Tier 1c: Aggressive optimization (strips top-level allOf/oneOf as last resort)
+ opt_schema_aggr = optimize_schema_for_compiler(
+ raw_ext_schema, remove_allof=True
+ )
+ ext_val = fastjsonschema.compile(
+ opt_schema_aggr,
+ handlers=handlers_dict,
+ )
+
+ # 2. Pre-compile unrolled branch validator as a lazy diagnostic backup
+ branch_val = None
+ has_branches = "oneOf" in raw_ext_schema or "anyOf" in raw_ext_schema
+ # Also check for oneOf/anyOf nested inside allOf (e.g., Projection v2.0.0)
+ if (
+ not has_branches
+ and "allOf" in raw_ext_schema
+ and isinstance(raw_ext_schema["allOf"], list)
+ ):
+ for elem in raw_ext_schema["allOf"]:
+ if isinstance(elem, dict) and ("oneOf" in elem or "anyOf" in elem):
+ has_branches = True
+ break
+
+ if has_branches:
+ try:
+ branch_val = compile_unrolled_schema(raw_ext_schema, quiet=quiet)
+ except Exception:
+ pass
- ext_validators.append(ext_val)
+ ext_validators.append((ext, ext_val, branch_val))
logger.debug(f"Successfully compiled STAC extension: {ext}")
- if not QUIET_MODE:
+ if not quiet:
click.secho(f" ✅ {ext}", fg="green", dim=True)
+
except Exception as e:
- # Log to standard Python logging for FastAPI/Uvicorn integration
logger.warning(
f"Skipped extension due to compiler incompatibility: {ext} - {type(e).__name__}: {str(e)[:100]}"
)
- # Safety net for genuinely broken URLs or unfixable schemas
- if not QUIET_MODE:
- click.secho(
- f" ❌ {ext}: {type(e).__name__}",
- fg="red",
- dim=True,
- )
+ if not quiet:
+ click.secho(f" ❌ {ext}: {type(e).__name__}", fg="red", dim=True)
skipped_extensions.append(ext)
- if skipped_extensions and not QUIET_MODE:
+ if skipped_extensions and not quiet:
click.secho(
f" [Warning] Skipped {len(skipped_extensions)} extension(s) due to fastjsonschema incompatibility:",
fg="yellow",
@@ -273,18 +623,49 @@ def get_validator(stac_type: str, stac_version: str, extensions: List[str]):
)
def validator(data: Dict[str, Any]) -> None:
- old_limit = sys.getrecursionlimit()
- sys.setrecursionlimit(10000)
+ collected_errors: List[FastSTACValidationError] = []
+
+ # 1. Base STAC Schema
try:
- # Execute the pre-compiled native Python functions
base_validator(data)
- for ext_val in ext_validators:
+ except fastjsonschema.JsonSchemaValueException as e:
+ clean_path = parse_json_pointer(e.name)
+ msg = e.message.replace(e.name, "").strip()
+ collected_errors.append(
+ FastSTACValidationError(f"Base {stac_type}", clean_path, msg)
+ )
+
+ # 2. Individual Extension Schemas (evaluate ALL extensions)
+ for ext_uri, ext_val, branch_val in ext_validators:
+ try:
ext_val(data)
- finally:
- sys.setrecursionlimit(old_limit)
+ except fastjsonschema.JsonSchemaValueException as e:
+ clean_path = parse_json_pointer(e.name)
+ msg = e.message.replace(e.name, "").strip()
+
+ # If we have a branch validator, try it to unmask composition errors
+ # (e.g., oneOf/anyOf where fastjsonschema swallows the real error)
+ if branch_val is not None:
+ try:
+ branch_val(data)
+ # Branch validator passed - item is valid, don't report error
+ continue
+ except fastjsonschema.JsonSchemaValueException as branch_e:
+ # Branch validator also failed - use its more specific error
+ clean_path = parse_json_pointer(branch_e.name)
+ msg = branch_e.message.replace(branch_e.name, "").strip()
+
+ collected_errors.append(
+ FastSTACValidationError(f"Extension: {ext_uri}", clean_path, msg)
+ )
- # Cache the resulting validator so future items use it instantly
- VALIDATOR_CACHE[cache_key] = validator
+ # Raise all accumulated errors at the end of the item pass
+ if collected_errors:
+ raise FastSTACMultiValidationError(collected_errors)
+
+ # Cache the resulting validator so future items use it instantly (Thread-safe Write)
+ with CACHE_LOCK:
+ VALIDATOR_CACHE[cache_key] = validator
return validator, False
@@ -297,7 +678,6 @@ class FastValidator:
limit: Optional[int] = None,
validate_geometry: bool = False,
):
- global QUIET_MODE
self.stac_file = stac_file
self.quiet = quiet
self.valid = True
@@ -305,7 +685,6 @@ class FastValidator:
self.limit = limit
self.validate_geometry = validate_geometry
self.message: List[Dict[str, Any]] = []
- QUIET_MODE = quiet
def _validate_datetime_range(self, data: Dict[str, Any]) -> None:
"""Ensures start_datetime is not strictly after end_datetime per STAC Spec.
@@ -511,6 +890,101 @@ class FastValidator:
]
return [future.result() for future in futures]
+ def _validate_single_item(
+ self, item: Dict[str, Any], item_index: int
+ ) -> Tuple[bool, float, float, str, List[str]]:
+ """Validate a single STAC item and return (is_valid, setup_ms, exec_ms, item_id, error_messages).
+
+ This helper eliminates code duplication between run() and run_dict() by providing
+ a unified validation pipeline for individual items.
+
+ Returns:
+ Tuple of (is_valid, setup_time_ms, exec_time_ms, item_id, error_messages)
+ """
+ item_id = item.get("id", f"unknown-{item_index}")
+ stac_version = item.get("stac_version", "1.0.0")
+ extensions = item.get("stac_extensions", [])
+
+ # Map Feature->Item, others keep their type
+ actual_type = (
+ "Item" if item.get("type") == "Feature" else item.get("type", "Catalog")
+ )
+
+ # --- Setup Timer ---
+ t0 = time.perf_counter()
+ try:
+ validator, _ = get_validator(
+ actual_type, stac_version, extensions, quiet=self.quiet
+ )
+ except Exception as e:
+ t1 = time.perf_counter()
+ setup_time = (t1 - t0) * 1000
+ error_msg = f"Setup failed: {str(e)}"
+ return False, setup_time, 0.0, item_id, [error_msg]
+
+ t1 = time.perf_counter()
+ setup_time = (t1 - t0) * 1000
+
+ # --- Execution Timer ---
+ t2 = time.perf_counter()
+ error_messages: List[str] = []
+
+ try:
+ validator(item)
+ # Run logical firewalls
+ self._validate_datetime_range(item)
+ if self.validate_geometry:
+ self._validate_geometry(item)
+ t3 = time.perf_counter()
+ exec_time = (t3 - t2) * 1000
+ return True, setup_time, exec_time, item_id, error_messages
+
+ except FastSTACMultiValidationError as e:
+ t3 = time.perf_counter()
+ exec_time = (t3 - t2) * 1000
+ error_messages = [str(single_err) for single_err in e.errors]
+ return False, setup_time, exec_time, item_id, error_messages
+
+ except FastSTACValidationError as e:
+ t3 = time.perf_counter()
+ exec_time = (t3 - t2) * 1000
+ error_messages = [str(e)]
+ return False, setup_time, exec_time, item_id, error_messages
+
+ except fastjsonschema.JsonSchemaValueException as e:
+ t3 = time.perf_counter()
+ exec_time = (t3 - t2) * 1000
+ error_msg = f"{e.name} {e.message.replace(e.name, '').strip()}"
+ if "disallowed definition" in error_msg and "collection" in error_msg:
+ error_msg = (
+ "STAC Spec Violation: Missing {'rel': 'collection'} in links array."
+ )
+ error_messages = [error_msg]
+ return False, setup_time, exec_time, item_id, error_messages
+
+ except ValueError as e:
+ t3 = time.perf_counter()
+ exec_time = (t3 - t2) * 1000
+ error_messages = [str(e)]
+ return False, setup_time, exec_time, item_id, error_messages
+
+ except Exception as e:
+ t3 = time.perf_counter()
+ exec_time = (t3 - t2) * 1000
+
+ if self._is_ref_resolution_error(e):
+ try:
+ self._validate_with_jsonschema_fallback(
+ item, actual_type, stac_version, extensions
+ )
+ return True, setup_time, exec_time, item_id, error_messages
+ except Exception as fallback_err:
+ error_messages = [str(fallback_err)]
+ return False, setup_time, exec_time, item_id, error_messages
+ else:
+ error_messages = [str(e)]
+ return False, setup_time, exec_time, item_id, error_messages
+
def run(self):
"""Universal high-speed STAC Validator (Items, Collections, Catalogs, FeatureCollections)"""
if not self.quiet:
@@ -581,19 +1055,20 @@ class FastValidator:
schemas_checked: Set[str] = set()
for index, item in enumerate(items_to_validate):
- # Determine specific STAC attributes for this object
- item_id = item.get("id", f"unknown-{index}")
- stac_version = item.get("stac_version", "1.0.0")
- extensions = item.get("stac_extensions", [])
+ # Use unified validation helper
+ is_valid, setup_time, exec_time, item_id, error_messages = (
+ self._validate_single_item(item, index)
+ )
# Track versions and schemas
- stac_versions_found.add(stac_version)
-
- # Map Feature->Item, others keep their type
+ stac_version = item.get("stac_version", "1.0.0")
+ extensions = item.get("stac_extensions", [])
actual_type = (
"Item" if item.get("type") == "Feature" else item.get("type", "Catalog")
)
+ stac_versions_found.add(stac_version)
+
# Build schema URI for this object type
try:
base_schema = self._get_base_schema_uri(actual_type, stac_version)
@@ -607,114 +1082,28 @@ class FastValidator:
for ext in extensions:
schemas_checked.add(ext)
- # --- Setup Timer ---
- t0 = time.perf_counter()
- try:
- validator, is_cached = get_validator(
- actual_type, stac_version, extensions
- )
- except Exception as e:
- if not self.quiet:
- click.secho(f"❌ Setup failed for {item_id}: {e}", fg="red")
- invalid_count += 1
- self.valid = False
- error_msg = f"Setup failed: {str(e)}"
- if error_msg not in error_registry:
- error_registry[error_msg] = []
- error_registry[error_msg].append(item_id)
- continue
- t1 = time.perf_counter()
- setup_time = (t1 - t0) * 1000
+ # Accumulate metrics
total_setup_ms += setup_time
+ total_exec_ms += exec_time
- # --- Execution Timer ---
- t2 = time.perf_counter()
- try:
- validator(item)
- # Run logical firewalls
- self._validate_datetime_range(item)
- if self.validate_geometry:
- self._validate_geometry(item)
- t3 = time.perf_counter()
- exec_time = (t3 - t2) * 1000
- total_exec_ms += exec_time
+ if is_valid:
valid_count += 1
status_text = click.style("✅ VALID", fg="green")
-
- except fastjsonschema.JsonSchemaValueException as e:
- t3 = time.perf_counter()
- exec_time = (t3 - t2) * 1000
- total_exec_ms += exec_time
- invalid_count += 1
- self.valid = False
-
- # --- The STAC Error Translator ---
- error_msg = f"{e.name} {e.message.replace(e.name, '').strip()}"
- if "disallowed definition" in error_msg:
- if "collection" in error_msg:
- error_msg = "STAC Spec Violation: Missing {'rel': 'collection'} in links array."
- else:
- error_msg = (
- f"{e.name} violated a 'not' rule. Value: {repr(e.value)}"
- )
-
- # Group errors
- if error_msg not in error_registry:
- error_registry[error_msg] = []
- error_registry[error_msg].append(item_id)
- status_text = click.style("❌ INVALID", fg="red")
-
- except ValueError as e:
- t3 = time.perf_counter()
- exec_time = (t3 - t2) * 1000
- total_exec_ms += exec_time
+ else:
invalid_count += 1
self.valid = False
-
- # Logical validation errors (datetime range, geometry)
- error_msg = str(e)
- if error_msg not in error_registry:
- error_registry[error_msg] = []
- error_registry[error_msg].append(item_id)
status_text = click.style("❌ INVALID", fg="red")
-
- except Exception as e:
- t3 = time.perf_counter()
- exec_time = (t3 - t2) * 1000
- total_exec_ms += exec_time
-
- if self._is_ref_resolution_error(e):
- try:
- self._validate_with_jsonschema_fallback(
- item,
- actual_type,
- stac_version,
- extensions,
- )
- valid_count += 1
- status_text = click.style("✅ VALID", fg="green")
- except Exception as fallback_err:
- invalid_count += 1
- self.valid = False
- error_msg = str(fallback_err)
- if error_msg not in error_registry:
- error_registry[error_msg] = []
- error_registry[error_msg].append(item_id)
- status_text = click.style("❌ INVALID", fg="red")
- else:
- invalid_count += 1
- self.valid = False
- error_msg = str(e)
+ # Register all errors for this item
+ for error_msg in error_messages:
if error_msg not in error_registry:
error_registry[error_msg] = []
error_registry[error_msg].append(item_id)
- status_text = click.style("❌ INVALID", fg="red")
if not self.quiet:
if self.verbose or index < 5 or (len(items_to_validate) < 20):
- cache_icon = "⚡" if is_cached else "🐌"
+ # Note: is_cached info is not available from helper, use placeholder
click.echo(
- f"[{index + 1}] ID: {item_id} | Type: {actual_type} | Cache {cache_icon} | Setup: {setup_time:>6.2f}ms | Exec: {exec_time:>5.2f}ms | {status_text}"
+ f"[{index + 1}] ID: {item_id} | Type: {actual_type} | Setup: {setup_time:>6.2f}ms | Exec: {exec_time:>5.2f}ms | {status_text}"
)
elif index == 5:
click.secho(
@@ -844,16 +1233,20 @@ class FastValidator:
self.valid = True
for index, item in enumerate(items_to_validate):
- item_id = item.get("id", f"unknown-{index}")
+ # Use unified validation helper
+ is_valid, setup_time, exec_time, item_id, error_messages = (
+ self._validate_single_item(item, index)
+ )
+
+ # Track versions and schemas
stac_version = item.get("stac_version", "1.0.0")
extensions = item.get("stac_extensions", [])
-
- stac_versions_found.add(stac_version)
-
actual_type = (
"Item" if item.get("type") == "Feature" else item.get("type", "Catalog")
)
+ stac_versions_found.add(stac_version)
+
try:
base_schema = self._get_base_schema_uri(actual_type, stac_version)
except ValueError:
@@ -865,74 +1258,17 @@ class FastValidator:
for ext in extensions:
schemas_checked.add(ext)
- t0 = time.perf_counter()
- try:
- validator, _ = get_validator(actual_type, stac_version, extensions)
- except Exception as e:
- invalid_count += 1
- self.valid = False
- error_msg = str(e)
- logger.error(f"Schema setup failed for item {item_id}: {error_msg}")
- if error_msg not in error_registry:
- error_registry[error_msg] = []
- error_registry[error_msg].append(item_id)
- continue
- t1 = time.perf_counter()
- total_setup_ms += (t1 - t0) * 1000
+ # Accumulate metrics
+ total_setup_ms += setup_time
+ total_exec_ms += exec_time
- t2 = time.perf_counter()
- try:
- validator(item)
- # Run logical firewalls
- self._validate_datetime_range(item)
- if self.validate_geometry:
- self._validate_geometry(item)
- t3 = time.perf_counter()
- total_exec_ms += (t3 - t2) * 1000
+ if is_valid:
valid_count += 1
- except fastjsonschema.JsonSchemaValueException as e:
- t3 = time.perf_counter()
- total_exec_ms += (t3 - t2) * 1000
- invalid_count += 1
- self.valid = False
- error_msg = f"{e.name} {e.message.replace(e.name, '').strip()}"
- if "disallowed definition" in error_msg and "collection" in error_msg:
- error_msg = "STAC Spec Violation: Missing {'rel': 'collection'} in links array."
- if error_msg not in error_registry:
- error_registry[error_msg] = []
- error_registry[error_msg].append(item_id)
- except ValueError as e:
- t3 = time.perf_counter()
- total_exec_ms += (t3 - t2) * 1000
+ else:
invalid_count += 1
self.valid = False
- error_msg = str(e)
- if error_msg not in error_registry:
- error_registry[error_msg] = []
- error_registry[error_msg].append(item_id)
- except Exception as e:
- t3 = time.perf_counter()
- total_exec_ms += (t3 - t2) * 1000
- if self._is_ref_resolution_error(e):
- try:
- self._validate_with_jsonschema_fallback(
- item,
- actual_type,
- stac_version,
- extensions,
- )
- valid_count += 1
- except Exception as fallback_err:
- invalid_count += 1
- self.valid = False
- error_msg = str(fallback_err)
- if error_msg not in error_registry:
- error_registry[error_msg] = []
- error_registry[error_msg].append(item_id)
- else:
- invalid_count += 1
- self.valid = False
- error_msg = str(e)
+ # Register all errors for this item
+ for error_msg in error_messages:
if error_msg not in error_registry:
error_registry[error_msg] = []
error_registry[error_msg].append(item_id)
@@ -962,7 +1298,6 @@ class FastValidator:
def run_recursive(self):
"""Recursively validate a local STAC catalog/collection and all its children."""
- sys.setrecursionlimit(10000)
start_time = time.perf_counter()
# Load the root STAC object
@@ -982,7 +1317,9 @@ class FastValidator:
results = []
visited = set()
visited.add(root_path)
- self._validate_recursive(root_data, root_path, results, visited, is_api=False)
+ self._validate_recursive(
+ root_data, root_path, results, visited, is_api=False, depth=0
+ )
if self.limit is not None and not self.quiet and len(results) >= self.limit:
click.secho(
@@ -1042,7 +1379,6 @@ class FastValidator:
def run_api(self):
"""Recursively validate a STAC API catalog and all its collections/items."""
- sys.setrecursionlimit(10000)
start_time = time.perf_counter()
if not self.quiet:
@@ -1077,7 +1413,9 @@ class FastValidator:
dim=True,
)
- self._validate_recursive(root_data, root_path, results, visited, is_api=True)
+ self._validate_recursive(
+ root_data, root_path, results, visited, is_api=True, depth=0
+ )
if self.limit is not None and not self.quiet and len(results) >= self.limit:
click.secho(
@@ -1144,6 +1482,7 @@ class FastValidator:
is_api: bool = False,
collection_id: Optional[str] = None,
prefetched_resources: Optional[Dict[str, Dict[str, Any]]] = None,
+ depth: int = 0,
):
"""Recursively validate a STAC object and its children.
@@ -1154,7 +1493,20 @@ class FastValidator:
visited: Set of already-visited paths to prevent circular references
is_api: If True, follow API-specific links (data, items, next); if False, follow catalog links (child, item)
collection_id: Optional collection ID for items from FeatureCollections
+ depth: Current recursion depth (0 at root)
"""
+ # Protect against deeply nested catalog structures
+ MAX_RECURSION_DEPTH = 250
+ if depth > MAX_RECURSION_DEPTH:
+ results.append(
+ {
+ "path": file_path,
+ "valid_stac": False,
+ "error_message": f"Maximum catalog depth ({MAX_RECURSION_DEPTH}) exceeded.",
+ }
+ )
+ return
+
if self._limit_reached(results):
return
@@ -1203,11 +1555,20 @@ class FastValidator:
# Mute noisy "[Fallback]" and "[Network]" prints from validation execution path
with redirect_stdout(io.StringIO()), redirect_stderr(io.StringIO()):
- validator, _ = get_validator(stac_type, stac_version, extensions)
+ validator, _ = get_validator(
+ stac_type, stac_version, extensions, quiet=self.quiet
+ )
validator(data)
is_valid = True
error_msg = None
+ except FastSTACMultiValidationError as e:
+ is_valid = False
+ # Aggregate all errors into a single message for display
+ error_msg = "; ".join(str(err) for err in e.errors)
+ except FastSTACValidationError as e:
+ is_valid = False
+ error_msg = str(e)
except fastjsonschema.JsonSchemaValueException as e:
is_valid = False
error_msg = f"{e.name} {e.message.replace(e.name, '').strip()}"
@@ -1351,11 +1712,17 @@ class FastValidator:
visited,
is_api,
prefetched_resources=prefetched_collection_resources,
+ depth=depth + 1,
)
else:
# Not a collections list, validate as normal
self._validate_recursive(
- child_data, child_path, results, visited, is_api
+ child_data,
+ child_path,
+ results,
+ visited,
+ is_api,
+ depth=depth + 1,
)
# If this is an items endpoint (GeoJSON FeatureCollection), validate only Features
elif rel == "items" and is_api and isinstance(child_data, dict):
@@ -1386,11 +1753,17 @@ class FastValidator:
visited,
is_api,
collection_id_from_items,
+ depth=depth + 1,
)
else:
# Recursively validate child
self._validate_recursive(
- child_data, child_path, results, visited, is_api
+ child_data,
+ child_path,
+ results,
+ visited,
+ is_api,
+ depth=depth + 1,
)
except Exception as e:
if self._limit_reached(results):
=====================================
tests/test_error_reporting.py
=====================================
@@ -0,0 +1,641 @@
+"""Tests for precision error reporting improvements (v4.6.0) in FastValidator.
+
+These tests cover the --fast validator mode enhancements including:
+- RFC 6901 JSON Pointer path normalization
+- Multi-error accumulation across extensions
+- Extension attribution in error messages
+- Compile-time branch unrolling for precise field paths
+
+Network-isolated tests with mocked schema caches to prevent HTTP calls during CI/CD.
+"""
+
+import json
+
+import pytest
+
+from stac_validator import fast_validator
+from stac_validator.fast_validator import (
+ FastSTACMultiValidationError,
+ FastSTACValidationError,
+ FastValidator,
+ parse_json_pointer,
+)
+
+
+ at pytest.fixture(autouse=True)
+def mock_schema_cache(monkeypatch):
+ """Pre-populates SCHEMA_CACHE and restores original state after test completion.
+
+ This fixture ensures tests run offline and are not flaky due to network issues,
+ while preventing cache leakage across test suites.
+ """
+ # Save original cache state for restoration
+ orig_schema_cache = fast_validator.SCHEMA_CACHE.copy()
+ orig_validator_cache = fast_validator.VALIDATOR_CACHE.copy()
+
+ fast_validator.SCHEMA_CACHE.clear()
+ fast_validator.VALIDATOR_CACHE.clear()
+
+ # Minimal offline base item schema
+ base_item_schema = {
+ "$schema": "http://json-schema.org/draft-07/schema#",
+ "type": "object",
+ "required": [
+ "stac_version",
+ "type",
+ "id",
+ "geometry",
+ "properties",
+ "links",
+ "assets",
+ ],
+ "properties": {
+ "stac_version": {"type": "string"},
+ "type": {"const": "Feature"},
+ "id": {"type": "string"},
+ "geometry": {"type": ["object", "null"]},
+ "properties": {
+ "type": "object",
+ "properties": {
+ "datetime": {"type": ["string", "null"]},
+ "gsd": {"type": "number"},
+ },
+ },
+ "links": {"type": "array"},
+ "assets": {"type": "object"},
+ },
+ }
+
+ # Synthetic extension schema containing top-level oneOf composition
+ eo_oneof_schema = {
+ "$schema": "http://json-schema.org/draft-07/schema#",
+ "type": "object",
+ "oneOf": [
+ {
+ "properties": {
+ "properties": {
+ "type": "object",
+ "properties": {
+ "eo:cloud_cover": {
+ "type": "number",
+ "minimum": 0,
+ "maximum": 100,
+ }
+ },
+ "required": ["eo:cloud_cover"],
+ }
+ }
+ }
+ ],
+ }
+
+ fast_validator.SCHEMA_CACHE.update(
+ {
+ "https://schemas.stacspec.org/v1.0.0/item-spec/json-schema/item.json": base_item_schema,
+ "https://stac-extensions.github.io/eo/v1.0.0/schema.json": eo_oneof_schema,
+ }
+ )
+
+ yield
+
+ # Restore pre-test cache state to prevent leakage
+ fast_validator.SCHEMA_CACHE.clear()
+ fast_validator.SCHEMA_CACHE.update(orig_schema_cache)
+ fast_validator.VALIDATOR_CACHE.clear()
+ fast_validator.VALIDATOR_CACHE.update(orig_validator_cache)
+
+
+class TestParseJsonPointer:
+ """Test parse_json_pointer function for RFC 6901 JSON Pointer conversion.
+
+ FastValidator utility for normalizing fastjsonschema variable expressions
+ (both bracket and dot notation) into standard JSON Pointers.
+ """
+
+ def test_bracket_notation_single_key(self):
+ """Test conversion of bracket notation with single key."""
+ result = parse_json_pointer("data['properties']")
+ assert result == "$.properties"
+
+ def test_bracket_notation_multiple_keys(self):
+ """Test conversion of bracket notation with multiple keys."""
+ result = parse_json_pointer("data['properties']['eo:cloud_cover']")
+ assert result == "$.properties.eo:cloud_cover"
+
+ def test_dot_notation_single_key(self):
+ """Test conversion of dot notation with single key."""
+ result = parse_json_pointer("data.properties")
+ assert result == "$.properties"
+
+ def test_dot_notation_multiple_keys(self):
+ """Test conversion of dot notation with multiple keys."""
+ result = parse_json_pointer("data.properties.eo:cloud_cover")
+ assert result == "$.properties.eo:cloud_cover"
+
+ def test_root_data_only(self):
+ """Test that 'data' alone returns '$'."""
+ result = parse_json_pointer("data")
+ assert result == "$"
+
+ def test_empty_string(self):
+ """Test that empty string returns '$'."""
+ result = parse_json_pointer("")
+ assert result == "$"
+
+ def test_none_input(self):
+ """Test that None returns '$'."""
+ result = parse_json_pointer(None)
+ assert result == "$"
+
+ def test_double_quote_bracket_notation(self):
+ """Test bracket notation with double quotes."""
+ result = parse_json_pointer('data["properties"]["eo:cloud_cover"]')
+ assert result == "$.properties.eo:cloud_cover"
+
+ def test_mixed_quote_styles(self):
+ """Test bracket notation with mixed quote styles."""
+ result = parse_json_pointer("data['properties'][\"eo:cloud_cover\"]")
+ assert result == "$.properties.eo:cloud_cover"
+
+
+class TestFastSTACValidationError:
+ """Test FastSTACValidationError exception class."""
+
+ def test_error_formatting(self):
+ """Test error message formatting."""
+ err = FastSTACValidationError(
+ "Extension: https://stac-extensions.github.io/eo/v1.0.0/schema.json",
+ "$.properties.eo:cloud_cover",
+ "must be number",
+ )
+ expected = (
+ "[Extension: https://stac-extensions.github.io/eo/v1.0.0/schema.json] "
+ "Field '$.properties.eo:cloud_cover': must be number"
+ )
+ assert str(err) == expected
+
+ def test_error_attributes(self):
+ """Test error object attributes."""
+ err = FastSTACValidationError(
+ "Base Item",
+ "$.properties.gsd",
+ "must be number",
+ )
+ assert err.source == "Base Item"
+ assert err.field_path == "$.properties.gsd"
+ assert err.raw_message == "must be number"
+
+ def test_base_schema_error(self):
+ """Test error from base schema."""
+ err = FastSTACValidationError(
+ "Base Item",
+ "$.properties.gsd",
+ "must be number",
+ )
+ assert "Base Item" in str(err)
+ assert "$.properties.gsd" in str(err)
+
+
+class TestFastSTACMultiValidationError:
+ """Test FastSTACMultiValidationError container class."""
+
+ def test_multi_error_container(self):
+ """Test multi-error container with multiple errors."""
+ err1 = FastSTACValidationError(
+ "Base Item",
+ "$.properties.gsd",
+ "must be number",
+ )
+ err2 = FastSTACValidationError(
+ "Extension: https://stac-extensions.github.io/eo/v1.0.0/schema.json",
+ "$.properties.eo:cloud_cover",
+ "must be number",
+ )
+ multi_err = FastSTACMultiValidationError([err1, err2])
+
+ assert len(multi_err.errors) == 2
+ assert multi_err.errors[0] == err1
+ assert multi_err.errors[1] == err2
+ assert "Found 2 validation error(s)" in str(multi_err)
+
+ def test_multi_error_single_error(self):
+ """Test multi-error container with single error."""
+ err = FastSTACValidationError(
+ "Base Item",
+ "$.properties.gsd",
+ "must be number",
+ )
+ multi_err = FastSTACMultiValidationError([err])
+
+ assert len(multi_err.errors) == 1
+ assert "Found 1 validation error(s)" in str(multi_err)
+
+
+class TestOneOfBranchUnmasking:
+ """Verifies that fastjsonschema oneOf swallowed errors are unmasked to exact JSON Pointers.
+
+ Tests the compile-time branch unrolling mechanism that prevents oneOf/anyOf
+ from collapsing nested field errors to generic root ($) errors.
+ """
+
+ def test_oneof_branch_unmasking_returns_exact_json_pointer(self, tmp_path):
+ """Ensures nested oneOf failures unmask to $.properties.eo:cloud_cover.
+
+ This test verifies that when fastjsonschema's oneOf handler would normally
+ swallow a nested field error and report only a root ($) error, our branch
+ unrolling mechanism unmasking it to the exact field path.
+ """
+ item_path = tmp_path / "oneof_invalid_item.json"
+ item_data = {
+ "stac_version": "1.0.0",
+ "type": "Feature",
+ "id": "oneof-item",
+ "geometry": None,
+ "properties": {
+ "datetime": "2026-01-01T00:00:00Z",
+ "eo:cloud_cover": "not_a_number", # Invalid type inside oneOf
+ },
+ "links": [{"rel": "self", "href": "http://example.com"}],
+ "assets": {},
+ "stac_extensions": [
+ "https://stac-extensions.github.io/eo/v1.0.0/schema.json"
+ ],
+ }
+ item_path.write_text(json.dumps(item_data))
+
+ fv = FastValidator(str(item_path), quiet=True)
+ fv.run()
+
+ assert fv.valid is False
+ errors = fv.message[0]["errors"]
+ assert len(errors) >= 1
+
+ # Assert exact JSON pointer path rather than generic $
+ error_messages = [str(e) for e in errors]
+ error_text = " ".join(error_messages)
+
+ # Should contain the exact field path, not just root $
+ assert "eo:cloud_cover" in error_text or "$.properties" in error_text
+ assert "must be" in error_text.lower()
+
+ def test_oneof_does_not_collapse_to_root_error(self, tmp_path):
+ """Verifies oneOf errors are not collapsed to generic root ($) errors."""
+ item_path = tmp_path / "oneof_no_collapse.json"
+ item_data = {
+ "stac_version": "1.0.0",
+ "type": "Feature",
+ "id": "oneof-item",
+ "geometry": None,
+ "properties": {
+ "datetime": "2026-01-01T00:00:00Z",
+ "eo:cloud_cover": 150, # Invalid: exceeds maximum of 100
+ },
+ "links": [{"rel": "self", "href": "http://example.com"}],
+ "assets": {},
+ "stac_extensions": [
+ "https://stac-extensions.github.io/eo/v1.0.0/schema.json"
+ ],
+ }
+ item_path.write_text(json.dumps(item_data))
+
+ fv = FastValidator(str(item_path), quiet=True)
+ fv.run()
+
+ assert fv.valid is False
+ errors = fv.message[0]["errors"]
+ error_text = " ".join([str(e) for e in errors])
+
+ # Should not be just a generic root error
+ assert error_text != "$"
+ # Should mention the field or constraint
+ assert "cloud_cover" in error_text.lower() or "maximum" in error_text.lower()
+
+
+class TestErrorReportingIntegration:
+ """Integration tests for error reporting in FastValidator validation.
+
+ Tests the complete error reporting pipeline including multi-error accumulation,
+ extension attribution, and JSON Pointer path normalization.
+ """
+
+ def test_single_field_error_reported_with_extension(self, tmp_path):
+ """Test that single field errors are reported with extension context."""
+ item_path = tmp_path / "invalid_item.json"
+ item_data = {
+ "stac_version": "1.0.0",
+ "type": "Feature",
+ "id": "test-item",
+ "geometry": None,
+ "properties": {
+ "datetime": "2023-01-01T00:00:00Z",
+ "eo:cloud_cover": "not_a_number", # Invalid: should be number
+ },
+ "links": [{"rel": "self", "href": "http://example.com"}],
+ "assets": {},
+ "stac_extensions": [
+ "https://stac-extensions.github.io/eo/v1.0.0/schema.json"
+ ],
+ }
+ item_path.write_text(json.dumps(item_data))
+
+ fv = FastValidator(str(item_path), quiet=True)
+ fv.run()
+
+ assert fv.valid is False
+ assert len(fv.message[0]["errors"]) > 0
+ # Check that error mentions the extension
+ error_str = str(fv.message[0]["errors"])
+ assert "eo" in error_str.lower() or "cloud_cover" in error_str.lower()
+
+ def test_multiple_field_errors_accumulated(self, tmp_path):
+ """Test that multiple field errors are accumulated and reported."""
+ item_path = tmp_path / "multi_error_item.json"
+ item_data = {
+ "stac_version": "1.0.0",
+ "type": "Feature",
+ "id": "test-item",
+ "geometry": None,
+ "properties": {
+ "datetime": "2023-01-01T00:00:00Z",
+ "gsd": "not_a_number", # Invalid: should be number
+ "eo:cloud_cover": "also_not_a_number", # Invalid: should be number
+ },
+ "links": [{"rel": "self", "href": "http://example.com"}],
+ "assets": {},
+ "stac_extensions": [
+ "https://stac-extensions.github.io/eo/v1.0.0/schema.json"
+ ],
+ }
+ item_path.write_text(json.dumps(item_data))
+
+ fv = FastValidator(str(item_path), quiet=True)
+ fv.run()
+
+ assert fv.valid is False
+ # Should have at least 2 errors (gsd and eo:cloud_cover)
+ assert len(fv.message[0]["errors"]) >= 2
+
+ def test_error_message_contains_json_pointer(self, tmp_path):
+ """Test that error messages contain RFC 6901 JSON Pointers."""
+ item_path = tmp_path / "pointer_test_item.json"
+ item_data = {
+ "stac_version": "1.0.0",
+ "type": "Feature",
+ "id": "test-item",
+ "geometry": None,
+ "properties": {
+ "datetime": "2023-01-01T00:00:00Z",
+ "gsd": "invalid", # Should be number
+ },
+ "links": [{"rel": "self", "href": "http://example.com"}],
+ "assets": {},
+ }
+ item_path.write_text(json.dumps(item_data))
+
+ fv = FastValidator(str(item_path), quiet=True)
+ fv.run()
+
+ assert fv.valid is False
+ error_str = str(fv.message[0]["errors"])
+ # Should contain $ (JSON Pointer prefix) or the field name
+ assert "$" in error_str or "gsd" in error_str
+
+ def test_base_item_error_attribution(self, tmp_path):
+ """Test that base item errors are attributed to 'Base Item'."""
+ item_path = tmp_path / "base_error_item.json"
+ item_data = {
+ "stac_version": "1.0.0",
+ "type": "Feature",
+ "id": "test-item",
+ "geometry": None,
+ "properties": {
+ "datetime": "2023-01-01T00:00:00Z",
+ "gsd": "not_a_number", # Base schema field
+ },
+ "links": [{"rel": "self", "href": "http://example.com"}],
+ "assets": {},
+ }
+ item_path.write_text(json.dumps(item_data))
+
+ fv = FastValidator(str(item_path), quiet=True)
+ fv.run()
+
+ assert fv.valid is False
+ error_str = str(fv.message[0]["errors"])
+ # Should mention Base Item
+ assert "Base" in error_str or "gsd" in error_str
+
+ def test_extension_error_attribution(self, tmp_path):
+ """Test that extension errors are attributed to the extension URI."""
+ item_path = tmp_path / "ext_error_item.json"
+ item_data = {
+ "stac_version": "1.0.0",
+ "type": "Feature",
+ "id": "test-item",
+ "geometry": None,
+ "properties": {
+ "datetime": "2023-01-01T00:00:00Z",
+ "eo:cloud_cover": "not_a_number", # Extension field
+ },
+ "links": [{"rel": "self", "href": "http://example.com"}],
+ "assets": {},
+ "stac_extensions": [
+ "https://stac-extensions.github.io/eo/v1.0.0/schema.json"
+ ],
+ }
+ item_path.write_text(json.dumps(item_data))
+
+ fv = FastValidator(str(item_path), quiet=True)
+ fv.run()
+
+ assert fv.valid is False
+ error_str = str(fv.message[0]["errors"])
+ # Should mention the extension
+ assert "eo" in error_str.lower() or "Extension" in error_str
+
+ def test_valid_item_no_errors(self, tmp_path):
+ """Test that valid items produce no errors."""
+ item_path = tmp_path / "valid_item.json"
+ item_data = {
+ "stac_version": "1.0.0",
+ "type": "Feature",
+ "id": "test-item",
+ "geometry": None,
+ "properties": {"datetime": "2023-01-01T00:00:00Z"},
+ "links": [{"rel": "self", "href": "http://example.com"}],
+ "assets": {},
+ }
+ item_path.write_text(json.dumps(item_data))
+
+ fv = FastValidator(str(item_path), quiet=True)
+ fv.run()
+
+ assert fv.valid is True
+ assert len(fv.message[0]["errors"]) == 0
+
+ def test_error_registry_aggregates_same_errors(self, tmp_path):
+ """Test that error registry aggregates items with same error."""
+ fc_path = tmp_path / "error_agg_fc.json"
+ fc_data = {
+ "type": "FeatureCollection",
+ "features": [
+ {
+ "stac_version": "1.0.0",
+ "type": "Feature",
+ "id": "item-1",
+ "geometry": None,
+ "properties": {
+ "datetime": "2023-01-01T00:00:00Z",
+ "eo:cloud_cover": "invalid",
+ },
+ "links": [{"rel": "self", "href": "http://example.com"}],
+ "assets": {},
+ "stac_extensions": [
+ "https://stac-extensions.github.io/eo/v1.0.0/schema.json"
+ ],
+ },
+ {
+ "stac_version": "1.0.0",
+ "type": "Feature",
+ "id": "item-2",
+ "geometry": None,
+ "properties": {
+ "datetime": "2023-01-01T00:00:00Z",
+ "eo:cloud_cover": "also_invalid",
+ },
+ "links": [{"rel": "self", "href": "http://example.com"}],
+ "assets": {},
+ "stac_extensions": [
+ "https://stac-extensions.github.io/eo/v1.0.0/schema.json"
+ ],
+ },
+ ],
+ }
+ fc_path.write_text(json.dumps(fc_data))
+
+ fv = FastValidator(str(fc_path), quiet=True)
+ fv.run()
+
+ assert fv.valid is False
+ # Both items should have the same error aggregated
+ assert fv.message[0]["invalid_objects"] == 2
+
+ def test_strict_multi_error_accumulation(self, tmp_path):
+ """Asserts exact multi-error payloads for base and extension fields.
+
+ Verifies that when an item has errors in both base schema fields (gsd)
+ and extension fields (eo:cloud_cover), both are reported with exact
+ JSON Pointers and proper source attribution.
+ """
+ item_path = tmp_path / "strict_multi_item.json"
+ item_data = {
+ "stac_version": "1.0.0",
+ "type": "Feature",
+ "id": "strict-item",
+ "geometry": None,
+ "properties": {
+ "datetime": "2026-01-01T00:00:00Z",
+ "gsd": "invalid_string", # Base schema error
+ "eo:cloud_cover": "invalid_string", # Extension error
+ },
+ "links": [{"rel": "self", "href": "http://example.com"}],
+ "assets": {},
+ "stac_extensions": [
+ "https://stac-extensions.github.io/eo/v1.0.0/schema.json"
+ ],
+ }
+ item_path.write_text(json.dumps(item_data))
+
+ fv = FastValidator(str(item_path), quiet=True)
+ fv.run()
+
+ assert fv.valid is False
+ errors = fv.message[0]["errors"]
+
+ # Should have at least 2 errors (gsd and eo:cloud_cover)
+ assert len(errors) >= 2
+
+ # Convert errors to strings for assertion
+ error_strings = [str(e) for e in errors]
+ error_text = " ".join(error_strings)
+
+ # Verify base schema error is present
+ assert "gsd" in error_text or "Base" in error_text
+
+ # Verify extension error is present
+ assert "eo:cloud_cover" in error_text or "Extension" in error_text
+
+ # Verify both mention "must be"
+ assert error_text.lower().count("must be") >= 2
+
+
+class TestRunDictErrorReporting:
+ """Test error reporting in FastValidator.run_dict() method.
+
+ Tests in-memory dictionary validation with the improved error reporting.
+ """
+
+ def test_run_dict_single_error(self):
+ """Test run_dict with single validation error."""
+ payload = {
+ "stac_version": "1.0.0",
+ "type": "Feature",
+ "id": "test-item",
+ "geometry": None,
+ "properties": {
+ "datetime": "2023-01-01T00:00:00Z",
+ "gsd": "invalid", # Should be number
+ },
+ "links": [{"rel": "self", "href": "http://example.com"}],
+ "assets": {},
+ }
+
+ fv = FastValidator("", quiet=True)
+ fv.run_dict(payload)
+
+ assert fv.valid is False
+ assert len(fv.message[0]["errors"]) > 0
+
+ def test_run_dict_multiple_errors(self):
+ """Test run_dict with multiple validation errors."""
+ payload = {
+ "stac_version": "1.0.0",
+ "type": "Feature",
+ "id": "test-item",
+ "geometry": None,
+ "properties": {
+ "datetime": "2023-01-01T00:00:00Z",
+ "gsd": "invalid", # Should be number
+ "eo:cloud_cover": "also_invalid", # Should be number
+ },
+ "links": [{"rel": "self", "href": "http://example.com"}],
+ "assets": {},
+ "stac_extensions": [
+ "https://stac-extensions.github.io/eo/v1.0.0/schema.json"
+ ],
+ }
+
+ fv = FastValidator("", quiet=True)
+ fv.run_dict(payload)
+
+ assert fv.valid is False
+ # Should have multiple errors
+ assert len(fv.message[0]["errors"]) >= 2
+
+ def test_run_dict_no_errors(self):
+ """Test run_dict with valid item."""
+ payload = {
+ "stac_version": "1.0.0",
+ "type": "Feature",
+ "id": "test-item",
+ "geometry": None,
+ "properties": {"datetime": "2023-01-01T00:00:00Z"},
+ "links": [{"rel": "self", "href": "http://example.com"}],
+ "assets": {},
+ }
+
+ fv = FastValidator("", quiet=True)
+ fv.run_dict(payload)
+
+ assert fv.valid is True
+ assert len(fv.message[0]["errors"]) == 0
=====================================
tests/test_fast_validator_enhancements.py
=====================================
@@ -0,0 +1,215 @@
+"""Tests for FastValidator enhancements: type safety, cache isolation, and scoped additionalProperties.
+
+This module verifies the three targeted fixes applied to improve-fast-error-msg:
+1. Type guard in parse_json_pointer for defensive parsing
+2. Test fixture teardown to prevent cache leakage
+3. Scoped additionalProperties stripping to preserve nested object strictness
+"""
+
+import json
+
+from stac_validator import fast_validator
+from stac_validator.fast_validator import FastValidator, parse_json_pointer
+
+
+class TestParseJsonPointerTypeSafety:
+ """Verifies defensive type handling in parse_json_pointer."""
+
+ def test_non_string_inputs_return_root_pointer(self):
+ """Non-string inputs should safely return root pointer without raising exceptions."""
+ assert parse_json_pointer(None) == "$"
+ assert parse_json_pointer(123) == "$" # type: ignore
+ assert parse_json_pointer([]) == "$" # type: ignore
+ assert parse_json_pointer({}) == "$" # type: ignore
+
+ def test_valid_string_conversions(self):
+ """Valid string inputs should convert correctly to JSON Pointers."""
+ assert parse_json_pointer("data") == "$"
+ assert parse_json_pointer("data.properties.gsd") == "$.properties.gsd"
+ assert (
+ parse_json_pointer("data['properties']['eo:cloud_cover']")
+ == "$.properties.eo:cloud_cover"
+ )
+
+ def test_empty_string_returns_root(self):
+ """Empty string should return root pointer."""
+ assert parse_json_pointer("") == "$"
+
+
+class TestScopedAdditionalProperties:
+ """Verifies that additionalProperties: false is stripped ONLY in shared top-level properties."""
+
+ def test_nested_additional_properties_not_stripped_by_optimizer(self):
+ """Verify that optimize_schema_for_compiler does NOT strip additionalProperties: false
+ from nested objects (only from top-level shared properties).
+ """
+ from stac_validator.fast_validator import optimize_schema_for_compiler
+
+ # Schema with nested additionalProperties: false
+ schema = {
+ "type": "object",
+ "properties": {
+ "assets": {
+ "type": "object",
+ "patternProperties": {
+ ".*": {
+ "type": "object",
+ "properties": {"href": {"type": "string"}},
+ "additionalProperties": False, # Should be preserved
+ }
+ },
+ }
+ },
+ }
+
+ optimized = optimize_schema_for_compiler(schema)
+
+ # Navigate to the nested additionalProperties
+ nested_additional_props = optimized["properties"]["assets"][
+ "patternProperties"
+ ][".*"].get("additionalProperties")
+
+ # Should still be False (not stripped)
+ assert nested_additional_props is False
+
+ def test_top_level_properties_additional_properties_stripped(self):
+ """Verify that optimize_schema_for_compiler DOES strip additionalProperties: false
+ from the top-level shared properties block for multi-extension compatibility.
+ """
+ from stac_validator.fast_validator import optimize_schema_for_compiler
+
+ # Schema with additionalProperties: false at top-level properties
+ schema = {
+ "type": "object",
+ "properties": {
+ "properties": {
+ "type": "object",
+ "properties": {"datetime": {"type": "string"}},
+ "additionalProperties": False, # Should be stripped
+ }
+ },
+ }
+
+ optimized = optimize_schema_for_compiler(schema)
+
+ # Navigate to the top-level properties additionalProperties
+ top_level_additional_props = optimized["properties"]["properties"].get(
+ "additionalProperties"
+ )
+
+ # Should be None/missing (stripped for multi-extension compatibility)
+ assert top_level_additional_props is None
+
+ def test_top_level_properties_allows_extension_fields(self, tmp_path, monkeypatch):
+ """Top-level properties should allow extension fields even if base schema
+ specifies additionalProperties: false (after scoped stripping).
+ """
+ # Save original caches
+ orig_schema_cache = fast_validator.SCHEMA_CACHE.copy()
+ orig_validator_cache = fast_validator.VALIDATOR_CACHE.copy()
+
+ try:
+ fast_validator.SCHEMA_CACHE.clear()
+ fast_validator.VALIDATOR_CACHE.clear()
+
+ base_item_schema = {
+ "$schema": "http://json-schema.org/draft-07/schema#",
+ "type": "object",
+ "required": [
+ "stac_version",
+ "type",
+ "id",
+ "geometry",
+ "properties",
+ "links",
+ "assets",
+ ],
+ "properties": {
+ "stac_version": {"type": "string"},
+ "type": {"const": "Feature"},
+ "id": {"type": "string"},
+ "geometry": {"type": ["object", "null"]},
+ "properties": {
+ "type": "object",
+ "properties": {
+ "datetime": {"type": ["string", "null"]},
+ },
+ "additionalProperties": False, # Will be stripped for multi-extension compatibility
+ },
+ "links": {"type": "array"},
+ "assets": {"type": "object"},
+ },
+ }
+
+ fast_validator.SCHEMA_CACHE[
+ "https://schemas.stacspec.org/v1.0.0/item-spec/json-schema/item.json"
+ ] = base_item_schema
+
+ # Payload with extension field in top-level properties
+ item_path = tmp_path / "extension_field_item.json"
+ item_data = {
+ "stac_version": "1.0.0",
+ "type": "Feature",
+ "id": "extension-item",
+ "geometry": None,
+ "properties": {
+ "datetime": "2026-01-01T00:00:00Z",
+ "eo:cloud_cover": 42.5, # Extension field should be allowed
+ },
+ "links": [{"rel": "self", "href": "http://example.com"}],
+ "assets": {},
+ }
+ item_path.write_text(json.dumps(item_data))
+
+ fv = FastValidator(str(item_path), quiet=True)
+ fv.run()
+
+ # Should pass because additionalProperties: false was stripped from top-level properties
+ assert fv.valid is True
+
+ finally:
+ # Restore original caches
+ fast_validator.SCHEMA_CACHE.clear()
+ fast_validator.SCHEMA_CACHE.update(orig_schema_cache)
+ fast_validator.VALIDATOR_CACHE.clear()
+ fast_validator.VALIDATOR_CACHE.update(orig_validator_cache)
+
+
+class TestCacheIsolationTeardown:
+ """Verifies that tests do not leak global cache state across execution runs."""
+
+ def test_fixture_teardown_restores_global_caches(self):
+ """Simulate cache isolation and restoration to prevent test leakage."""
+ original_schema_cache = dict(fast_validator.SCHEMA_CACHE)
+ original_validator_cache = dict(fast_validator.VALIDATOR_CACHE)
+
+ # Simulate isolated test modification
+ fast_validator.SCHEMA_CACHE["mock_uri"] = {"test": "schema"}
+ fast_validator.VALIDATOR_CACHE["mock_key"] = lambda x: True
+
+ # Restore
+ fast_validator.SCHEMA_CACHE.clear()
+ fast_validator.SCHEMA_CACHE.update(original_schema_cache)
+ fast_validator.VALIDATOR_CACHE.clear()
+ fast_validator.VALIDATOR_CACHE.update(original_validator_cache)
+
+ assert "mock_uri" not in fast_validator.SCHEMA_CACHE
+ assert "mock_key" not in fast_validator.VALIDATOR_CACHE
+
+ def test_cache_state_independent_across_tests(self):
+ """Verify that cache modifications in one test don't affect another."""
+ # Record initial state
+ initial_schema_keys = set(fast_validator.SCHEMA_CACHE.keys())
+
+ # Add a test entry
+ test_key = "test_isolation_key_12345"
+ fast_validator.SCHEMA_CACHE[test_key] = {"test": "data"}
+
+ # Verify it's there
+ assert test_key in fast_validator.SCHEMA_CACHE
+
+ # Clean up
+ del fast_validator.SCHEMA_CACHE[test_key]
+
+ # Verify we're back to initial state
+ assert set(fast_validator.SCHEMA_CACHE.keys()) == initial_schema_keys
=====================================
tests/test_validate_item_collection.py
=====================================
@@ -332,209 +332,209 @@ def test_validate_item_collection_remote_pages():
assert stac.message == [
{
- "version": "1.0.0",
+ "version": "1.1.0",
"path": "https://stac.geobon.org/collections/chelsa-clim/items/bio9",
"schema": [
- "https://stac-extensions.github.io/projection/v1.0.0/schema.json",
- "https://schemas.stacspec.org/v1.0.0/item-spec/json-schema/item.json",
+ "https://stac-extensions.github.io/projection/v2.0.0/schema.json",
+ "https://schemas.stacspec.org/v1.1.0/item-spec/json-schema/item.json",
],
"valid_stac": True,
"asset_type": "ITEM",
"validation_method": "default",
},
{
- "version": "1.0.0",
+ "version": "1.1.0",
"path": "https://stac.geobon.org/collections/chelsa-clim/items/bio8",
"schema": [
- "https://stac-extensions.github.io/projection/v1.0.0/schema.json",
- "https://schemas.stacspec.org/v1.0.0/item-spec/json-schema/item.json",
+ "https://stac-extensions.github.io/projection/v2.0.0/schema.json",
+ "https://schemas.stacspec.org/v1.1.0/item-spec/json-schema/item.json",
],
"valid_stac": True,
"asset_type": "ITEM",
"validation_method": "default",
},
{
- "version": "1.0.0",
+ "version": "1.1.0",
"path": "https://stac.geobon.org/collections/chelsa-clim/items/bio7",
"schema": [
- "https://stac-extensions.github.io/projection/v1.0.0/schema.json",
- "https://schemas.stacspec.org/v1.0.0/item-spec/json-schema/item.json",
+ "https://stac-extensions.github.io/projection/v2.0.0/schema.json",
+ "https://schemas.stacspec.org/v1.1.0/item-spec/json-schema/item.json",
],
"valid_stac": True,
"asset_type": "ITEM",
"validation_method": "default",
},
{
- "version": "1.0.0",
+ "version": "1.1.0",
"path": "https://stac.geobon.org/collections/chelsa-clim/items/bio6",
"schema": [
- "https://stac-extensions.github.io/projection/v1.0.0/schema.json",
- "https://schemas.stacspec.org/v1.0.0/item-spec/json-schema/item.json",
+ "https://stac-extensions.github.io/projection/v2.0.0/schema.json",
+ "https://schemas.stacspec.org/v1.1.0/item-spec/json-schema/item.json",
],
"valid_stac": True,
"asset_type": "ITEM",
"validation_method": "default",
},
{
- "version": "1.0.0",
+ "version": "1.1.0",
"path": "https://stac.geobon.org/collections/chelsa-clim/items/bio5",
"schema": [
- "https://stac-extensions.github.io/projection/v1.0.0/schema.json",
- "https://schemas.stacspec.org/v1.0.0/item-spec/json-schema/item.json",
+ "https://stac-extensions.github.io/projection/v2.0.0/schema.json",
+ "https://schemas.stacspec.org/v1.1.0/item-spec/json-schema/item.json",
],
"valid_stac": True,
"asset_type": "ITEM",
"validation_method": "default",
},
{
- "version": "1.0.0",
+ "version": "1.1.0",
"path": "https://stac.geobon.org/collections/chelsa-clim/items/bio4",
"schema": [
- "https://stac-extensions.github.io/projection/v1.0.0/schema.json",
- "https://schemas.stacspec.org/v1.0.0/item-spec/json-schema/item.json",
+ "https://stac-extensions.github.io/projection/v2.0.0/schema.json",
+ "https://schemas.stacspec.org/v1.1.0/item-spec/json-schema/item.json",
],
"valid_stac": True,
"asset_type": "ITEM",
"validation_method": "default",
},
{
- "version": "1.0.0",
+ "version": "1.1.0",
"path": "https://stac.geobon.org/collections/chelsa-clim/items/bio3",
"schema": [
- "https://stac-extensions.github.io/projection/v1.0.0/schema.json",
- "https://schemas.stacspec.org/v1.0.0/item-spec/json-schema/item.json",
+ "https://stac-extensions.github.io/projection/v2.0.0/schema.json",
+ "https://schemas.stacspec.org/v1.1.0/item-spec/json-schema/item.json",
],
"valid_stac": True,
"asset_type": "ITEM",
"validation_method": "default",
},
{
- "version": "1.0.0",
+ "version": "1.1.0",
"path": "https://stac.geobon.org/collections/chelsa-clim/items/bio2",
"schema": [
- "https://stac-extensions.github.io/projection/v1.0.0/schema.json",
- "https://schemas.stacspec.org/v1.0.0/item-spec/json-schema/item.json",
+ "https://stac-extensions.github.io/projection/v2.0.0/schema.json",
+ "https://schemas.stacspec.org/v1.1.0/item-spec/json-schema/item.json",
],
"valid_stac": True,
"asset_type": "ITEM",
"validation_method": "default",
},
{
- "version": "1.0.0",
+ "version": "1.1.0",
"path": "https://stac.geobon.org/collections/chelsa-clim/items/bio19",
"schema": [
- "https://stac-extensions.github.io/projection/v1.0.0/schema.json",
- "https://schemas.stacspec.org/v1.0.0/item-spec/json-schema/item.json",
+ "https://stac-extensions.github.io/projection/v2.0.0/schema.json",
+ "https://schemas.stacspec.org/v1.1.0/item-spec/json-schema/item.json",
],
"valid_stac": True,
"asset_type": "ITEM",
"validation_method": "default",
},
{
- "version": "1.0.0",
+ "version": "1.1.0",
"path": "https://stac.geobon.org/collections/chelsa-clim/items/bio18",
"schema": [
- "https://stac-extensions.github.io/projection/v1.0.0/schema.json",
- "https://schemas.stacspec.org/v1.0.0/item-spec/json-schema/item.json",
+ "https://stac-extensions.github.io/projection/v2.0.0/schema.json",
+ "https://schemas.stacspec.org/v1.1.0/item-spec/json-schema/item.json",
],
"valid_stac": True,
"asset_type": "ITEM",
"validation_method": "default",
},
{
- "version": "1.0.0",
+ "version": "1.1.0",
"path": "https://stac.geobon.org/collections/chelsa-clim/items/bio17",
"schema": [
- "https://stac-extensions.github.io/projection/v1.0.0/schema.json",
- "https://schemas.stacspec.org/v1.0.0/item-spec/json-schema/item.json",
+ "https://stac-extensions.github.io/projection/v2.0.0/schema.json",
+ "https://schemas.stacspec.org/v1.1.0/item-spec/json-schema/item.json",
],
"valid_stac": True,
"asset_type": "ITEM",
"validation_method": "default",
},
{
- "version": "1.0.0",
+ "version": "1.1.0",
"path": "https://stac.geobon.org/collections/chelsa-clim/items/bio16",
"schema": [
- "https://stac-extensions.github.io/projection/v1.0.0/schema.json",
- "https://schemas.stacspec.org/v1.0.0/item-spec/json-schema/item.json",
+ "https://stac-extensions.github.io/projection/v2.0.0/schema.json",
+ "https://schemas.stacspec.org/v1.1.0/item-spec/json-schema/item.json",
],
"valid_stac": True,
"asset_type": "ITEM",
"validation_method": "default",
},
{
- "version": "1.0.0",
+ "version": "1.1.0",
"path": "https://stac.geobon.org/collections/chelsa-clim/items/bio15",
"schema": [
- "https://stac-extensions.github.io/projection/v1.0.0/schema.json",
- "https://schemas.stacspec.org/v1.0.0/item-spec/json-schema/item.json",
+ "https://stac-extensions.github.io/projection/v2.0.0/schema.json",
+ "https://schemas.stacspec.org/v1.1.0/item-spec/json-schema/item.json",
],
"valid_stac": True,
"asset_type": "ITEM",
"validation_method": "default",
},
{
- "version": "1.0.0",
+ "version": "1.1.0",
"path": "https://stac.geobon.org/collections/chelsa-clim/items/bio14",
"schema": [
- "https://stac-extensions.github.io/projection/v1.0.0/schema.json",
- "https://schemas.stacspec.org/v1.0.0/item-spec/json-schema/item.json",
+ "https://stac-extensions.github.io/projection/v2.0.0/schema.json",
+ "https://schemas.stacspec.org/v1.1.0/item-spec/json-schema/item.json",
],
"valid_stac": True,
"asset_type": "ITEM",
"validation_method": "default",
},
{
- "version": "1.0.0",
+ "version": "1.1.0",
"path": "https://stac.geobon.org/collections/chelsa-clim/items/bio13",
"schema": [
- "https://stac-extensions.github.io/projection/v1.0.0/schema.json",
- "https://schemas.stacspec.org/v1.0.0/item-spec/json-schema/item.json",
+ "https://stac-extensions.github.io/projection/v2.0.0/schema.json",
+ "https://schemas.stacspec.org/v1.1.0/item-spec/json-schema/item.json",
],
"valid_stac": True,
"asset_type": "ITEM",
"validation_method": "default",
},
{
- "version": "1.0.0",
+ "version": "1.1.0",
"path": "https://stac.geobon.org/collections/chelsa-clim/items/bio12",
"schema": [
- "https://stac-extensions.github.io/projection/v1.0.0/schema.json",
- "https://schemas.stacspec.org/v1.0.0/item-spec/json-schema/item.json",
+ "https://stac-extensions.github.io/projection/v2.0.0/schema.json",
+ "https://schemas.stacspec.org/v1.1.0/item-spec/json-schema/item.json",
],
"valid_stac": True,
"asset_type": "ITEM",
"validation_method": "default",
},
{
- "version": "1.0.0",
+ "version": "1.1.0",
"path": "https://stac.geobon.org/collections/chelsa-clim/items/bio11",
"schema": [
- "https://stac-extensions.github.io/projection/v1.0.0/schema.json",
- "https://schemas.stacspec.org/v1.0.0/item-spec/json-schema/item.json",
+ "https://stac-extensions.github.io/projection/v2.0.0/schema.json",
+ "https://schemas.stacspec.org/v1.1.0/item-spec/json-schema/item.json",
],
"valid_stac": True,
"asset_type": "ITEM",
"validation_method": "default",
},
{
- "version": "1.0.0",
+ "version": "1.1.0",
"path": "https://stac.geobon.org/collections/chelsa-clim/items/bio10",
"schema": [
- "https://stac-extensions.github.io/projection/v1.0.0/schema.json",
- "https://schemas.stacspec.org/v1.0.0/item-spec/json-schema/item.json",
+ "https://stac-extensions.github.io/projection/v2.0.0/schema.json",
+ "https://schemas.stacspec.org/v1.1.0/item-spec/json-schema/item.json",
],
"valid_stac": True,
"asset_type": "ITEM",
"validation_method": "default",
},
{
- "version": "1.0.0",
+ "version": "1.1.0",
"path": "https://stac.geobon.org/collections/chelsa-clim/items/bio1",
"schema": [
- "https://stac-extensions.github.io/projection/v1.0.0/schema.json",
- "https://schemas.stacspec.org/v1.0.0/item-spec/json-schema/item.json",
+ "https://stac-extensions.github.io/projection/v2.0.0/schema.json",
+ "https://schemas.stacspec.org/v1.1.0/item-spec/json-schema/item.json",
],
"valid_stac": True,
"asset_type": "ITEM",
@@ -586,16 +586,3 @@ def test_validate_item_collection_remote_pages_1_v110():
schema in msg["schema"] for schema in expected_schemas
), f"Missing expected schemas in {msg['schema']}"
assert len(stac.message) == 10
-
-
-def test_validate_item_collection_remote_pages_3_v110():
- # Fix: Point to a massive, permanent historical collection, NOT a temporary Near Real-Time (nrt) one.
- stac_file = (
- "https://stac.dataspace.copernicus.eu/v1/collections/sentinel-2-l2a/items"
- )
- stac = stac_validator.StacValidate(stac_file, item_collection=True, pages=3)
- stac.validate_item_collection()
-
- # We expect 3 pages of results.
- # Using > 20 proves pagination successfully fetched multiple pages without hardcoding a brittle exact number.
- assert len(stac.message) > 20
View it on GitLab: https://salsa.debian.org/debian-gis-team/stac-validator/-/compare/1bcd735d41691cae3a207053718fa5c58ee658c9...e1febfc94cca15a5636a7f09e3ee130269f5a9d0
--
View it on GitLab: https://salsa.debian.org/debian-gis-team/stac-validator/-/compare/1bcd735d41691cae3a207053718fa5c58ee658c9...e1febfc94cca15a5636a7f09e3ee130269f5a9d0
You're receiving this email because of your account on salsa.debian.org. Manage all notifications: https://salsa.debian.org/-/profile/notifications | Help: https://salsa.debian.org/help
-------------- next part --------------
An HTML attachment was scrubbed...
URL: <http://alioth-lists.debian.net/pipermail/pkg-grass-devel/attachments/20260905/2cc40b6d/attachment-0001.htm>
More information about the Pkg-grass-devel
mailing list