Test definition

UML Diagram

AntaTest

AntaTest(device: AntaDevice, inputs: dict[str, Any] | AntaTest.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):
        name = "VerifyReachability"
        description = "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 instance on which this test is run	inputs: AntaTest.Input instance carrying the test inputs instance_commands: List of AntaCommand instances of this test result: TestResult instance representing the result of this test logger: Python logger for this test instance

Args:

device: AntaDevice instance on which the test will be run
inputs: dictionary of attributes used to instantiate the AntaTest.Input 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.

Source code in anta/models.py

def __init__(
    self,
    device: AntaDevice,
    inputs: dict[str, Any] | AntaTest.Input | None = None,
    eos_data: list[dict[Any, Any] | str] | None = None,
) -> None:
    """AntaTest Constructor.

    Args:
    ----
        device: AntaDevice instance on which the test will be run
        inputs: dictionary of attributes used to instantiate the AntaTest.Input 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.
    """
    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] = []
    self.result: TestResult = TestResult(
        name=device.name,
        test=self.name,
        categories=self.categories,
        description=self.description,
    )
    self._init_inputs(inputs)
    if self.result.result == "unset":
        self._init_commands(eos_data)

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	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	Tag of devices on which to run the test.

ResultOverwrite

Bases: BaseModel

Test inputs model to overwrite result fields.

Attributes:

Name	Type	Description
description	overwrite TestResult.description	categories: overwrite TestResult.categories custom_field: a free string that will be included in the TestResult object

__hash__

__hash__() -> int

Implement generic hashing for AntaTest.Input.

This will work in most cases but this does not consider 2 lists with different ordering as equal.

Source code in anta/models.py

