Test models

Anta Test definition

UML Diagram

AntaTest

AntaTest(device: AntaDevice, template_params: list[dict[str, Any]] | None = None, result_description: str | None = None, result_categories: list[str] | None = None, result_custom_field: str | None = None, eos_data: list[dict[Any, Any] | str] | None = None, labels: list[str] | None = None)

Bases: ABC

Abstract class defining a test for Anta

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

TODO - complete doctstring with example

Doc to be completed

Parameters:

Name	Type	Description	Default
result_custom_field	str	a free string that is included in the TestResult object	None

Source code in anta/models.py

def __init__(
    self,
    device: AntaDevice,
    template_params: list[dict[str, Any]] | None = None,
    result_description: str | None = None,
    result_categories: list[str] | None = None,
    result_custom_field: str | None = None,
    # TODO document very well the order of eos_data
    eos_data: list[dict[Any, Any] | str] | None = None,
    labels: list[str] | None = None,
):
    """
    AntaTest Constructor

    Doc to be completed

    Arguments:
        result_custom_field (str): a free string that is included in the TestResult object
    """
    # Accept 6 input arguments
    # pylint: disable=R0913
    self.logger: logging.Logger = logging.getLogger(f"{self.__module__}.{self.__class__.__name__}")
    self.device: AntaDevice = device
    self.result: TestResult = TestResult(
        name=device.name,
        test=self.name,
        categories=result_categories or self.categories,
        description=result_description or self.description,
        custom_field=result_custom_field,
    )
    self.labels: List[str] = labels or []
    self.instance_commands: List[AntaCommand] = []

    # TODO - check optimization for deepcopy
    # Generating instance_commands from list of commands and template
    if hasattr(self.__class__, "commands") and (cmds := self.__class__.commands) is not None:
        self.instance_commands.extend(deepcopy(cmds))
    if hasattr(self.__class__, "template") and (tpl := self.__class__.template) is not None:
        if template_params is None:
            self.result.is_error("Command has template but no params were given")
            return
        self.template_params = template_params
        for param in template_params:
            try:
                self.instance_commands.append(tpl.render(param))
            except KeyError:
                self.result.is_error(f"Cannot render template '{tpl.template}': wrong parameters")
                return

    if eos_data is not None:
        self.logger.debug("Test initialized with input data")
        self.save_commands_data(eos_data)

all_data_collected

all_data_collected() -> bool

returns True if output is populated for every command

Source code in anta/models.py

def all_data_collected(self) -> bool:
    """returns True if output is populated for every command"""
    return all(command.collected for command in self.instance_commands)

anta_test staticmethod

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

Decorator for anta_test that handles injecting test data if given and collecting it using asyncio if missing

Source code in anta/models.py

@staticmethod
def anta_test(function: F) -> Callable[..., Coroutine[Any, Any, TestResult]]:
    """
    Decorator for anta_test that handles injecting test data if given and collecting it using asyncio if missing
    """

    @wraps(function)
    async def wrapper(
        self: AntaTest,
        eos_data: list[dict[Any, Any] | str] | None = None,
        **kwargs: Any,
    ) -> TestResult:
        """
        Wraps the test function and implement (in this order):
        1. Instantiate the command outputs if `eos_data` is provided
        2. Collect missing command outputs from the device
        3. Run the test function
        4. Catches and set the result if the test function raises an exception

        Returns:
            TestResult: self.result, populated with the correct exit status
        """

        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

        # TODO maybe_skip decorators

        # 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.all_data_collected():
            await self.collect()
            if self.result.result != "unset":
                return self.result

        try:
            if cmds := self.get_failed_commands():
                self.result.is_error(
                    "\n".join([f"{cmd.command} has failed: {exc_to_str(cmd.failed)}" if cmd.failed else f"{cmd.command} has failed" for cmd in cmds])
                )
                return self.result
            function(self, **kwargs)
        except Exception as e:  # pylint: disable=broad-exception-caught
            message = f"Exception raised for test {self.name} (on device {self.device.name})"
            anta_log_exception(e, message, self.logger)
            self.result.is_error(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:
        await self.device.collect_commands(self.instance_commands)
    except Exception as e:  # pylint: disable=broad-exception-caught
        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(exc_to_str(e))

get_failed_commands

get_failed_commands() -> List[AntaCommand]

returns a list of all the commands that have a populated failed field

Source code in anta/models.py

def get_failed_commands(self) -> List[AntaCommand]:
    """returns a list of all the commands that have a populated failed field"""
    return [command for command in self.instance_commands if command.failed is not None]

save_commands_data

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

Called at init or at test execution time

Source code in anta/models.py

def save_commands_data(self, eos_data: list[dict[Any, Any] | str]) -> None:
    """Called at init or at test execution time"""
    if len(eos_data) != len(self.instance_commands):
        self.result.is_error("Test initialization error: Trying to save more 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. It MUST set the correct status of self.result with the appropriate error messages

it must be implemented as follow

@AntaTest.anta_test def test(self) -> None: ‘’’ assert code ‘’‘

Source code in anta/models.py

@abstractmethod
def test(self) -> Coroutine[Any, Any, TestResult]:
    """
    This abstract method is the core of the test.
    It MUST set the correct status of self.result with the appropriate error messages

    it must be implemented as follow

    @AntaTest.anta_test
    def test(self) -> None:
       '''
       assert code
       '''
    """

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

Anta Command definition

UML Diagram

AntaCommand

Bases: BaseModel

Class to define a test command with its API 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)]	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
template	Optional[AntaTemplate]	AntaTemplate object used to render this command
params	Optional[Dict[str, Any]]	dictionary of variables with string values to render the template
failed	Optional[Exception]	If the command execution fails, the Exception object is stored in this field

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

Template command definition

UML Diagram

AntaTemplate

Bases: BaseModel

Class to define a test command with its API version

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

render

render(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: 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:
         AntaCommand: The rendered AntaCommand.
                      This AntaCommand instance have a template attribute that references this
                      AntaTemplate instance.
    """
    return AntaCommand(command=self.template.format(**params), ofmt=self.ofmt, version=self.version, revision=self.revision, template=self, params=params)

---

Last update: August 8, 2023