feat(anta): Add support for JSON catalogs (#739)

anta/catalog.py

@@ -10,14 +10,14 @@
 import math
 from collections import defaultdict
 from inspect import isclass
+from json import load as json_load
 from pathlib import Path
-from typing import TYPE_CHECKING, Any, Optional, Union
+from typing import TYPE_CHECKING, Any, Literal, Optional, Union
 
-import yaml
 from pydantic import BaseModel, ConfigDict, RootModel, ValidationError, ValidationInfo, field_validator, model_serializer, model_validator
 from pydantic.types import ImportString
 from pydantic_core import PydanticCustomError
-from yaml import YAMLError, safe_load
+from yaml import YAMLError, safe_dump, safe_load
 
 from anta.logger import anta_log_exception
 from anta.models import AntaTest
@@ -238,7 +238,16 @@ def yaml(self) -> str:
         # This could be improved.
         # https://github.com/pydantic/pydantic/issues/1043
         # Explore if this worth using this: https://github.com/NowanIlfideme/pydantic-yaml
-        return yaml.safe_dump(yaml.safe_load(self.model_dump_json(serialize_as_any=True, exclude_unset=True)), indent=2, width=math.inf)
+        return safe_dump(safe_load(self.model_dump_json(serialize_as_any=True, exclude_unset=True)), indent=2, width=math.inf)
+
+    def to_json(self) -> str:
+        """Return a JSON representation string of this model.
+
+        Returns
+        -------
+            The JSON representation string of this model.
+        """
+        return self.model_dump_json(serialize_as_any=True, exclude_unset=True, indent=2)
 
 
 class AntaCatalog:
@@ -298,19 +307,24 @@ def tests(self, value: list[AntaTestDefinition]) -> None:
         self._tests = value
 
     @staticmethod
-    def parse(filename: str | Path) -> AntaCatalog:
+    def parse(filename: str | Path, file_format: Literal["yaml", "json"] = "yaml") -> AntaCatalog:
         """Create an AntaCatalog instance from a test catalog file.
 
         Parameters
         ----------
-            filename: Path to test catalog YAML file
+            filename: Path to test catalog YAML or JSON fil
+            file_format: Format of the file, either 'yaml' or 'json'
 
         """
+        if file_format not in ["yaml", "json"]:
+            message = f"'{file_format}' is not a valid format for an AntaCatalog file. Only 'yaml' and 'json' are supported."
+            raise ValueError(message)
+
         try:
             file: Path = filename if isinstance(filename, Path) else Path(filename)
             with file.open(encoding="UTF-8") as f:
-                data = safe_load(f)
-        except (TypeError, YAMLError, OSError) as e:
+                data = safe_load(f) if file_format == "yaml" else json_load(f)
+        except (TypeError, YAMLError, OSError, ValueError) as e:
             message = f"Unable to parse ANTA Test Catalog file '{filename}'"
             anta_log_exception(e, message, logger)
             raise

anta/cli/nrfu/__init__.py

@@ -116,6 +116,7 @@ def nrfu(
     ignore_status: bool,
     ignore_error: bool,
     dry_run: bool,
+    catalog_format: str = "yaml",
 ) -> None:
     """Run ANTA tests on selected inventory devices."""
     # If help is invoke somewhere, skip the command
@@ -129,6 +130,7 @@ def nrfu(
     ctx.obj["ignore_error"] = ignore_error
     ctx.obj["hide"] = set(hide) if hide else None
     ctx.obj["catalog"] = catalog
+    ctx.obj["catalog_format"] = catalog_format
     ctx.obj["inventory"] = inventory
     ctx.obj["tags"] = tags
     ctx.obj["device"] = device

anta/cli/utils.py

@@ -268,7 +268,7 @@ def catalog_options(f: Callable[..., Any]) -> Callable[..., Any]:
         "-c",
         envvar="ANTA_CATALOG",
         show_envvar=True,
-        help="Path to the test catalog YAML file",
+        help="Path to the test catalog file",
         type=click.Path(
             file_okay=True,
             dir_okay=False,
@@ -278,19 +278,29 @@ def catalog_options(f: Callable[..., Any]) -> Callable[..., Any]:
         ),
         required=True,
     )
+    @click.option(
+        "--catalog-format",
+        envvar="ANTA_CATALOG_FORMAT",
+        show_envvar=True,
+        help="Format of the catalog file, either 'yaml' or 'json'",
+        default="yaml",
+        type=click.Choice(["yaml", "json"], case_sensitive=False),
+    )
     @click.pass_context
     @functools.wraps(f)
     def wrapper(
         ctx: click.Context,
         *args: tuple[Any],
         catalog: Path,
+        catalog_format: str,
         **kwargs: dict[str, Any],
     ) -> Any:
         # If help is invoke somewhere, do not parse catalog
         if ctx.obj.get("_anta_help"):
             return f(*args, catalog=None, **kwargs)
         try:
-            c = AntaCatalog.parse(catalog)
+            file_format = catalog_format.lower()
+            c = AntaCatalog.parse(catalog, file_format=file_format)  # type: ignore[arg-type]
         except (TypeError, ValueError, YAMLError, OSError):
             ctx.exit(ExitCode.USAGE_ERROR)
         return f(*args, catalog=c, **kwargs)

docs/cli/check.md

@@ -27,10 +27,12 @@ Commands:
 ```bash
 Usage: anta check catalog [OPTIONS]
 
-  Check that the catalog is valid
+  Check that the catalog is valid.
 
 Options:
-  -c, --catalog FILE  Path to the test catalog YAML file  [env var:
-                      ANTA_CATALOG; required]
-  --help              Show this message and exit.
+  -c, --catalog FILE            Path to the test catalog file  [env var:
+                                ANTA_CATALOG; required]
+  --catalog-format [yaml|json]  Format of the catalog file, either 'yaml' or
+                                'json'  [env var: ANTA_CATALOG_FORMAT]
+  --help                        Show this message and exit.
 ```

docs/snippets/anta_nrfu_help.txt

@@ -29,8 +29,10 @@ Options:
                                   ANTA_INVENTORY; required]
   --tags TEXT                     List of tags using comma as separator:
                                   tag1,tag2,tag3.  [env var: ANTA_TAGS]
-  -c, --catalog FILE              Path to the test catalog YAML file  [env
-                                  var: ANTA_CATALOG; required]
+  -c, --catalog FILE              Path to the test catalog file  [env var:
+                                  ANTA_CATALOG; required]
+  --catalog-format [yaml|json]    Format of the catalog file, either 'yaml' or
+                                  'json'  [env var: ANTA_CATALOG_FORMAT]
   -d, --device TEXT               Run tests on a specific device. Can be
                                   provided multiple times.
   -t, --test TEXT                 Run a specific test. Can be provided

docs/usage-inventory-catalog.md

@@ -78,14 +78,24 @@ A test catalog is an instance of the [AntaCatalog](./api/catalog.md#anta.catalog
 
 In addition to the inventory file, you also have to define a catalog of tests to execute against your devices. This catalog list all your tests, their inputs and their tags.
 
-A valid test catalog file must have the following structure:
+A valid test catalog file must have the following structure in either YAML or JSON:
 ```yaml
 ---
 <Python module>:
     - <AntaTest subclass>:
         <AntaTest.Input compliant dictionary>
 ```
 
+```json
+{
+  "<Python module>": [
+    {
+      "<AntaTest subclass>": <AntaTest.Input compliant dictionary>
+    }
+  ]
+}
+```
+
 ### Example
 
 ```yaml
@@ -108,6 +118,43 @@ anta.tests.connectivity:
         custom_field: "Test run by John Doe"
 ```
 
+or equivalent in JSON:
+
+```json
+{
+  "anta.tests.connectivity": [
+    {
+      "VerifyReachability": {
+        "result_overwrite": {
+          "description": "Test with overwritten description",
+          "categories": [
+            "Overwritten category 1"
+          ],
+          "custom_field": "Test run by John Doe"
+        },
+        "filters": {
+          "tags": [
+            "leaf"
+          ]
+        },
+        "hosts": [
+          {
+            "destination": "1.1.1.1",
+            "source": "Management0",
+            "vrf": "MGMT"
+          },
+          {
+            "destination": "8.8.8.8",
+            "source": "Management0",
+            "vrf": "MGMT"
+          }
+        ]
+      }
+    }
+  ]
+}
+```
+
 It is also possible to nest Python module definition:
 ```yaml
 anta.tests:
@@ -165,7 +212,7 @@ anta.tests.software:
   - VerifyEOSVersion:
 ```
 
-It will load the test `VerifyEOSVersion` located in `anta.tests.software`. But since this test has mandatory inputs, we need to provide them as a dictionary in the YAML file:
+It will load the test `VerifyEOSVersion` located in `anta.tests.software`. But since this test has mandatory inputs, we need to provide them as a dictionary in the YAML or JSON file:
 
 ```yaml
 anta.tests.software:
@@ -176,6 +223,21 @@ anta.tests.software:
         - 4.26.1F
 ```
 
+```json
+{
+  "anta.tests.software": [
+    {
+      "VerifyEOSVersion": {
+        "versions": [
+          "4.25.4M",
+          "4.31.1F"
+        ]
+      }
+    }
+  ]
+}
+```
+
 The following example is a very minimal test catalog:
 
 ```yaml

tests/data/test_catalog.json

@@ -0,0 +1,11 @@
+{
+  "anta.tests.software": [
+    {
+      "VerifyEOSVersion": {
+        "versions": [
+          "4.31.1F"
+        ]
+      }
+    }
+  ]
+}

tests/data/test_catalog_invalid_json.json

@@ -0,0 +1 @@
+{aasas"anta.tests.software":[{"VerifyEOSVersion":{"versions":["4.31.1F"]}}]}

tests/units/cli/nrfu/test__init__.py

@@ -49,6 +49,13 @@ def test_anta_nrfu_dry_run(click_runner: CliRunner) -> None:
     assert "Dry-run" in result.output
 
 
+def test_anta_nrfu_wrong_catalog_format(click_runner: CliRunner) -> None:
+    """Test anta nrfu --dry-run, catalog is given via env."""
+    result = click_runner.invoke(anta, ["nrfu", "--dry-run", "--catalog-format", "toto"])
+    assert result.exit_code == ExitCode.USAGE_ERROR
+    assert "Invalid value for '--catalog-format': 'toto' is not one of 'yaml', 'json'." in result.output
+
+
 def test_anta_password_required(click_runner: CliRunner) -> None:
     """Test that password is provided."""
     env = default_anta_env()

tests/units/test_catalog.py

@@ -5,6 +5,7 @@
 
 from __future__ import annotations
 
+from json import load as json_load
 from pathlib import Path
 from typing import Any
 
@@ -42,6 +43,14 @@
             (VerifyEOSVersion, VerifyEOSVersion.Input(versions=["4.31.1F"])),
         ],
     },
+    {
+        "name": "test_catalog",
+        "filename": "test_catalog.json",
+        "file_format": "json",
+        "tests": [
+            (VerifyEOSVersion, VerifyEOSVersion.Input(versions=["4.31.1F"])),
+        ],
+    },
     {
         "name": "test_catalog_with_tags",
         "filename": "test_catalog_with_tags.yml",
@@ -83,6 +92,18 @@
     },
 ]
 CATALOG_PARSE_FAIL_DATA: list[dict[str, Any]] = [
+    {
+        "name": "undefined_tests",
+        "filename": "test_catalog_wrong_format.toto",
+        "file_format": "toto",
+        "error": "'toto' is not a valid format for an AntaCatalog file. Only 'yaml' and 'json' are supported.",
+    },
+    {
+        "name": "invalid_json",
+        "filename": "test_catalog_invalid_json.json",
+        "file_format": "json",
+        "error": "JSONDecodeError",
+    },
     {
         "name": "undefined_tests",
         "filename": "test_catalog_with_undefined_tests.yml",
@@ -185,7 +206,7 @@ class TestAntaCatalog:
     @pytest.mark.parametrize("catalog_data", INIT_CATALOG_DATA, ids=generate_test_ids_list(INIT_CATALOG_DATA))
     def test_parse(self, catalog_data: dict[str, Any]) -> None:
         """Instantiate AntaCatalog from a file."""
-        catalog: AntaCatalog = AntaCatalog.parse(DATA_DIR / catalog_data["filename"])
+        catalog: AntaCatalog = AntaCatalog.parse(DATA_DIR / catalog_data["filename"], file_format=catalog_data.get("file_format", "yaml"))
 
         assert len(catalog.tests) == len(catalog_data["tests"])
         for test_id, (test, inputs_data) in enumerate(catalog_data["tests"]):
@@ -211,7 +232,8 @@ def test_from_dict(self, catalog_data: dict[str, Any]) -> None:
         """Instantiate AntaCatalog from a dict."""
         file = DATA_DIR / catalog_data["filename"]
         with file.open(encoding="UTF-8") as file:
-            data = safe_load(file)
+            file_format = catalog_data.get("file_format", "yaml")
+            data = safe_load(file) if file_format == "yaml" else json_load(file)
             catalog: AntaCatalog = AntaCatalog.from_dict(data)
 
         assert len(catalog.tests) == len(catalog_data["tests"])
@@ -224,8 +246,8 @@ def test_from_dict(self, catalog_data: dict[str, Any]) -> None:
     @pytest.mark.parametrize("catalog_data", CATALOG_PARSE_FAIL_DATA, ids=generate_test_ids_list(CATALOG_PARSE_FAIL_DATA))
     def test_parse_fail(self, catalog_data: dict[str, Any]) -> None:
         """Errors when instantiating AntaCatalog from a file."""
-        with pytest.raises((ValidationError, TypeError)) as exec_info:
-            AntaCatalog.parse(DATA_DIR / catalog_data["filename"])
+        with pytest.raises((ValidationError, TypeError, ValueError, OSError)) as exec_info:
+            AntaCatalog.parse(DATA_DIR / catalog_data["filename"], file_format=catalog_data.get("file_format", "yaml"))
         if isinstance(exec_info.value, ValidationError):
             assert catalog_data["error"] in exec_info.value.errors()[0]["msg"]
         else: