diff --git a/anta/catalog.py b/anta/catalog.py index e94b8f750..30bd34066 100644 --- a/anta/catalog.py +++ b/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 diff --git a/anta/cli/nrfu/__init__.py b/anta/cli/nrfu/__init__.py index 27f8588e7..381f6c1bc 100644 --- a/anta/cli/nrfu/__init__.py +++ b/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 diff --git a/anta/cli/utils.py b/anta/cli/utils.py index f769de7a7..6d31e55ae 100644 --- a/anta/cli/utils.py +++ b/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) diff --git a/docs/cli/check.md b/docs/cli/check.md index d7dea620b..257ac73d8 100644 --- a/docs/cli/check.md +++ b/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. ``` diff --git a/docs/snippets/anta_nrfu_help.txt b/docs/snippets/anta_nrfu_help.txt index 68cb4b8cb..0717daa9e 100644 --- a/docs/snippets/anta_nrfu_help.txt +++ b/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 diff --git a/docs/usage-inventory-catalog.md b/docs/usage-inventory-catalog.md index f46993366..fd6aec320 100644 --- a/docs/usage-inventory-catalog.md +++ b/docs/usage-inventory-catalog.md @@ -78,7 +78,7 @@ 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 --- : @@ -86,6 +86,16 @@ A valid test catalog file must have the following structure: ``` +```json +{ + "": [ + { + "": + } + ] +} +``` + ### 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 diff --git a/tests/data/test_catalog.json b/tests/data/test_catalog.json new file mode 100644 index 000000000..298fcb4f4 --- /dev/null +++ b/tests/data/test_catalog.json @@ -0,0 +1,11 @@ +{ + "anta.tests.software": [ + { + "VerifyEOSVersion": { + "versions": [ + "4.31.1F" + ] + } + } + ] +} diff --git a/tests/data/test_catalog_invalid_json.json b/tests/data/test_catalog_invalid_json.json new file mode 100644 index 000000000..65b8c5bec --- /dev/null +++ b/tests/data/test_catalog_invalid_json.json @@ -0,0 +1 @@ +{aasas"anta.tests.software":[{"VerifyEOSVersion":{"versions":["4.31.1F"]}}]} diff --git a/tests/units/cli/nrfu/test__init__.py b/tests/units/cli/nrfu/test__init__.py index a9dcd9cb8..83369f344 100644 --- a/tests/units/cli/nrfu/test__init__.py +++ b/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() diff --git a/tests/units/test_catalog.py b/tests/units/test_catalog.py index 1c7ca8a0b..76358dd4a 100644 --- a/tests/units/test_catalog.py +++ b/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: