Merge pull request #1952 from braingram/asdftool_validate

add asdftool validate

Author: braingram

asdf/_commands/__init__.py

@@ -7,6 +7,7 @@
 from .search import search
 from .tags import list_tags
 from .to_yaml import to_yaml
+from .validate import validate
 
 __all__ = [
     "defragment",
@@ -19,4 +20,5 @@
     "list_tags",
     "search",
     "to_yaml",
+    "validate",
 ]

asdf/_commands/validate.py

@@ -0,0 +1,58 @@
+"""
+Commands for validating ASDF files
+"""
+
+import asdf
+
+from .main import Command
+
+__all__ = ["validate"]
+
+
+class Validate(Command):
+    @classmethod
+    def setup_arguments(cls, subparsers):
+        parser = subparsers.add_parser(
+            "validate",
+            help="Validates an ASDF file.",
+            description=(
+                "Validates tree against all tagged schemas (and optionally a custom schema) "
+                "and blocks with stored checksums."
+            ),
+        )
+
+        parser.add_argument("filename", help="path to ASDF file")
+        parser.add_argument(
+            "--custom-schema",
+            type=str,
+            help="path or URI of custom schema",
+        )
+        parser.add_argument(
+            "--skip-block-validation",
+            default=False,
+            action="store_true",
+            help="Do not compare blocks against stored checksums.",
+        )
+        parser.set_defaults(func=cls.run)
+
+        return parser
+
+    @classmethod
+    def run(cls, args):
+        validate(args.filename, args.custom_schema, args.skip_block_validation)
+
+
+def validate(filename, custom_schema, skip_block_validation):
+    # if we are skipping checksums we can lazy load, otherwise don't
+    with asdf.open(
+        filename,
+        custom_schema=custom_schema,
+        validate_checksums=not skip_block_validation,
+        lazy_load=skip_block_validation,
+    ) as af:  # noqa: F841
+        msg = f"{filename} is valid"
+        if custom_schema:
+            msg += f", conforms to {custom_schema}"
+        if not skip_block_validation:
+            msg += ", and block checksums match contents"
+        print(msg)

asdf/_tests/commands/test_validate.py

@@ -0,0 +1,84 @@
+import numpy as np
+import pytest
+
+import asdf
+from asdf._commands import main
+
+
+@pytest.fixture()
+def valid_file_path(tmp_path):
+    path = tmp_path / "valid_file.asdf"
+    asdf.dump({"foo": 42, "arr": np.arange(42)}, path)
+    return path
+
+
+@pytest.fixture()
+def invalid_file_path(tmp_path):
+    # don't use asdf as we're intentionally making an invalid file
+    contents = b"""#ASDF 1.0.0
+#ASDF_STANDARD 1.6.0
+%YAML 1.1
+%TAG ! tag:stsci.edu:asdf/
+--- !core/asdf-1.1.0
+arr: !core/ndarray-1.1.0
+  data:
+    nested: error
+..."""
+    path = tmp_path / "invalid_file.asdf"
+    with path.open("wb") as f:
+        f.write(contents)
+    return path
+
+
+@pytest.fixture()
+def custom_schema_path(tmp_path):
+    contents = """%YAML 1.1
+---
+id: "http://example.com/schemas/your-custom-schema"
+$schema: "http://stsci.edu/schemas/yaml-schema/draft-01"
+type: object
+properties:
+  foo:
+    type: string"""
+    path = tmp_path / "custom_schema.yaml"
+    with path.open("w") as f:
+        f.write(contents)
+    return path
+
+
+@pytest.fixture()
+def bad_blocks_file_path(tmp_path):
+    to_replace = b"REPLACE"
+    replacement = b"ABCDEFG"
+    buff = asdf.dumps({"arr": np.frombuffer(to_replace, np.uint8)})
+    buff = buff.replace(to_replace, replacement)
+    path = tmp_path / "bad_block.asdf"
+    with path.open("wb") as f:
+        f.write(buff)
+    return path
+
+
+def test_valid(capsys, valid_file_path):
+    assert main.main_from_args(["validate", str(valid_file_path)]) == 0
+
+    captured = capsys.readouterr()
+    assert "valid" in captured.out
+
+
+def test_invalid(invalid_file_path):
+    with pytest.raises(asdf.ValidationError):
+        main.main_from_args(["validate", str(invalid_file_path)])
+
+
+def test_custom_schema(valid_file_path, custom_schema_path):
+    with pytest.raises(asdf.ValidationError):
+        main.main_from_args(["validate", str(valid_file_path), "--custom-schema", str(custom_schema_path)])
+
+
+def test_block_checksum(bad_blocks_file_path):
+    with pytest.raises(ValueError, match="does not match given checksum"):
+        main.main_from_args(["validate", str(bad_blocks_file_path)])
+
+
+def test_skip_block_validation(bad_blocks_file_path):
+    assert main.main_from_args(["validate", str(bad_blocks_file_path), "--skip-block-validation"]) == 0

changes/1952.feature.rst

@@ -0,0 +1 @@
+Add ``asdftool validate`` to validate ASDF files.

docs/asdf/asdf_tool.rst

@@ -30,4 +30,6 @@ useful operations:
   - ``to_yaml``: Inline all of the data in an ASDF file so that it is
     pure YAML.
 
+  - ``validate``: Validate an ASDF file's blocks and schema.
+
 Run ``asdftool --help`` for more information.