def __hash__(self) -> int:
    """Implement generic hashing for AntaTest.Input.

    This will work in most cases but this does not consider 2 lists with different ordering as equal.
    """
    return hash(self.model_dump_json())

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.

        Args:
        ----
            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
        -------
            result: 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 cmds := self.failed_commands:
                unsupported_commands = [f"'{c.command}' is not supported on {self.device.hw_model}" for c in cmds if not c.supported]
                if unsupported_commands:
                    msg = f"Test {self.name} has been skipped because it is not supported on {self.device.hw_model}: {GITHUB_SUGGESTION}"
                    self.logger.warning(msg)
                    self.result.is_skipped("\n".join(unsupported_commands))
                else:
                    self.result.is_error(message="\n".join([f"{c.command} has failed: {', '.join(c.errors)}" for c in cmds]))
                AntaTest.update_progress()
                return self.result

        try:
            function(self, **kwargs)
        except Exception as e:  # pylint: disable=broad-exception-caught
            # 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:  # pylint: disable=broad-exception-caught
        # 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")
        ```

    """

Command definition

UML Diagram

AntaCommand

Bases: BaseModel

Class to define a command.

Info

eAPI models are revisioned, this means that if a model is modified in a non-backwards compatible way, then its revision will be bumped up (revisions are numbers, default value is 1).

By default an eAPI request will return revision 1 of the model instance, this ensures that older management software will not suddenly stop working when a switch is upgraded. A revision applies to a particular CLI command whereas a version is global and is internally translated to a specific revision for each CLI command in the RPC.

Revision has precedence over version.

Attributes:

Name	Type	Description
command	Device command	version: eAPI version - valid values are 1 or “latest”. revision: eAPI revision of the command. Valid values are 1 to 99. Revision has precedence over version. ofmt: eAPI output - json or text. output: Output of the command. Only defined if there was no errors. template: AntaTemplate object used to render this command. errors: If the command execution fails, eAPI returns a list of strings detailing the error(s). params: Pydantic Model containing the variables values used to render the template. use_cache: Enable or disable caching for this AntaCommand if the AntaDevice supports it.

collected property

collected: bool

Return True if the command has been collected, False otherwise.

A command that has not been collected could have returned an error. See error property.

error property

error: bool

Return True if the command returned an error, False otherwise.

json_output property

json_output: dict[str, Any]

Get the command output as JSON.

requires_privileges property

requires_privileges: bool

Return True if the command requires privileged mode, False otherwise.

Raises:

Type	Description
RuntimeError	If the command has not been collected and has not returned an error. AntaDevice.collect() must be called before this property.

supported property

supported: bool

Return True if the command is supported on the device hardware platform, False otherwise.

Raises:

Type	Description
RuntimeError	If the command has not been collected and has not returned an error. AntaDevice.collect() must be called before this property.

text_output property

text_output: str

Get the command output as a string.

uid property

uid: str

Generate a unique identifier for this command.

Warning

CLI commands are protected to avoid execution of critical commands such as reload or write erase.

• Reload command: ^reload\s*\w*
• Configure mode: ^conf\w*\s*(terminal|session)*
• Write: ^wr\w*\s*\w+

Template definition

UML Diagram

AntaTemplate

AntaTemplate(template: str, version: Literal[1, 'latest'] = 'latest', revision: Revision | None = None, ofmt: Literal['json', 'text'] = 'json', *, use_cache: bool = True)

Class to define a command template as Python f-string.

Can render a command from parameters.

Attributes:

Name	Type	Description
template	Python f-string. Example: 'show vlan {vlan_id}'	version: eAPI version - valid values are 1 or “latest”. revision: Revision of the command. Valid values are 1 to 99. Revision has precedence over version. ofmt: eAPI output - json or text. use_cache: Enable or disable caching for this AntaTemplate if the AntaDevice supports it.

Source code in anta/models.py

def __init__(  # noqa: PLR0913
    self,
    template: str,
    version: Literal[1, "latest"] = "latest",
    revision: Revision | None = None,
    ofmt: Literal["json", "text"] = "json",
    *,
    use_cache: bool = True,
) -> None:
    # pylint: disable=too-many-arguments
    self.template = template
    self.version = version
    self.revision = revision
    self.ofmt = ofmt
    self.use_cache = use_cache

    # Create a AntaTemplateParams model to elegantly store AntaTemplate variables
    field_names = [fname for _, fname, _, _ in Formatter().parse(self.template) if fname]
    # Extracting the type from the params based on the expected field_names from the template
    fields: dict[str, Any] = {key: (Any, ...) for key in field_names}
    self.params_schema = create_model(
        "AntaParams",
        __base__=AntaParamsBaseModel,
        **fields,
    )

__repr__

__repr__() -> str

Return the representation of the class.

Copying pydantic model style, excluding params_schema

Source code in anta/models.py

def __repr__(self) -> str:
    """Return the representation of the class.

    Copying pydantic model style, excluding `params_schema`
    """
    return " ".join(f"{a}={v!r}" for a, v in vars(self).items() if a != "params_schema")

render

render(**params: str | int | bool) -> AntaCommand

Render an AntaCommand from an AntaTemplate instance.

Keep the parameters used in the AntaTemplate instance.

Args:

params: dictionary of variables with string values to render the Python f-string

Returns:

Type	Description
The rendered AntaCommand.	This AntaCommand instance have a template attribute that references this AntaTemplate instance.

Raises:

Type	Description
AntaTemplateRenderError	If a parameter is missing to render the AntaTemplate instance.

Source code in anta/models.py

def render(self, **params: str | int | bool) -> AntaCommand:
    """Render an AntaCommand from an AntaTemplate instance.

    Keep the parameters used in the AntaTemplate instance.

    Args:
    ----
        params: dictionary of variables with string values to render the Python f-string

    Returns
    -------
        The rendered AntaCommand.
        This AntaCommand instance have a template attribute that references this
        AntaTemplate instance.

    Raises
    ------
        AntaTemplateRenderError
            If a parameter is missing to render the AntaTemplate instance.
    """
    try:
        command = self.template.format(**params)
    except (KeyError, SyntaxError) as e:
        raise AntaTemplateRenderError(self, e.args[0]) from e
    return AntaCommand(
        command=command,
        ofmt=self.ofmt,
        version=self.version,
        revision=self.revision,
        template=self,
        params=self.params_schema(**params),
        use_cache=self.use_cache,
    )