Test models

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:
                if command.params and ("src" and "dst") in command.params:
                    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: 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

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

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,
):
    """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

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

failed_commands property

failed_commands: list[AntaCommand]

Returns a list of all the commands that have failed.

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: 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	Optional[List[str]]	List of device’s tags for the test.

ResultOverwrite

Bases: BaseModel

Test inputs model to overwrite result fields

Attributes:

Name	Type	Description
description	Optional[str]	overwrite TestResult.description
categories	Optional[List[str]]	overwrite TestResult.categories
custom_field	Optional[str]	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]]

Decorator for the test() method.

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]]:
    """
    Decorator for the `test()` method.

    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: Any,
    ) -> TestResult:
        """
        Args:
            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.

        Returns:
            result: TestResult instance attribute populated with error status if any
        """

        def format_td(seconds: float, digits: int = 3) -> str:
            isec, fsec = divmod(round(seconds * 10**digits), 10**digits)
            return f"{timedelta(seconds=isec)}.{fsec:0{digits}.0f}"

        start_time = time.time()
        if self.result.result != "unset":
            return self.result

        # Data
        if eos_data is not None:
            self.save_commands_data(eos_data)
            self.logger.debug(f"Test {self.name} initialized with input data {eos_data}")

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

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

        test_duration = time.time() - start_time
        self.logger.debug(f"Executing test {self.name} on device {self.device.name} took {format_td(test_duration)}")

        AntaTest.update_progress()
        return self.result

    return wrapper

collect async

collect() -> None

Method used to 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:
    """
    Method used to 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)
    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."""
    raise NotImplementedError(f"AntaTemplate are provided but render() method has not been implemented for {self.__module__}.{self.name}")

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]

This abstract method is the core of the test logic. 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 reson")

Source code in anta/models.py

@abstractmethod
def test(self) -> Coroutine[Any, Any, TestResult]:
    """
    This abstract method is the core of the test logic.
    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 reson")
        ```
    """

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	str	Device command
version	Literal[1, 'latest']	eAPI version - valid values are 1 or “latest” - default is “latest”
revision	Optional[conint(ge=1, le=99)]	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 - default is json
output	Optional[Union[Dict[str, Any], str]]	Output of the command populated by the collect() function
template	Optional[AntaTemplate]	AntaTemplate object used to render this command
params	Dict[str, Any]	Dictionary of variables with string values to render the template
errors	List[str]	If the command execution fails, eAPI returns a list of strings detailing the error
use_cache	bool	Enable or disable caching for this AntaCommand if the AntaDevice supports it - default is True

collected property

collected: bool

Return True if the command has been collected

json_output property

json_output: dict[str, Any]

Get the command output as JSON

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

Bases: BaseModel

Class to define a command template as Python f-string. Can render a command from parameters.

Attributes:

Name	Type	Description
template	str	Python f-string. Example: ‘show vlan {vlan_id}’
version	Literal[1, 'latest']	eAPI version - valid values are 1 or “latest” - default is “latest”
revision	Optional[conint(ge=1, le=99)]	Revision of the command. Valid values are 1 to 99. Revision has precedence over version.
ofmt	Literal['json', 'text']	eAPI output - json or text - default is json
use_cache	bool	Enable or disable caching for this AntaTemplate if the AntaDevice supports it - default is True

render

render(**params: dict[str, Any]) -> AntaCommand

Render an AntaCommand from an AntaTemplate instance. Keep the parameters used in the AntaTemplate instance.

Parameters:

Name	Type	Description	Default
params	dict[str, Any]	dictionary of variables with string values to render the Python f-string	{}

Returns:

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

Source code in anta/models.py

def render(self, **params: dict[str, Any]) -> 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:
        command: The rendered AntaCommand.
                 This AntaCommand instance have a template attribute that references this
                 AntaTemplate instance.
    """
    try:
        return AntaCommand(
            command=self.template.format(**params),
            ofmt=self.ofmt,
            version=self.version,
            revision=self.revision,
            template=self,
            params=params,
            use_cache=self.use_cache,
        )
    except KeyError as e:
        raise AntaTemplateRenderError(self, e.args[0]) from e