diff --git a/CHANGELOG.md b/CHANGELOG.md index b3db25a..f68fcaa 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,15 @@ The format is (loosely) based on [Keep a Changelog](http://keepachangelog.com/) ## Unreleased +## [v1.11.0] - 2025-06-22 + +### Added + +- Results summary for options that produce numerous results, ie. --collections, --item-collection, --recursive ([#138](https://github.com/stac-utils/stac-check/pull/138)) +- Support for --verbose flag to show verbose results summary ([#138](https://github.com/stac-utils/stac-check/pull/138)) +- Added `--output`/`-o` option to save validation results to a file ([#138](https://github.com/stac-utils/stac-check/pull/138)) +- Tests for CLI options ([#138](https://github.com/stac-utils/stac-check/pull/138)) + ## [v1.10.1] - 2025-06-21 ### Fixed @@ -285,7 +294,8 @@ The format is (loosely) based on [Keep a Changelog](http://keepachangelog.com/) - Validation from stac-validator 2.3.0 - Links and assets validation checks -[Unreleased]: https://github.com/stac-utils/stac-check/compare/v1.10.1...main +[Unreleased]: https://github.com/stac-utils/stac-check/compare/v1.11.0...main +[v1.11.0]: https://github.com/stac-utils/stac-check/compare/v1.10.1...v1.11.0 [v1.10.1]: https://github.com/stac-utils/stac-check/compare/v1.10.0...v1.10.1 [v1.10.0]: https://github.com/stac-utils/stac-check/compare/v1.9.1...v1.10.0 [v1.9.1]: https://github.com/stac-utils/stac-check/compare/v1.9.0...v1.9.1 diff --git a/README.md b/README.md index 9307705..ef92e98 100644 --- a/README.md +++ b/README.md @@ -111,6 +111,7 @@ Options: multiple times. --pydantic Use stac-pydantic for enhanced validation with Pydantic models. --verbose Show verbose error messages. + -o, --output FILE Save output to the specified file. --item-collection Validate item collection response. Can be combined with --pages. Defaults to one page. --collections Validate collections endpoint response. Can be combined with diff --git a/setup.py b/setup.py index 9b730c2..49dd399 100644 --- a/setup.py +++ b/setup.py @@ -3,7 +3,7 @@ from setuptools import find_packages, setup -__version__ = "1.10.1" +__version__ = "1.11.0" with open("README.md", "r") as fh: long_description = fh.read() diff --git a/stac_check/api_lint.py b/stac_check/api_lint.py index 4ffafb7..ed529d4 100644 --- a/stac_check/api_lint.py +++ b/stac_check/api_lint.py @@ -37,11 +37,13 @@ def __init__( object_list_key: str, pages: Optional[int] = 1, headers: Optional[Dict] = None, + verbose: bool = False, ): self.source = source self.object_list_key = object_list_key self.pages = pages if pages is not None else 1 self.headers = headers or {} + self.verbose = verbose self.version = None self.validator_version = self._get_validator_version() @@ -148,7 +150,7 @@ def lint_all(self) -> List[Dict]: results_by_url = {} for obj, obj_url in self.iterate_objects(): try: - linter = Linter(obj) + linter = Linter(obj, verbose=self.verbose) msg = dict(linter.message) msg["path"] = obj_url msg["best_practices"] = linter.best_practices_msg diff --git a/stac_check/cli.py b/stac_check/cli.py index 6750380..bba4a70 100644 --- a/stac_check/cli.py +++ b/stac_check/cli.py @@ -1,5 +1,6 @@ import importlib.metadata import sys +from typing import Optional import click @@ -12,6 +13,7 @@ recursive_message, ) from stac_check.lint import Linter +from stac_check.utilities import handle_output @click.option( @@ -48,6 +50,12 @@ @click.option( "-l", "--links", is_flag=True, help="Validate links for format and response." ) +@click.option( + "--output", + "-o", + type=click.Path(dir_okay=False, writable=True), + help="Save output to the specified file. Only works with --collections, --item-collection, or --recursive.", +) @click.option( "--no-assets-urls", is_flag=True, @@ -74,19 +82,44 @@ @click.argument("file") @click.version_option(version=importlib.metadata.distribution("stac-check").version) def main( - file, - collections, - item_collection, - pages, - recursive, - max_depth, - assets, - links, - no_assets_urls, - header, - pydantic, - verbose, -): + file: str, + collections: bool, + item_collection: bool, + pages: Optional[int], + recursive: bool, + max_depth: Optional[int], + assets: bool, + links: bool, + no_assets_urls: bool, + header: tuple[tuple[str, str], ...], + pydantic: bool, + verbose: bool, + output: Optional[str], +) -> None: + """Main entry point for the stac-check CLI. + + Args: + file: The STAC file or URL to validate + collections: Validate a collections endpoint + item_collection: Validate an item collection + pages: Number of pages to validate (for API endpoints) + recursive: Recursively validate linked STAC objects + max_depth: Maximum depth for recursive validation + assets: Validate assets + links: Validate links + no_assets_urls: Disable URL validation for assets + header: Additional HTTP headers + pydantic: Use stac-pydantic for validation + verbose: Show verbose output + output: Save output to file (only with --collections, --item-collection, or --recursive) + """ + # Check if output is used without --collections, --item-collection, or --recursive + if output and not any([collections, item_collection, recursive]): + click.echo( + "Error: --output can only be used with --collections, --item-collection, or --recursive", + err=True, + ) + sys.exit(1) # Check if pydantic validation is requested but not installed if pydantic: try: @@ -99,8 +132,52 @@ def main( ) pydantic = False - if not collections and not item_collection: - # Create a standard Linter for single file or recursive validation + if collections or item_collection: + # Handle API-based validation (collections or item collections) + api_linter = ApiLinter( + source=file, + object_list_key="collections" if collections else "features", + pages=pages if pages else 1, + headers=dict(header), + verbose=verbose, + ) + results = api_linter.lint_all() + + # Create a dummy Linter instance for display purposes + display_linter = Linter( + file, + assets=assets, + links=links, + headers=dict(header), + pydantic=pydantic, + verbose=verbose, + ) + + # Show intro message in the terminal + intro_message(display_linter) + + # Define output generation function (without intro message since we already showed it) + def generate_output(): + if collections: + collections_message( + api_linter, + results=results, + cli_message_func=cli_message, + verbose=verbose, + ) + elif item_collection: + item_collection_message( + api_linter, + results=results, + cli_message_func=cli_message, + verbose=verbose, + ) + + # Handle output (without duplicating the intro message) + handle_output(output, generate_output) + sys.exit(0 if all(msg.get("valid_stac") is True for msg in results) else 1) + else: + # Handle file-based validation (single file or recursive) linter = Linter( file, assets=assets, @@ -112,34 +189,20 @@ def main( pydantic=pydantic, verbose=verbose, ) - intro_message(linter) - # If recursive validation is enabled, use recursive_message - if recursive: - # Pass the cli_message function to avoid circular imports - recursive_message(linter, cli_message_func=cli_message) - else: - # Otherwise, just display the standard CLI message - cli_message(linter) - sys.exit(0 if linter.valid_stac else 1) - else: - if item_collection: - object_list_key = "features" - elif collections: - object_list_key = "collections" + intro_message(linter) - linter = ApiLinter( - source=file, - object_list_key=object_list_key, - pages=pages, - headers=dict(header), - ) - results = linter.lint_all() + # Show intro message in the terminal intro_message(linter) - if collections: - collections_message(linter, results=results, cli_message_func=cli_message) - elif item_collection: - item_collection_message( - linter, results=results, cli_message_func=cli_message - ) - sys.exit(0 if all(msg.get("valid_stac") is True for msg in results) else 1) + + # Define output generation function (without intro message since we already showed it) + def generate_output(): + if recursive: + recursive_message(linter, cli_message_func=cli_message, verbose=verbose) + else: + cli_message(linter) + + # Handle output (without duplicating the intro message) + handle_output(output if recursive else None, generate_output) + + sys.exit(0 if linter.valid_stac else 1) diff --git a/stac_check/display_messages.py b/stac_check/display_messages.py index 2660a8f..4a6062d 100644 --- a/stac_check/display_messages.py +++ b/stac_check/display_messages.py @@ -216,12 +216,93 @@ def _display_disclaimer() -> None: click.secho() +def _display_validation_summary( + results: List[Dict[str, Any]], verbose: bool = False +) -> None: + """Display a summary of validation results, including warnings and best practice issues. + + Args: + results: List of validation result dictionaries + verbose: Whether to show detailed output + """ + passed = 0 + failed = [] + warnings = [] + all_paths = [] + + for result in results: + path = result.get("path", "unknown") + all_paths.append(path) + + # Check for validation status + if result.get("valid_stac"): + passed += 1 + else: + failed.append(path) + + # Check for best practice warnings in the result + best_practices = [] + if result.get("best_practices"): + best_practices = [ + p + for p in result["best_practices"] + if p and p.strip() and p != "STAC Best Practices: " + ] + # Also check for best practices in the message if it exists + elif ( + result.get("message") + and isinstance(result["message"], dict) + and result["message"].get("best_practices") + ): + best_practices = [ + p + for p in result["message"]["best_practices"] + if p and p.strip() and p != "STAC Best Practices: " + ] + + # Only add to warnings if there are actual messages + if best_practices: + warnings.append((path, best_practices)) + + click.secho("\n Validation Summary", bold=True, bg="black", fg="white") + click.secho() + click.secho(f"✅ Passed: {passed}/{len(all_paths)}") + + if failed: + click.secho(f"❌ Failed: {len(failed)}/{len(all_paths)}", fg="red") + click.secho("\nFailed Assets:", fg="red") + for path in failed: + click.secho(f" - {path}") + + if warnings: + click.secho( + f"\n⚠️ Best Practice Warnings ({len(warnings)} assets)", fg="yellow" + ) + if verbose or len(warnings) <= 12: + for path, msgs in warnings: + click.secho(f"\n {path}:", fg="yellow") + for msg in msgs: + click.secho(f" • {msg}", fg="yellow") + else: + click.secho(" (Use --verbose to see details)", fg="yellow") + + click.secho(f"\n🔍 All {len(all_paths)} Assets Checked") + if verbose or len(all_paths) <= 12: + for path in all_paths: + click.secho(f" - {path}") + else: + click.secho(" (Use --verbose to see all assets)", fg="yellow") + + click.secho() + + def _display_validation_results( results: List[Dict[str, Any]], title: str, metadata: Optional[Dict[str, Any]] = None, cli_message_func: Optional[Callable[[Linter], None]] = None, create_linter_func: Optional[Callable[[Dict[str, Any]], Linter]] = None, + verbose: bool = False, ) -> None: """Shared helper function to display validation results consistently. @@ -253,7 +334,6 @@ def _display_validation_results( click.secho(f"{key} = {value}") click.secho("-------------------------") - for count, msg in enumerate(results): # Get the path or use a fallback path = msg.get("path", f"(unknown-{count + 1})") @@ -265,14 +345,32 @@ def _display_validation_results( if create_linter_func: item_linter = create_linter_func(msg) - # Set validation status and error info for invalid items - if not msg.get("valid_stac", True): - item_linter.valid_stac = False - item_linter.error_type = msg.get("error_type") - item_linter.error_msg = msg.get("error_message") - - # Display using the provided message function - cli_message_func(item_linter) + # If create_linter_func returns None (for recursive validation), use fallback + if item_linter is None: + _display_fallback_message(msg) + else: + # Set validation status and error info for invalid items + if not msg.get("valid_stac", True): + item_linter.valid_stac = False + item_linter.error_type = msg.get("error_type") + item_linter.error_msg = msg.get("error_message") + + # Ensure best practices are included in the result + if ( + hasattr(item_linter, "best_practices_msg") + and item_linter.best_practices_msg + ): + # Skip the first line which is just the header + bp_msgs = [ + msg + for msg in item_linter.best_practices_msg[1:] + if msg.strip() + ] + if bp_msgs: + msg["best_practices"] = bp_msgs + + # Display using the provided message function + cli_message_func(item_linter) else: # No linter creation function provided, use fallback _display_fallback_message(msg) @@ -282,11 +380,15 @@ def _display_validation_results( click.secho("-------------------------") + # Display summary at the end for better visibility with many items + _display_validation_summary(results, verbose=verbose) + def item_collection_message( linter: ApiLinter, results: Optional[List[Dict[str, Any]]] = None, cli_message_func: Optional[Callable[[Linter], None]] = None, + verbose: bool = False, ) -> None: """Displays messages related to the validation of assets in a feature collection. @@ -323,6 +425,7 @@ def create_api_linter(msg): metadata={"Pages": linter.pages}, cli_message_func=cli_message_func, create_linter_func=create_api_linter, + verbose=verbose, ) @@ -365,13 +468,17 @@ def _display_fallback_message( # Display best practices bp = msg.get("best_practices", []) - if bp and len(bp) > 0: - click.secho() - click.secho("\n STAC Best Practices: ", bg="blue") - click.secho() + # Filter out empty strings and the default "STAC Best Practices: " message + bp = [p for p in bp if p and p.strip() and p != "STAC Best Practices: "] + + if bp: + click.echo() + click.secho("\nSTAC Best Practices: ", bg="blue") + click.echo() for practice in bp: - if practice: # Skip empty strings - click.secho(practice, fg="black") + click.echo(f" • {practice}", fg="black") + # Update the best_practices in the message for the summary + msg["best_practices"] = bp # Display geometry errors geo = msg.get("geometry_errors", []) @@ -390,6 +497,7 @@ def collections_message( linter: ApiLinter, results: Optional[List[Dict[str, Any]]] = None, cli_message_func: Optional[Callable[[Linter], None]] = None, + verbose: bool = False, ) -> None: """Displays messages related to the validation of STAC collections from a collections endpoint. @@ -426,11 +534,14 @@ def create_collection_linter(msg): metadata={"Pages": linter.pages}, cli_message_func=cli_message_func, create_linter_func=create_collection_linter, + verbose=verbose, ) def recursive_message( - linter: Linter, cli_message_func: Optional[Callable[[Linter], None]] = None + linter: Linter, + cli_message_func: Optional[Callable[[Linter], None]] = None, + verbose: bool = False, ) -> None: """Displays messages related to the recursive validation of assets in a collection or catalog. @@ -462,6 +573,7 @@ def create_recursive_linter(msg): metadata={"Max-depth": linter.max_depth}, cli_message_func=cli_message_func, create_linter_func=create_recursive_linter, + verbose=verbose, ) diff --git a/stac_check/utilities.py b/stac_check/utilities.py index 1cf72b1..0ab222e 100644 --- a/stac_check/utilities.py +++ b/stac_check/utilities.py @@ -1,3 +1,9 @@ +import contextlib +from typing import Callable + +import click + + def determine_asset_type(data): """Determine the STAC asset type from the given data dictionary. @@ -35,6 +41,32 @@ def determine_asset_type(data): return "" +def handle_output( + output_file: str, callback: Callable[[], None], output_path: str = None +) -> None: + """Helper function to handle output redirection to a file or stdout. + + Args: + output_file: Path to the output file, or None to use stdout + callback: Function that performs the actual output generation + output_path: Optional path to display in the success message + """ + + if output_file: + with open(output_file, "w") as f: + with contextlib.redirect_stdout(f): + callback() + click.secho( + f"Output written to {output_path or output_file}", + fg="green", + err=True, + bold=True, + ) + click.secho() + else: + callback() + + def format_verbose_error(error_data): """Format verbose error data into a human-readable string.""" if not error_data or not isinstance(error_data, dict): diff --git a/tests/test_cli.py b/tests/test_cli.py new file mode 100644 index 0000000..42de225 --- /dev/null +++ b/tests/test_cli.py @@ -0,0 +1,220 @@ +"""Tests for the stac-check CLI.""" + +import os +import tempfile +from unittest.mock import MagicMock, patch + +import pytest +from click.testing import CliRunner + +from stac_check.cli import main as cli_main + + +@pytest.fixture +def runner(): + """Fixture for invoking command-line interfaces.""" + return CliRunner() + + +def test_cli_help(runner): + """Test the CLI help output.""" + result = runner.invoke(cli_main, ["--help"]) + assert result.exit_code == 0 + assert "Show this message and exit." in result.output + + +def test_cli_version(runner): + """Test the --version flag.""" + result = runner.invoke(cli_main, ["--version"]) + assert result.exit_code == 0 + # The version output is in the format: main, version X.Y.Z + assert "version" in result.output + + +def test_cli_validate_local_file(runner): + """Test validating a local file.""" + test_file = os.path.join( + os.path.dirname(__file__), "../sample_files/1.0.0/core-item.json" + ) + result = runner.invoke(cli_main, [test_file]) + assert result.exit_code == 0 + assert "Passed: True" in result.output + + +def test_cli_validate_recursive(runner): + """Test recursive validation.""" + test_dir = os.path.join( + os.path.dirname(__file__), "../sample_files/1.0.0/catalog-with-bad-item.json" + ) + result = runner.invoke(cli_main, [test_dir, "--recursive"]) + assert result.exit_code == 0 + assert "Assets Checked" in result.output + + +def test_cli_output_to_file(runner): + """Test output to file with --output flag.""" + test_file = os.path.join( + os.path.dirname(__file__), "../sample_files/1.0.0/core-item.json" + ) + with tempfile.NamedTemporaryFile(delete=False) as tmp: + output_file = tmp.name + + try: + result = runner.invoke( + cli_main, [test_file, "--recursive", "--output", output_file] + ) + assert result.exit_code == 0 + assert os.path.exists(output_file) + with open(output_file, "r") as f: + content = f.read() + assert "Passed: True" in content + finally: + if os.path.exists(output_file): + os.unlink(output_file) + + +def test_cli_collections(runner): + """Test --collections flag with mock.""" + with patch("stac_check.cli.ApiLinter") as mock_api_linter, patch( + "stac_check.cli.Linter" + ) as mock_linter: + # Mock ApiLinter instance + mock_api_instance = MagicMock() + mock_api_instance.lint_all.return_value = [{"valid_stac": True}] + mock_api_linter.return_value = mock_api_instance + + # Mock Linter instance used for display + mock_linter_instance = MagicMock() + mock_linter.return_value = mock_linter_instance + + result = runner.invoke( + cli_main, + ["https://example.com/collections", "--collections", "--pages", "1"], + ) + + assert result.exit_code == 0 + mock_api_linter.assert_called_once_with( + source="https://example.com/collections", + object_list_key="collections", + pages=1, + headers={}, + verbose=False, + ) + + +def test_cli_item_collection(runner): + """Test --item-collection flag with mock.""" + with patch("stac_check.cli.ApiLinter") as mock_api_linter, patch( + "stac_check.cli.Linter" + ) as mock_linter: + # Mock ApiLinter instance + mock_api_instance = MagicMock() + mock_api_instance.lint_all.return_value = [{"valid_stac": True}] + mock_api_linter.return_value = mock_api_instance + + # Mock Linter instance used for display + mock_linter_instance = MagicMock() + mock_linter.return_value = mock_linter_instance + + result = runner.invoke( + cli_main, ["https://example.com/items", "--item-collection", "--pages", "2"] + ) + + assert result.exit_code == 0 + mock_api_linter.assert_called_once_with( + source="https://example.com/items", + object_list_key="features", + pages=2, + headers={}, + verbose=False, + ) + + +def test_cli_output_without_required_flags(runner): + """Test that --output requires --collections, --item-collection, or --recursive.""" + with tempfile.NamedTemporaryFile() as tmp: + result = runner.invoke( + cli_main, ["https://example.com/catalog.json", "--output", tmp.name] + ) + assert result.exit_code == 1 + assert ( + "--output can only be used with --collections, --item-collection, or --recursive" + in result.output + ) + + +def test_cli_verbose_flag(runner): + """Test that --verbose flag is passed correctly.""" + with patch("stac_check.cli.Linter") as mock_linter: + mock_instance = MagicMock() + mock_linter.return_value = mock_instance + + test_file = os.path.join( + os.path.dirname(__file__), "../sample_files/1.0.0/core-item.json" + ) + result = runner.invoke(cli_main, [test_file, "--verbose"]) + + assert result.exit_code == 0 + mock_linter.assert_called_once() + assert mock_linter.call_args[1]["verbose"] is True + + +def test_cli_headers(runner): + """Test that custom headers are passed correctly.""" + with patch("stac_check.cli.Linter") as mock_linter: + mock_instance = MagicMock() + mock_linter.return_value = mock_instance + + test_file = os.path.join( + os.path.dirname(__file__), "../sample_files/1.0.0/core-item.json" + ) + # The header format is: --header KEY VALUE (space-separated, not colon-separated) + result = runner.invoke( + cli_main, + [ + test_file, + "--header", + "Authorization", + "Bearer token", + "--header", + "X-Custom", + "value", + ], + ) + + assert result.exit_code == 0 + mock_linter.assert_called_once() + # The headers should be passed as a dictionary to the Linter + headers = mock_linter.call_args[1]["headers"] + assert isinstance(headers, dict) + assert headers.get("Authorization") == "Bearer token" + assert headers.get("X-Custom") == "value" + + +def test_cli_pydantic_flag(runner): + """Test that the --pydantic flag is passed correctly.""" + with patch("stac_check.cli.Linter") as mock_linter, patch( + "stac_check.cli.importlib.import_module" + ): + mock_instance = MagicMock() + mock_linter.return_value = mock_instance + + test_file = os.path.join( + os.path.dirname(__file__), "../sample_files/1.0.0/core-item.json" + ) + + # Test with --pydantic flag + result = runner.invoke(cli_main, [test_file, "--pydantic"]) + + assert result.exit_code == 0 + mock_linter.assert_called_once() + # Check that pydantic=True was passed to Linter + assert mock_linter.call_args[1]["pydantic"] is True + + # Test without --pydantic flag (should default to False) + mock_linter.reset_mock() + result = runner.invoke(cli_main, [test_file]) + + assert result.exit_code == 0 + mock_linter.assert_called_once() + assert mock_linter.call_args[1]["pydantic"] is False