feat(anta): add AntaCatalog.dump() and AntaCatalog.merge() (#657)

.pre-commit-config.yaml

@@ -10,6 +10,7 @@ repos:
       - id: trailing-whitespace
       - id: end-of-file-fixer
       - id: check-added-large-files
+        exclude: tests/data/.*$
       - id: check-merge-conflict
 
   - repo: https://github.com/Lucas-C/pre-commit-hooks

anta/catalog.py

@@ -7,12 +7,14 @@
 
 import importlib
 import logging
+import math
 from collections import defaultdict
 from inspect import isclass
 from pathlib import Path
 from typing import TYPE_CHECKING, Any, Optional, Union
 
-from pydantic import BaseModel, ConfigDict, RootModel, ValidationError, ValidationInfo, field_validator, model_validator
+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
@@ -44,6 +46,22 @@ class AntaTestDefinition(BaseModel):
     test: type[AntaTest]
     inputs: AntaTest.Input
 
+    @model_serializer()
+    def serialize_model(self) -> dict[str, AntaTest.Input]:
+        """Serialize the AntaTestDefinition model.
+
+        The dictionary representing the model will be look like:
+        ```
+        <AntaTest subclass name>:
+                <AntaTest.Input compliant dictionary>
+        ```
+
+        Returns
+        -------
+            A dictionary representing the model.
+        """
+        return {self.test.__name__: self.inputs}
+
     def __init__(self, **data: type[AntaTest] | AntaTest.Input | dict[str, Any] | None) -> None:
         """Inject test in the context to allow to instantiate Input in the BeforeValidator.
 
@@ -178,10 +196,15 @@ def check_tests(cls: type[AntaCatalogFile], data: Any) -> Any:  # noqa: ANN401
         with provided value to validate test inputs.
         """
         if isinstance(data, dict):
+            if not data:
+                return data
             typed_data: dict[ModuleType, list[Any]] = AntaCatalogFile.flatten_modules(data)
             for module, tests in typed_data.items():
                 test_definitions: list[AntaTestDefinition] = []
                 for test_definition in tests:
+                    if isinstance(test_definition, AntaTestDefinition):
+                        test_definitions.append(test_definition)
+                        continue
                     if not isinstance(test_definition, dict):
                         msg = f"Syntax error when parsing: {test_definition}\nIt must be a dictionary. Check the test catalog."
                         raise ValueError(msg)  # noqa: TRY004 pydantic catches ValueError or AssertionError, no TypeError
@@ -201,7 +224,21 @@ def check_tests(cls: type[AntaCatalogFile], data: Any) -> Any:  # noqa: ANN401
                             raise ValueError(msg)
                         test_definitions.append(AntaTestDefinition(test=test, inputs=test_inputs))
                 typed_data[module] = test_definitions
-        return typed_data
+            return typed_data
+        return data
+
+    def yaml(self) -> str:
+        """Return a YAML representation string of this model.
+
+        Returns
+        -------
+            The YAML representation string of this model.
+        """
+        # TODO: Pydantic and YAML serialization/deserialization is not supported natively.
+        # 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)
 
 
 class AntaCatalog:
@@ -304,7 +341,7 @@ def from_dict(data: RawCatalogInput, filename: str | Path | None = None) -> Anta
             raise TypeError(msg)
 
         try:
-            catalog_data = AntaCatalogFile(**data)  # type: ignore[arg-type]
+            catalog_data = AntaCatalogFile(data)  # type: ignore[arg-type]
         except ValidationError as e:
             anta_log_exception(
                 e,
@@ -335,6 +372,32 @@ def from_list(data: ListAntaTestTuples) -> AntaCatalog:
             raise
         return AntaCatalog(tests)
 
+    def merge(self, catalog: AntaCatalog) -> AntaCatalog:
+        """Merge two AntaCatalog instances.
+
+        Args:
+        ----
+            catalog: AntaCatalog instance to merge to this instance.
+
+        Returns
+        -------
+            A new AntaCatalog instance containing the tests of the two instances.
+        """
+        return AntaCatalog(tests=self.tests + catalog.tests)
+
+    def dump(self) -> AntaCatalogFile:
+        """Return an AntaCatalogFile instance from this AntaCatalog instance.
+
+        Returns
+        -------
+            An AntaCatalogFile instance containing tests of this AntaCatalog instance.
+        """
+        root: dict[ImportString[Any], list[AntaTestDefinition]] = {}
+        for test in self.tests:
+            # Cannot use AntaTest.module property as the class is not instantiated
+            root.setdefault(test.test.__module__, []).append(test)
+        return AntaCatalogFile(root=root)
+
     def build_indexes(self, filtered_tests: set[str] | None = None) -> None:
         """Indexes tests by their tags for quick access during filtering operations.
 

anta/models.py

@@ -380,7 +380,7 @@ def __init__(
                       This list must have the same length and order than the `instance_commands` instance attribute.
 
         """
-        self.logger: logging.Logger = logging.getLogger(f"{self.__module__}.{self.__class__.__name__}")
+        self.logger: logging.Logger = logging.getLogger(f"{self.module}.{self.__class__.__name__}")
         self.device: AntaDevice = device
         self.inputs: AntaTest.Input
         self.instance_commands: list[AntaCommand] = []
@@ -409,7 +409,7 @@ def _init_inputs(self, inputs: dict[str, Any] | AntaTest.Input | None) -> None:
             elif isinstance(inputs, dict):
                 self.inputs = self.Input(**inputs)
         except ValidationError as e:
-            message = f"{self.__module__}.{self.__class__.__name__}: Inputs are not valid\n{e}"
+            message = f"{self.module}.{self.name}: Inputs are not valid\n{e}"
             self.logger.error(message)
             self.result.is_error(message=message)
             return
@@ -446,7 +446,7 @@ def _init_commands(self, eos_data: list[dict[Any, Any] | str] | None) -> None:
                         # render() is user-defined code.
                         # We need to catch everything if we want the AntaTest object
                         # to live until the reporting
-                        message = f"Exception in {self.__module__}.{self.__class__.__name__}.render()"
+                        message = f"Exception in {self.module}.{self.__class__.__name__}.render()"
                         anta_log_exception(e, message, self.logger)
                         self.result.is_error(message=f"{message}: {exc_to_str(e)}")
                         return
@@ -474,14 +474,19 @@ def __init_subclass__(cls) -> None:
                 msg = f"Class {cls.__module__}.{cls.__name__} is missing required class attribute {attr}"
                 raise NotImplementedError(msg)
 
+    @property
+    def module(self) -> str:
+        """Return the Python module in which this AntaTest class is defined."""
+        return self.__module__
+
     @property
     def collected(self) -> bool:
-        """Returns True if all commands for this test have been collected."""
+        """Return True if all commands for this test have been collected."""
         return all(command.collected for command in self.instance_commands)
 
     @property
     def failed_commands(self) -> list[AntaCommand]:
-        """Returns a list of all the commands that have failed."""
+        """Return a list of all the commands that have failed."""
         return [command for command in self.instance_commands if command.error]
 
     def render(self, template: AntaTemplate) -> list[AntaCommand]:
@@ -491,7 +496,7 @@ def render(self, template: AntaTemplate) -> list[AntaCommand]:
         no AntaTemplate for this test.
         """
         _ = template
-        msg = f"AntaTemplate are provided but render() method has not been implemented for {self.__module__}.{self.name}"
+        msg = f"AntaTemplate are provided but render() method has not been implemented for {self.module}.{self.__class__.__name__}"
         raise NotImplementedError(msg)
 
     @property

anta/runner.py

@@ -179,7 +179,7 @@ def get_coroutines(selected_tests: defaultdict[AntaDevice, set[AntaTestDefinitio
                 # We need to catch everything and exit gracefully with an error message.
                 message = "\n".join(
                     [
-                        f"There is an error when creating test {test.test.__module__}.{test.test.__name__}.",
+                        f"There is an error when creating test {test.test.module}.{test.test.__name__}.",
                         f"If this is not a custom test implementation: {GITHUB_SUGGESTION}",
                     ],
                 )

docs/usage-inventory-catalog.md

@@ -244,3 +244,32 @@ Once you run `anta nrfu table`, you will see following output:
 │ spine01   │ VerifyInterfaceUtilization │ success     │            │ Verifies interfaces utilization is below 75%. │ interfaces    │
 └───────────┴────────────────────────────┴─────────────┴────────────┴───────────────────────────────────────────────┴───────────────┘
 ```
+
+### Example script to merge catalogs
+
+The following script reads all the files in `intended/test_catalogs/` with names `<device_name>-catalog.yml` and merge them together inside one big catalog `anta-catalog.yml`.
+
+```python
+#!/usr/bin/env python
+from anta.catalog import AntaCatalog
+
+from pathlib import Path
+from anta.models import AntaTest
+
+
+CATALOG_SUFFIX = '-catalog.yml'
+CATALOG_DIR = 'intended/test_catalogs/'
+
+if __name__ == "__main__":
+    catalog = AntaCatalog()
+    for file in Path(CATALOG_DIR).glob('*'+CATALOG_SUFFIX):
+        c = AntaCatalog.parse(file)
+        device = str(file).removesuffix(CATALOG_SUFFIX).removeprefix(CATALOG_DIR)
+        print(f"Merging test catalog for device {device}")
+        # Apply filters to all tests for this device
+        for test in c.tests:
+            test.inputs.filters = AntaTest.Input.Filters(tags=[device])
+        catalog.merge(c)
+    with open(Path('anta-catalog.yml'), "w") as f:
+        f.write(catalog.dump().yaml())
+```

tests/conftest.py

@@ -35,7 +35,7 @@ def build_test_id(val: dict[str, Any]) -> str:
         ...
     }
     """
-    return f"{val['test'].__module__}.{val['test'].__name__}-{val['name']}"
+    return f"{val['test'].module}.{val['test'].__name__}-{val['name']}"
 
 
 def pytest_generate_tests(metafunc: pytest.Metafunc) -> None: