ANTA Test API

AntaTest

AntaTest(
    device: AntaDevice,
    inputs: dict[str, Any] | Input | None = None,
    eos_data: list[dict[Any, Any] | str] | None = None,
)

Bases: ABC

Abstract class defining a test in ANTA.

The goal of this class is to handle the heavy lifting and make writing a test as simple as possible.

Examples

The following is an example of an AntaTest subclass implementation:

    class VerifyReachability(AntaTest):
        '''Test the network reachability to one or many destination IP(s).'''
        categories = ["connectivity"]
        commands = [AntaTemplate(template="ping vrf {vrf} {dst} source {src} repeat 2")]

        class Input(AntaTest.Input):
            hosts: list[Host]
            class Host(BaseModel):
                dst: IPv4Address
                src: IPv4Address
                vrf: str = "default"

        def render(self, template: AntaTemplate) -> list[AntaCommand]:
            return [template.render(dst=host.dst, src=host.src, vrf=host.vrf) for host in self.inputs.hosts]

        @AntaTest.anta_test
        def test(self) -> None:
            failures = []
            for command in self.instance_commands:
                src, dst = command.params.src, command.params.dst
                if "2 received" not in command.json_output["messages"][0]:
                    failures.append((str(src), str(dst)))
            if not failures:
                self.result.is_success()
            else:
                self.result.is_failure(f"Connectivity test failed for the following source-destination pairs: {failures}")

Attributes:

Name	Type	Description
device	AntaDevice	AntaDevice instance on which this test is run.
inputs	Input	AntaTest.Input instance carrying the test inputs.
instance_commands	list[AntaCommand]	List of AntaCommand instances of this test.
result	TestResult	TestResult instance representing the result of this test.
logger	Logger	Python logger for this test instance.

Parameters:

Name	Type	Description	Default
device	AntaDevice	AntaDevice instance on which the test will be run.	required
inputs	dict[str, Any] | Input | None	Dictionary of attributes used to instantiate the AntaTest.Input instance.	None
eos_data	list[dict[Any, Any] | str] | None	Populate outputs of the test commands instead of collecting from devices. This list must have the same length and order than the instance_commands instance attribute.	None

blocked property

blocked: bool

Check if CLI commands contain a blocked keyword.

collected property

collected: bool

Return True if all commands for this test have been collected.

failed_commands property

failed_commands: list[AntaCommand]

Return a list of all the commands that have failed.

module property

module: str

Return the Python module in which this AntaTest class is defined.

Input

Bases: BaseModel

Class defining inputs for a test in ANTA.

Examples

A valid test catalog will look like the following:

<Python module>:
- <AntaTest subclass>:
    result_overwrite:
        categories:
        - "Overwritten category 1"
        description: "Test with overwritten description"
        custom_field: "Test run by John Doe"

Attributes:

Name	Type	Description
result_overwrite	ResultOverwrite | None	Define fields to overwrite in the TestResult object.

Filters

Bases: BaseModel

Runtime filters to map tests with list of tags or devices.

Attributes:

Name	Type	Description
tags	set[str] | None	Tag of devices on which to run the test.

ResultOverwrite

Bases: BaseModel

Test inputs model to overwrite result fields.

Attributes:

Name	Type	Description
description	str | None	Overwrite TestResult.description.
categories	list[str] | None	Overwrite TestResult.categories.
custom_field	str | None	A free string that will be included in the TestResult object.

anta_test staticmethod

anta_test(
    function: F,
) -> Callable[..., Coroutine[Any, Any, TestResult]]

Decorate the test() method in child classes.

This decorator implements (in this order):

1. Instantiate the command outputs if eos_data is provided to the test() method
2. Collect the commands from the device
3. Run the test() method
4. Catches any exception in test() user code and set the result instance attribute

Source code in anta/models.py

@staticmethod
def anta_test(function: F) -> Callable[..., Coroutine[Any, Any, TestResult]]:
    """Decorate the `test()` method in child classes.

    This decorator implements (in this order):

    1. Instantiate the command outputs if `eos_data` is provided to the `test()` method
    2. Collect the commands from the device
    3. Run the `test()` method
    4. Catches any exception in `test()` user code and set the `result` instance attribute
    """

    @wraps(function)
    async def wrapper(
        self: AntaTest,
        eos_data: list[dict[Any, Any] | str] | None = None,
        **kwargs: dict[str, Any],
    ) -> TestResult:
        """Inner function for the anta_test decorator.

        Parameters
        ----------
        self
            The test instance.
        eos_data
            Populate outputs of the test commands instead of collecting from devices.
            This list must have the same length and order than the `instance_commands` instance attribute.
        kwargs
            Any keyword argument to pass to the test.

        Returns
        -------
        TestResult
            The TestResult instance attribute populated with error status if any.

        """
        if self.result.result != "unset":
            return self.result

        # Data
        if eos_data is not None:
            self.save_commands_data(eos_data)
            self.logger.debug("Test %s initialized with input data %s", self.name, eos_data)

        # If some data is missing, try to collect
        if not self.collected:
            await self.collect()
            if self.result.result != "unset":
                AntaTest.update_progress()
                return self.result

            if self.failed_commands:
                self._handle_failed_commands()

                AntaTest.update_progress()
                return self.result

        try:
            function(self, **kwargs)
        except Exception as e:  # noqa: BLE001
            # test() is user-defined code.
            # We need to catch everything if we want the AntaTest object
            # to live until the reporting
            message = f"Exception raised for test {self.name} (on device {self.device.name})"
            anta_log_exception(e, message, self.logger)
            self.result.is_error(message=exc_to_str(e))

        # TODO: find a correct way to time test execution
        AntaTest.update_progress()
        return self.result

    return wrapper

collect async

collect() -> None

Collect outputs of all commands of this test class from the device of this test instance.

Source code in anta/models.py

async def collect(self) -> None:
    """Collect outputs of all commands of this test class from the device of this test instance."""
    try:
        if self.blocked is False:
            await self.device.collect_commands(self.instance_commands, collection_id=self.name)
    except Exception as e:  # noqa: BLE001
        # device._collect() is user-defined code.
        # We need to catch everything if we want the AntaTest object
        # to live until the reporting
        message = f"Exception raised while collecting commands for test {self.name} (on device {self.device.name})"
        anta_log_exception(e, message, self.logger)
        self.result.is_error(message=exc_to_str(e))

render

render(template: AntaTemplate) -> list[AntaCommand]

Render an AntaTemplate instance of this AntaTest using the provided AntaTest.Input instance at self.inputs.

This is not an abstract method because it does not need to be implemented if there is no AntaTemplate for this test.

Source code in anta/models.py

def render(self, template: AntaTemplate) -> list[AntaCommand]:
    """Render an AntaTemplate instance of this AntaTest using the provided AntaTest.Input instance at self.inputs.

    This is not an abstract method because it does not need to be implemented if there is
    no AntaTemplate for this test.
    """
    _ = template
    msg = f"AntaTemplate are provided but render() method has not been implemented for {self.module}.{self.__class__.__name__}"
    raise NotImplementedError(msg)

save_commands_data

save_commands_data(
    eos_data: list[dict[str, Any] | str],
) -> None

Populate output of all AntaCommand instances in instance_commands.

Source code in anta/models.py

def save_commands_data(self, eos_data: list[dict[str, Any] | str]) -> None:
    """Populate output of all AntaCommand instances in `instance_commands`."""
    if len(eos_data) > len(self.instance_commands):
        self.result.is_error(message="Test initialization error: Trying to save more data than there are commands for the test")
        return
    if len(eos_data) < len(self.instance_commands):
        self.result.is_error(message="Test initialization error: Trying to save less data than there are commands for the test")
        return
    for index, data in enumerate(eos_data or []):
        self.instance_commands[index].output = data

test abstractmethod

test() -> Coroutine[Any, Any, TestResult]

Core of the test logic.

This is an abstractmethod that must be implemented by child classes. It must set the correct status of the result instance attribute with the appropriate outcome of the test.

Examples

It must be implemented using the AntaTest.anta_test decorator:

@AntaTest.anta_test
def test(self) -> None:
    self.result.is_success()
    for command in self.instance_commands:
        if not self._test_command(command): # _test_command() is an arbitrary test logic
            self.result.is_failure("Failure reason")

Source code in anta/models.py

@abstractmethod
def test(self) -> Coroutine[Any, Any, TestResult]:
    """Core of the test logic.

    This is an abstractmethod that must be implemented by child classes.
    It must set the correct status of the `result` instance attribute with the appropriate outcome of the test.

    Examples
    --------
    It must be implemented using the `AntaTest.anta_test` decorator:
        ```python
        @AntaTest.anta_test
        def test(self) -> None:
            self.result.is_success()
            for command in self.instance_commands:
                if not self._test_command(command): # _test_command() is an arbitrary test logic
                    self.result.is_failure("Failure reason")
        ```

    """

update_progress classmethod

update_progress() -> None

Update progress bar for all AntaTest objects if it exists.

Source code in anta/models.py

@classmethod
def update_progress(cls: type[AntaTest]) -> None:
    """Update progress bar for all AntaTest objects if it exists."""
    if cls.progress and (cls.nrfu_task is not None):
        cls.progress.update(cls.nrfu_task, advance=1)