add test, update tests

Author: jonhealy1

stac_validator/stac_validator.py

@@ -8,9 +8,11 @@
 from .validate import StacValidate
 
 
-def _print_summary(title: str, valid_count: int, total_count: int, obj_type: str = "STAC objects") -> None:
+def _print_summary(
+    title: str, valid_count: int, total_count: int, obj_type: str = "STAC objects"
+) -> None:
     """Helper function to print a consistent summary line.
-    
+
     Args:
         title (str): Title of the summary section
         valid_count (int): Number of valid items
@@ -21,17 +23,19 @@ def _print_summary(title: str, valid_count: int, total_count: int, obj_type: str
     click.secho(f"{title}:", bold=True)
     if total_count > 0:
         percentage = (valid_count / total_count) * 100
-        click.secho(f"  {obj_type.capitalize()} passed: {valid_count}/{total_count} ({percentage:.1f}%)")
+        click.secho(
+            f"  {obj_type.capitalize()} passed: {valid_count}/{total_count} ({percentage:.1f}%)"
+        )
     else:
         click.secho(f"  No {obj_type} found to validate")
 
 
 def format_duration(seconds: float) -> str:
     """Format duration in seconds to a human-readable string.
-    
+
     Args:
         seconds (float): Duration in seconds
-        
+
     Returns:
         str: Formatted duration string (e.g., '1m 23.45s' or '456.78ms')
     """
@@ -91,41 +95,45 @@ def collections_summary(message: List[Dict[str, Any]]) -> None:
 
 def recursive_validation_summary(message: List[Dict[str, Any]]) -> None:
     """Prints a summary of the recursive validation results.
-    
+
     Args:
         message (List[Dict[str, Any]]): The validation results from recursive validation.
-        
+
     Returns:
         None
     """
     # Count valid and total objects by type
     type_counts = {}
     total_valid = 0
-    
+
     for item in message:
         if not isinstance(item, dict):
             continue
-            
+
         obj_type = item.get("asset_type", "unknown").lower()
         is_valid = item.get("valid_stac", False) is True
-        
+
         if obj_type not in type_counts:
             type_counts[obj_type] = {"valid": 0, "total": 0}
-            
+
         type_counts[obj_type]["total"] += 1
         if is_valid:
             type_counts[obj_type]["valid"] += 1
             total_valid += 1
-    
+
     # Print overall summary
     _print_summary("-- Recursive Validation Summary", total_valid, len(message))
-    
+
     # Print breakdown by type if there are multiple types
     if len(type_counts) > 1:
         click.secho("\n  Breakdown by type:")
         for obj_type, counts in sorted(type_counts.items()):
-            percentage = (counts["valid"] / counts["total"]) * 100 if counts["total"] > 0 else 0
-            click.secho(f"    {obj_type.capitalize()}: {counts['valid']}/{counts['total']} ({percentage:.1f}%)")
+            percentage = (
+                (counts["valid"] / counts["total"]) * 100 if counts["total"] > 0 else 0
+            )
+            click.secho(
+                f"    {obj_type.capitalize()}: {counts['valid']}/{counts['total']} ({percentage:.1f}%)"
+            )
 
 
 @click.command()
@@ -281,12 +289,12 @@ def main(
     """
     start_time = time.time()
     valid = True
-    
+
     if schema_map == ():
         schema_map_dict: Optional[Dict[str, str]] = None
     else:
         schema_map_dict = dict(schema_map)
-        
+
     stac = StacValidate(
         stac_file=stac_file,
         collections=collections,
@@ -308,34 +316,36 @@ def main(
         pydantic=pydantic,
         verbose=verbose,
     )
-    
+
     try:
         if not item_collection and not collections:
             valid = stac.run()
         elif collections:
             stac.validate_collections()
         else:
             stac.validate_item_collection()
-            
+
         message = stac.message
         if "version" in message[0]:
             print_update_message(message[0]["version"])
 
         if no_output is False:
             click.echo(json.dumps(message, indent=4))
-            
+
         # Print appropriate summary based on validation mode
         if item_collection:
             item_collection_summary(message)
         elif collections:
             collections_summary(message)
         elif recursive:
             recursive_validation_summary(message)
-            
+
     finally:
         # Always print the duration, even if validation fails
         duration = time.time() - start_time
-        click.secho(f"\nValidation completed in {format_duration(duration)}", fg='green')
+        click.secho(
+            f"\nValidation completed in {format_duration(duration)}", fg="green"
+        )
         click.secho()
     sys.exit(0 if valid else 1)
 

stac_validator/validate.py

@@ -242,7 +242,11 @@ def _create_verbose_err_msg(self, error_input: Any) -> Union[Dict[str, Any], str
         return str(error_input)  # Fallback to string representation
 
     def create_err_msg(
-        self, err_type: str, err_msg: str, error_obj: Optional[Exception] = None
+        self,
+        err_type: str,
+        err_msg: str,
+        error_obj: Optional[Exception] = None,
+        schema_uri: str = "",
     ) -> Dict[str, Union[str, bool, List[str], Dict[str, Any]]]:
         """
         Create a standardized error message dictionary and mark validation as failed.
@@ -251,26 +255,59 @@ def create_err_msg(
             err_type (str): The type of error.
             err_msg (str): The error message.
             error_obj (Optional[Exception]): The raw exception object for verbose details.
+            schema_uri (str, optional): The URI of the schema that failed validation.
 
         Returns:
             dict: Dictionary containing error information.
         """
         self.valid = False
 
         # Ensure all values are of the correct type
-        version_str: str = str(self.version) if self.version is not None else ""
-        path_str: str = str(self.stac_file) if self.stac_file is not None else ""
+        if not isinstance(err_type, str):
+            err_type = str(err_type)
+        if not isinstance(err_msg, str):
+            err_msg = str(err_msg)
+
+        # Initialize the message with common fields
+        message: Dict[str, Any] = {
+            "version": str(self.version) if hasattr(self, "version") else "",
+            "path": str(self.stac_file) if hasattr(self, "stac_file") else "",
+            "schema": (
+                [self._original_schema_paths.get(self.schema, self.schema)]
+                if hasattr(self, "schema")
+                else [""]
+            ),
+            "valid_stac": False,
+            "error_type": err_type,
+            "error_message": err_msg,
+            "failed_schema": schema_uri if schema_uri else "",
+            "recommendation": "For more accurate error information, rerun with --verbose.",
+        }
+
+        # Add asset_type and validation_method if available
+        if hasattr(self, "stac_content"):
+            try:
+                stac_type = get_stac_type(self.stac_content)
+                if stac_type:
+                    message["asset_type"] = stac_type.upper()
+                    message["validation_method"] = (
+                        "recursive"
+                        if hasattr(self, "recursive") and self.recursive
+                        else "default"
+                    )
+            except Exception:  # noqa: BLE001
+                pass
 
         # Ensure schema is properly typed
         schema_value: str = ""
         if self.schema is not None:
             schema_value = str(self.schema)
         schema_field: List[str] = [schema_value] if schema_value else []
 
-        # Initialize the message with common fields
-        message: Dict[str, Union[str, bool, List[str], Dict[str, Any]]] = {
-            "version": version_str,
-            "path": path_str,
+        # Initialize the error message with common fields
+        error_message: Dict[str, Union[str, bool, List[str], Dict[str, Any]]] = {
+            "version": str(self.version) if self.version is not None else "",
+            "path": str(self.stac_file) if self.stac_file is not None else "",
             "schema": schema_field,  # All schemas that were checked
             "valid_stac": False,
             "error_type": err_type,
@@ -281,31 +318,31 @@ def create_err_msg(
         # Try to extract the failed schema from the error message if it's a validation error
         if error_obj and hasattr(error_obj, "schema"):
             if isinstance(error_obj.schema, dict) and "$id" in error_obj.schema:
-                message["failed_schema"] = error_obj.schema["$id"]
+                error_message["failed_schema"] = error_obj.schema["$id"]
             elif hasattr(error_obj, "schema_url"):
-                message["failed_schema"] = error_obj.schema_url
+                error_message["failed_schema"] = error_obj.schema_url
             # If we can't find a schema ID, try to get it from the schema map
             elif schema_field and len(schema_field) == 1:
-                message["failed_schema"] = schema_field[0]
+                error_message["failed_schema"] = schema_field[0]
 
         if self.verbose and error_obj is not None:
             verbose_err = self._create_verbose_err_msg(error_obj)
             if isinstance(verbose_err, dict):
-                message["error_verbose"] = verbose_err
+                error_message["error_verbose"] = verbose_err
             else:
-                message["error_verbose"] = {"detail": str(verbose_err)}
+                error_message["error_verbose"] = {"detail": str(verbose_err)}
             # Add recommendation to check the schema if the error is not clear
-            if "failed_schema" in message and message["failed_schema"]:
-                message["recommendation"] = (
+            if error_message.get("failed_schema"):
+                error_message["recommendation"] = (
                     "If the error is unclear, please check the schema documentation at: "
-                    f"{message['failed_schema']}"
+                    f"{error_message['failed_schema']}"
                 )
         else:
-            message["recommendation"] = (
+            error_message["recommendation"] = (
                 "For more accurate error information, rerun with --verbose."
             )
 
-        return message
+        return error_message
 
     def create_links_message(self) -> Dict:
         """
@@ -604,7 +641,9 @@ def recursive_validator(self, stac_type: str) -> bool:
                         err_type="JSONSchemaValidationError",
                         err_msg=err_msg,
                         error_obj=e,
-                        schema_uri=e.schema.get("$id", "") if hasattr(e, "schema") else "",
+                        schema_uri=(
+                            e.schema.get("$id", "") if hasattr(e, "schema") else ""
+                        ),
                     )
                 )
                 self.message.append(message)
@@ -686,7 +725,9 @@ def recursive_validator(self, stac_type: str) -> bool:
                 if link["rel"] == "item":
                     self.schema = set_schema_addr(self.version, stac_type.lower())
                     message = self.create_message(stac_type, "recursive")
-                    message["validator_engine"] = "pydantic" if self.pydantic else "jsonschema"
+                    message["validator_engine"] = (
+                        "pydantic" if self.pydantic else "jsonschema"
+                    )
                     try:
                         if self.pydantic:
                             # Set pydantic model info in schema field for child items

tests/test_pydantic.py

@@ -59,3 +59,34 @@ def test_pydantic_invalid_item():
     assert stac.message[0]["validation_method"] == "pydantic"
     assert "error_type" in stac.message[0]
     assert "error_message" in stac.message[0]
+
+
+def test_pydantic_recursive():
+    """Test pydantic validation in recursive mode."""
+    stac_file = "tests/test_data/local_cat/example-catalog/catalog.json"
+    stac = stac_validator.StacValidate(
+        stac_file, pydantic=True, recursive=True, max_depth=2  # Limit depth for testing
+    )
+    stac.run()
+
+    # Check that we have validation messages
+    assert len(stac.message) > 0
+
+    # Check each validation message
+    for msg in stac.message:
+        # Check that validator_engine is set to pydantic
+        assert msg["validator_engine"] == "pydantic"
+
+        # Check that validation_method is recursive
+        assert msg["validation_method"] == "recursive"
+
+        # Check that valid_stac is a boolean
+        assert isinstance(msg["valid_stac"], bool)
+
+        # Check that schema is set to pydantic model based on asset type
+        if msg["asset_type"] == "ITEM":
+            assert msg["schema"] == ["stac-pydantic Item model"]
+        elif msg["asset_type"] == "COLLECTION":
+            assert msg["schema"] == ["stac-pydantic Collection model"]
+        elif msg["asset_type"] == "CATALOG":
+            assert msg["schema"] == ["stac-pydantic Catalog model"]