Skip to content

Test models

Test definition

AntaTest UML model

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):
        '''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

__removal_in_version class-attribute

__removal_in_version: str

Internal class variable set by the deprecated_test_class decorator.

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.

__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
391
392
393
394
395
396
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
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
@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
589
590
591
592
593
594
595
596
597
598
599
600
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
563
564
565
566
567
568
569
570
571
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
522
523
524
525
526
527
528
529
530
531
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
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
@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

AntaCommand UML model

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 str

Device command.

version Literal[1, 'latest']

eAPI version - valid values are 1 or “latest”.

revision Revision | None

eAPI revision of the command. Valid values are 1 to 99. Revision has precedence over version.

ofmt Literal['json', 'text']

eAPI output - json or text.

output dict[str, Any] | str | None

Output of the command. Only defined if there was no errors.

template AntaTemplate | None

AntaTemplate object used to render this command.

errors list[str]

If the command execution fails, eAPI returns a list of strings detailing the error(s).

params AntaParamsBaseModel

Pydantic Model containing the variables values used to render the template.

use_cache bool

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.

returned_known_eos_error property

returned_known_eos_error: bool

Return True if the command returned a known_eos_error on the device, False otherwise.

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

Indicates if the command is supported on the device.

Returns:

Type Description
bool

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

AntaTemplate UML model

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.

__repr__

__repr__() -> str

Return the representation of the class.

Copying pydantic model style, excluding params_schema

Source code in anta/models.py
91
92
93
94
95
96
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.

Parameters:

Name Type Description Default
params str | int | bool

Dictionary of variables with string values to render the Python f-string.

{}

Returns:

Type Description
AntaCommand

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
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
def render(self, **params: str | int | bool) -> AntaCommand:
    """Render an AntaCommand from an AntaTemplate instance.

    Keep the parameters used in the AntaTemplate instance.

    Parameters
    ----------
    params
        Dictionary of variables with string values to render the Python f-string.

    Returns
    -------
    AntaCommand
        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,
    )