Test models

Test definition

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	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

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.

    Parameters
    ----------
    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 == AntaTestStatus.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	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

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.

        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 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:  # 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")
        ```

    """

Command definition

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.

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

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__(
    self,
    template: str,
    version: Literal[1, "latest"] = "latest",
    revision: Revision | None = None,
    ofmt: Literal["json", "text"] = "json",
    *,
    use_cache: bool = True,
) -> None:
    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.

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

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