Test Catalog

AntaCatalog

AntaCatalog(
    tests: list[AntaTestDefinition] | None = None,
    filename: str | Path | None = None,
)

Class representing an ANTA Catalog.

It can be instantiated using its constructor or one of the static methods: parse(), from_list() or from_dict()

Parameters:

Name	Type	Description	Default
tests	list[AntaTestDefinition] | None	A list of AntaTestDefinition instances.	None
filename	str | Path | None	The path from which the catalog is loaded.	None

Source code in anta/catalog.py

def __init__(
    self,
    tests: list[AntaTestDefinition] | None = None,
    filename: str | Path | None = None,
) -> None:
    """Instantiate an AntaCatalog instance.

    Parameters
    ----------
    tests
        A list of AntaTestDefinition instances.
    filename
        The path from which the catalog is loaded.

    """
    self._tests: list[AntaTestDefinition] = []
    if tests is not None:
        self._tests = tests
    self._filename: Path | None = None
    if filename is not None:
        if isinstance(filename, Path):
            self._filename = filename
        else:
            self._filename = Path(filename)

    self.indexes_built: bool
    self.tag_to_tests: defaultdict[str | None, set[AntaTestDefinition]]
    self._init_indexes()

filename property

filename: Path | None

Path of the file used to create this AntaCatalog instance.

tests property writable

tests: list[AntaTestDefinition]

List of AntaTestDefinition in this catalog.

build_indexes

build_indexes(
    filtered_tests: set[str] | None = None,
) -> None

Indexes tests by their tags for quick access during filtering operations.

If a filtered_tests set is provided, only the tests in this set will be indexed.

This method populates the tag_to_tests attribute, which is a dictionary mapping tags to sets of tests.

Once the indexes are built, the indexes_built attribute is set to True.

Source code in anta/catalog.py

def build_indexes(self, filtered_tests: set[str] | None = None) -> None:
    """Indexes tests by their tags for quick access during filtering operations.

    If a `filtered_tests` set is provided, only the tests in this set will be indexed.

    This method populates the tag_to_tests attribute, which is a dictionary mapping tags to sets of tests.

    Once the indexes are built, the `indexes_built` attribute is set to True.
    """
    for test in self.tests:
        # Skip tests that are not in the specified filtered_tests set
        if filtered_tests and test.test.name not in filtered_tests:
            continue

        # Indexing by tag
        if test.inputs.filters and (test_tags := test.inputs.filters.tags):
            for tag in test_tags:
                self.tag_to_tests[tag].add(test)
        else:
            self.tag_to_tests[None].add(test)

    self.indexes_built = True

clear_indexes

clear_indexes() -> None

Clear this AntaCatalog instance indexes.

Source code in anta/catalog.py

def clear_indexes(self) -> None:
    """Clear this AntaCatalog instance indexes."""
    self._init_indexes()

dump

dump() -> AntaCatalogFile

Return an AntaCatalogFile instance from this AntaCatalog instance.

Returns:

Type	Description
AntaCatalogFile	An AntaCatalogFile instance containing tests of this AntaCatalog instance.

Source code in anta/catalog.py

