Include href when schema resolution fails (#1263)

* fix(spell): test name

* feat: report href when failing to fetch schema

CHANGELOG.md

@@ -11,6 +11,7 @@
 ### Changed
 
 - `validate_all` now accepts a `STACObject` (in addition to accepting a dict, which is now deprecated), but prohibits supplying a value for `href`, which must be supplied _only_ when supplying an object as a dict.  Once `validate_all` removes support for an object as a dict, the `href` parameter will also be removed. ([#1246](https://github.com/stac-utils/pystac/pull/1246))
+- Report `href` when schema url resolution fails ([#1263](https://github.com/stac-utils/pystac/pull/1263))
 
 ### Fixed
 

pyproject.toml

@@ -53,6 +53,7 @@ test = [
     "pytest-mock~=3.10",
     "pytest-recording~=0.13.0",
     "pytest~=7.3",
+    "requests-mock~=1.11",
     "ruff==0.0.292",
     "types-html5lib~=1.1",
     "types-orjson~=3.6",

pystac/validation/__init__.py

@@ -9,6 +9,7 @@
 from pystac.stac_object import STACObjectType
 from pystac.utils import make_absolute_href
 from pystac.validation.schema_uri_map import OldExtensionSchemaUriMap
+from pystac.validation.stac_validator import GetSchemaError
 
 if TYPE_CHECKING:
     from pystac.stac_object import STACObject
@@ -245,3 +246,6 @@ def set_validator(validator: STACValidator) -> None:
             validation.
     """
     RegisteredValidator.set_validator(validator)
+
+
+__all__ = ["GetSchemaError"]

pystac/validation/stac_validator.py

@@ -25,6 +25,13 @@
 logger = logging.getLogger(__name__)
 
 
+class GetSchemaError(Exception):
+    """Raised when unable to fetch a schema."""
+
+    def __init__(self, href: str, error: Exception) -> None:
+        super().__init__(f"Error when fetching schema {href}: {error}")
+
+
 class STACValidator(ABC):
     """STACValidator defines methods for validating STAC
     JSON. Implementations define methods for validating core objects and extension.
@@ -155,7 +162,10 @@ def __init__(self, schema_uri_map: Optional[SchemaUriMap] = None) -> None:
 
     def _get_schema(self, schema_uri: str) -> dict[str, Any]:
         if schema_uri not in self.schema_cache:
-            s = json.loads(pystac.StacIO.default().read_text(schema_uri))
+            try:
+                s = json.loads(pystac.StacIO.default().read_text(schema_uri))
+            except Exception as error:
+                raise GetSchemaError(schema_uri, error) from error
             self.schema_cache[schema_uri] = s
             id_field = "$id" if "$id" in s else "id"
             if not s[id_field].startswith("http"):

tests/validation/test_validate.py

@@ -7,12 +7,14 @@
 
 import jsonschema
 import pytest
+from requests_mock import Mocker
 
 import pystac
 import pystac.validation
 from pystac.cache import CollectionCache
 from pystac.serialization.common_properties import merge_common_properties
 from pystac.utils import get_opt
+from pystac.validation import GetSchemaError
 from tests.utils import TestCases
 from tests.utils.test_cases import ExampleInfo
 
@@ -188,10 +190,21 @@ def test_catalog_latest_version_uses_local(catalog: pystac.Catalog) -> None:
 
 
 @pytest.mark.block_network
-def test_collection_latest_versio_uses_localn(collection: pystac.Collection) -> None:
+def test_collection_latest_version_uses_local(collection: pystac.Collection) -> None:
     assert collection.validate()
 
 
 @pytest.mark.block_network
 def test_item_latest_version_uses_local(item: pystac.Item) -> None:
     assert item.validate()
+
+
+def test_404_schema_url(requests_mock: Mocker, item: pystac.Item) -> None:
+    requests_mock.get(
+        "http://pystac-extensions.test/a-fake-schema.json", status_code=404
+    )
+    item.stac_extensions = ["http://pystac-extensions.test/a-fake-schema.json"]
+    with pytest.raises(
+        GetSchemaError, match="http://pystac-extensions.test/a-fake.schema.json"
+    ):
+        item.validate()