def dump(self) -> AntaCatalogFile:
    """Return an AntaCatalogFile instance from this AntaCatalog instance.

    Returns
    -------
    AntaCatalogFile
        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)

from_dict staticmethod

from_dict(
    data: RawCatalogInput,
    filename: str | Path | None = None,
) -> AntaCatalog

Create an AntaCatalog instance from a dictionary data structure.

See RawCatalogInput type alias for details. It is the data structure returned by yaml.load() function of a valid YAML Test Catalog file.

Parameters:

Name	Type	Description	Default
data	RawCatalogInput	Python dictionary used to instantiate the AntaCatalog instance.	required
filename	str | Path | None	value to be set as AntaCatalog instance attribute	None

Returns:

Type	Description
AntaCatalog	An AntaCatalog populated with the ‘data’ dictionary content.

Source code in anta/catalog.py

@staticmethod
def from_dict(data: RawCatalogInput, filename: str | Path | None = None) -> AntaCatalog:
    """Create an AntaCatalog instance from a dictionary data structure.

    See RawCatalogInput type alias for details.
    It is the data structure returned by `yaml.load()` function of a valid
    YAML Test Catalog file.

    Parameters
    ----------
    data
        Python dictionary used to instantiate the AntaCatalog instance.
    filename
        value to be set as AntaCatalog instance attribute

    Returns
    -------
    AntaCatalog
        An AntaCatalog populated with the 'data' dictionary content.
    """
    tests: list[AntaTestDefinition] = []
    if data is None:
        logger.warning("Catalog input data is empty")
        return AntaCatalog(filename=filename)

    if not isinstance(data, dict):
        msg = f"Wrong input type for catalog data{f' (from {filename})' if filename is not None else ''}, must be a dict, got {type(data).__name__}"
        raise TypeError(msg)

    try:
        catalog_data = AntaCatalogFile(data)  # type: ignore[arg-type]
    except ValidationError as e:
        anta_log_exception(
            e,
            f"Test catalog is invalid!{f' (from {filename})' if filename is not None else ''}",
            logger,
        )
        raise
    for t in catalog_data.root.values():
        tests.extend(t)
    return AntaCatalog(tests, filename=filename)

from_list staticmethod

from_list(data: ListAntaTestTuples) -> AntaCatalog

Create an AntaCatalog instance from a list data structure.

See ListAntaTestTuples type alias for details.

Parameters:

Name	Type	Description	Default
data	ListAntaTestTuples	Python list used to instantiate the AntaCatalog instance.	required

Returns:

Type	Description
AntaCatalog	An AntaCatalog populated with the ‘data’ list content.

Source code in anta/catalog.py

@staticmethod
def from_list(data: ListAntaTestTuples) -> AntaCatalog:
    """Create an AntaCatalog instance from a list data structure.

    See ListAntaTestTuples type alias for details.

    Parameters
    ----------
    data
        Python list used to instantiate the AntaCatalog instance.

    Returns
    -------
    AntaCatalog
        An AntaCatalog populated with the 'data' list content.
    """
    tests: list[AntaTestDefinition] = []
    try:
        tests.extend(AntaTestDefinition(test=test, inputs=inputs) for test, inputs in data)
    except ValidationError as e:
        anta_log_exception(e, "Test catalog is invalid!", logger)
        raise
    return AntaCatalog(tests)

get_tests_by_tags

get_tests_by_tags(
    tags: set[str], *, strict: bool = False
) -> set[AntaTestDefinition]

Return all tests that match a given set of tags, according to the specified strictness.

Parameters:

Name	Type	Description	Default
tags	set[str]	The tags to filter tests by. If empty, return all tests without tags.	required
strict	bool	If True, returns only tests that contain all specified tags (intersection). If False, returns tests that contain any of the specified tags (union).	False

Returns:

Type	Description
set[AntaTestDefinition]	A set of tests that match the given tags.

Raises:

Type	Description
ValueError	If the indexes have not been built prior to method call.

Source code in anta/catalog.py

def get_tests_by_tags(self, tags: set[str], *, strict: bool = False) -> set[AntaTestDefinition]:
    """Return all tests that match a given set of tags, according to the specified strictness.

    Parameters
    ----------
    tags
        The tags to filter tests by. If empty, return all tests without tags.
    strict
        If True, returns only tests that contain all specified tags (intersection).
        If False, returns tests that contain any of the specified tags (union).

    Returns
    -------
    set[AntaTestDefinition]
        A set of tests that match the given tags.

    Raises
    ------
    ValueError
        If the indexes have not been built prior to method call.
    """
    if not self.indexes_built:
        msg = "Indexes have not been built yet. Call build_indexes() first."
        raise ValueError(msg)
    if not tags:
        return self.tag_to_tests[None]

    filtered_sets = [self.tag_to_tests[tag] for tag in tags if tag in self.tag_to_tests]
    if not filtered_sets:
        return set()

    if strict:
        return set.intersection(*filtered_sets)
    return set.union(*filtered_sets)

merge

merge(catalog: AntaCatalog) -> AntaCatalog

Merge two AntaCatalog instances.

Warning

This method is deprecated and will be removed in ANTA v2.0. Use AntaCatalog.merge_catalogs() instead.

Parameters:

Name	Type	Description	Default
catalog	AntaCatalog	AntaCatalog instance to merge to this instance.	required

Returns:

Type	Description
AntaCatalog	A new AntaCatalog instance containing the tests of the two instances.

Source code in anta/catalog.py

def merge(self, catalog: AntaCatalog) -> AntaCatalog:
    """Merge two AntaCatalog instances.

    Warning
    -------
    This method is deprecated and will be removed in ANTA v2.0. Use `AntaCatalog.merge_catalogs()` instead.

    Parameters
    ----------
    catalog
        AntaCatalog instance to merge to this instance.

    Returns
    -------
    AntaCatalog
        A new AntaCatalog instance containing the tests of the two instances.
    """
    # TODO: Use a decorator to deprecate this method instead. See https://github.com/aristanetworks/anta/issues/754
    warn(
        message="AntaCatalog.merge() is deprecated and will be removed in ANTA v2.0. Use AntaCatalog.merge_catalogs() instead.",
        category=DeprecationWarning,
        stacklevel=2,
    )
    return self.merge_catalogs([self, catalog])

merge_catalogs classmethod

merge_catalogs(catalogs: list[AntaCatalog]) -> AntaCatalog

Merge multiple AntaCatalog instances.

Parameters:

Name	Type	Description	Default
catalogs	list[AntaCatalog]	A list of AntaCatalog instances to merge.	required

Returns:

Type	Description
AntaCatalog	A new AntaCatalog instance containing the tests of all the input catalogs.

Source code in anta/catalog.py

@classmethod
def merge_catalogs(cls, catalogs: list[AntaCatalog]) -> AntaCatalog:
    """Merge multiple AntaCatalog instances.

    Parameters
    ----------
    catalogs
        A list of AntaCatalog instances to merge.

    Returns
    -------
    AntaCatalog
        A new AntaCatalog instance containing the tests of all the input catalogs.
    """
    combined_tests = list(chain(*(catalog.tests for catalog in catalogs)))
    return cls(tests=combined_tests)

parse staticmethod

parse(
    filename: str | Path,
    file_format: Literal["yaml", "json"] = "yaml",
) -> AntaCatalog

Create an AntaCatalog instance from a test catalog file.

Parameters:

Name	Type	Description	Default
filename	str | Path	Path to test catalog YAML or JSON file.	required
file_format	Literal['yaml', 'json']	Format of the file, either ‘yaml’ or ‘json’.	'yaml'

Returns:

Type	Description
AntaCatalog	An AntaCatalog populated with the file content.

Source code in anta/catalog.py

@staticmethod
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 or JSON file.
    file_format
        Format of the file, either 'yaml' or 'json'.

    Returns
    -------
    AntaCatalog
        An AntaCatalog populated with the file content.
    """
    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) 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

    return AntaCatalog.from_dict(data, filename=filename)

AntaTestDefinition

AntaTestDefinition(
    **data: (
        type[AntaTest]
        | AntaTest.Input
        | dict[str, Any]
        | None
    )
)

Bases: BaseModel

Define a test with its associated inputs.

Attributes:

Name	Type	Description
test	type[AntaTest]	An AntaTest concrete subclass.
inputs	Input	The associated AntaTest.Input subclass instance.

https://docs.pydantic.dev/2.0/usage/validators/#using-validation-context-with-basemodel-initialization.

Source code in anta/catalog.py

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.

    https://docs.pydantic.dev/2.0/usage/validators/#using-validation-context-with-basemodel-initialization.
    """
    self.__pydantic_validator__.validate_python(
        data,
        self_instance=self,
        context={"test": data["test"]},
    )
    super(BaseModel, self).__init__()

check_inputs

check_inputs() -> Self

Check the inputs field typing.

The inputs class attribute needs to be an instance of the AntaTest.Input subclass defined in the class test.

Source code in anta/catalog.py

@model_validator(mode="after")
def check_inputs(self) -> Self:
    """Check the `inputs` field typing.

    The `inputs` class attribute needs to be an instance of the AntaTest.Input subclass defined in the class `test`.
    """
    if not isinstance(self.inputs, self.test.Input):
        msg = f"Test input has type {self.inputs.__class__.__qualname__} but expected type {self.test.Input.__qualname__}"
        raise ValueError(msg)  # noqa: TRY004 pydantic catches ValueError or AssertionError, no TypeError
    return self

instantiate_inputs classmethod

instantiate_inputs(
    data: AntaTest.Input | dict[str, Any] | None,
    info: ValidationInfo,
) -> AntaTest.Input

Ensure the test inputs can be instantiated and thus are valid.

If the test has no inputs, allow the user to omit providing the inputs field. If the test has inputs, allow the user to provide a valid dictionary of the input fields. This model validator will instantiate an Input class from the test class field.

Source code in anta/catalog.py

@field_validator("inputs", mode="before")
@classmethod
def instantiate_inputs(
    cls: type[AntaTestDefinition],
    data: AntaTest.Input | dict[str, Any] | None,
    info: ValidationInfo,
) -> AntaTest.Input:
    """Ensure the test inputs can be instantiated and thus are valid.

    If the test has no inputs, allow the user to omit providing the `inputs` field.
    If the test has inputs, allow the user to provide a valid dictionary of the input fields.
    This model validator will instantiate an Input class from the `test` class field.
    """
    if info.context is None:
        msg = "Could not validate inputs as no test class could be identified"
        raise ValueError(msg)
    # Pydantic guarantees at this stage that test_class is a subclass of AntaTest because of the ordering
    # of fields in the class definition - so no need to check for this
    test_class = info.context["test"]
    if not (isclass(test_class) and issubclass(test_class, AntaTest)):
        msg = f"Could not validate inputs as no test class {test_class} is not a subclass of AntaTest"
        raise ValueError(msg)

    if isinstance(data, AntaTest.Input):
        return data
    try:
        if data is None:
            return test_class.Input()
        if isinstance(data, dict):
            return test_class.Input(**data)
    except ValidationError as e:
        inputs_msg = str(e).replace("\n", "\n\t")
        err_type = "wrong_test_inputs"
        raise PydanticCustomError(
            err_type,
            f"{test_class.name} test inputs are not valid: {inputs_msg}\n",
            {"errors": e.errors()},
        ) from e
    msg = f"Could not instantiate inputs as type {type(data).__name__} is not valid"
    raise ValueError(msg)

serialize_model

serialize_model() -> 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:

Type	Description
dict	A dictionary representing the model.

Source code in anta/catalog.py

@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
    -------
    dict
        A dictionary representing the model.
    """
    return {self.test.__name__: self.inputs}

AntaCatalogFile

Bases: RootModel[dict[ImportString[Any], list[AntaTestDefinition]]]

Represents an ANTA Test Catalog File.

Example

A valid test catalog file must have the following structure:

<Python module>:
    - <AntaTest subclass>:
        <AntaTest.Input compliant dictionary>

check_tests classmethod

check_tests(data: Any) -> Any

Allow the user to provide a Python data structure that only has string values.

This validator will try to flatten and import Python modules, check if the tests classes are actually defined in their respective Python module and instantiate Input instances with provided value to validate test inputs.

Source code in anta/catalog.py

@model_validator(mode="before")
@classmethod
def check_tests(cls: type[AntaCatalogFile], data: Any) -> Any:  # noqa: ANN401
    """Allow the user to provide a Python data structure that only has string values.

    This validator will try to flatten and import Python modules, check if the tests classes
    are actually defined in their respective Python module and instantiate Input instances
    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
                if len(test_definition) != 1:
                    msg = (
                        f"Syntax error when parsing: {test_definition}\n"
                        "It must be a dictionary with a single entry. Check the indentation in the test catalog."
                    )
                    raise ValueError(msg)
                for test_name, test_inputs in test_definition.copy().items():
                    test: type[AntaTest] | None = getattr(module, test_name, None)
                    if test is None:
                        msg = (
                            f"{test_name} is not defined in Python module {module.__name__}"
                            f"{f' (from {module.__file__})' if module.__file__ is not None else ''}"
                        )
                        raise ValueError(msg)
                    test_definitions.append(AntaTestDefinition(test=test, inputs=test_inputs))
            typed_data[module] = test_definitions
        return typed_data
    return data

flatten_modules staticmethod

flatten_modules(
    data: dict[str, Any], package: str | None = None
) -> dict[ModuleType, list[Any]]

Allow the user to provide a data structure with nested Python modules.

Example

anta.tests.routing:
  generic:
    - <AntaTestDefinition>
  bgp:
    - <AntaTestDefinition>

anta.tests.routing.generic and anta.tests.routing.bgp are importable Python modules.

Source code in anta/catalog.py

@staticmethod
def flatten_modules(data: dict[str, Any], package: str | None = None) -> dict[ModuleType, list[Any]]:
    """Allow the user to provide a data structure with nested Python modules.

    Example
    -------
    ```
    anta.tests.routing:
      generic:
        - <AntaTestDefinition>
      bgp:
        - <AntaTestDefinition>
    ```
    `anta.tests.routing.generic` and `anta.tests.routing.bgp` are importable Python modules.

    """
    modules: dict[ModuleType, list[Any]] = {}
    for module_name, tests in data.items():
        if package and not module_name.startswith("."):
            # PLW2901 - we redefine the loop variable on purpose here.
            module_name = f".{module_name}"  # noqa: PLW2901
        try:
            module: ModuleType = importlib.import_module(name=module_name, package=package)
        except Exception as e:
            # A test module is potentially user-defined code.
            # We need to catch everything if we want to have meaningful logs
            module_str = f"{module_name[1:] if module_name.startswith('.') else module_name}{f' from package {package}' if package else ''}"
            message = f"Module named {module_str} cannot be imported. Verify that the module exists and there is no Python syntax issues."
            anta_log_exception(e, message, logger)
            raise ValueError(message) from e
        if isinstance(tests, dict):
            # This is an inner Python module
            modules.update(AntaCatalogFile.flatten_modules(data=tests, package=module.__name__))
        elif isinstance(tests, list):
            # This is a list of AntaTestDefinition
            modules[module] = tests
        else:
            msg = f"Syntax error when parsing: {tests}\nIt must be a list of ANTA tests. Check the test catalog."
            raise ValueError(msg)  # noqa: TRY004 pydantic catches ValueError or AssertionError, no TypeError
    return modules

to_json

to_json() -> str

Return a JSON representation string of this model.

Returns:

Type	Description
str	The JSON representation string of this model.

Source code in anta/catalog.py

def to_json(self) -> str:
    """Return a JSON representation string of this model.

    Returns
    -------
    str
        The JSON representation string of this model.
    """
    return self.model_dump_json(serialize_as_any=True, exclude_unset=True, indent=2)

yaml

yaml() -> str

Return a YAML representation string of this model.

Returns:

Type	Description
str	The YAML representation string of this model.

Source code in anta/catalog.py

def yaml(self) -> str:
    """Return a YAML representation string of this model.

    Returns
    -------
    str
        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 safe_dump(safe_load(self.model_dump_json(serialize_as_any=True, exclude_unset=True)), indent=2, width=math.inf)