Skip to content

Commit

Permalink
doc: create ANTA UML Class Diagram (#1004)
Browse files Browse the repository at this point in the history
* doc: create ANTA UML Class Diagram

* doc: add class diagram

* doc: regorg api/tests

* doc: regorg reporters

* doc: titles

* doc: update Command API

* doc: update class diagram page

* doc: Update result API doc

* doc: update API doc

* doc: fix tests api navigation

* test: remove testing for deprecation warning...

* doc: address comments
  • Loading branch information
mtache authored Jan 29, 2025
1 parent 08f4a3d commit d402ed1
Show file tree
Hide file tree
Showing 73 changed files with 495 additions and 349 deletions.
27 changes: 23 additions & 4 deletions anta/constants.py
Original file line number Diff line number Diff line change
Expand Up @@ -44,12 +44,31 @@
r"IS-IS (.*) is disabled because: .*",
r"No source interface .*",
]
"""List of known EOS errors that should set a test status to 'failure' with the error message."""
"""List of known EOS errors.
!!! failure "Generic EOS Error Handling"
When catching these errors, **ANTA will fail the affected test** and reported the error message.
"""

EOS_BLACKLIST_CMDS = [
r"^reload.*",
r"^conf.*",
r"^wr.*",
]
"""List of blacklisted EOS commands.
!!! success "Disruptive commands safeguard"
ANTA implements a mechanism to **prevent the execution of disruptive commands** such as `reload`, `write erase` or `configure terminal`.
"""

UNSUPPORTED_PLATFORM_ERRORS = [
"not supported on this hardware platform",
"Invalid input (at token 2: 'trident')",
]
"""Error messages indicating platform or hardware unsupported commands.
Will set the test status to 'skipped'. Includes both general hardware
platform errors and specific ASIC family limitations."""
"""Error messages indicating platform or hardware unsupported commands. Includes both general hardware
platform errors and specific ASIC family limitations.
!!! tip "Running EOS commands unsupported by hardware"
When catching these errors, ANTA will skip the affected test and raise a warning. The **test catalog must be updated** to remove execution of the affected test
on unsupported devices.
"""
3 changes: 0 additions & 3 deletions anta/custom_types.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,9 +10,6 @@
from pydantic.functional_validators import AfterValidator, BeforeValidator

# Regular Expression definition
# TODO: make this configurable - with an env var maybe?
REGEXP_EOS_BLACKLIST_CMDS = [r"^reload.*", r"^conf\w*\s*(terminal|session)*", r"^wr\w*\s*\w+"]
"""List of regular expressions to blacklist from eos commands."""
REGEXP_PATH_MARKERS = r"[\\\/\s]"
"""Match directory path from string."""
REGEXP_INTERFACE_ID = r"\d+(\/\d+)*(\.\d+)?"
Expand Down
10 changes: 5 additions & 5 deletions anta/models.py
Original file line number Diff line number Diff line change
Expand Up @@ -15,8 +15,8 @@

from pydantic import BaseModel, ConfigDict, ValidationError, create_model

from anta.constants import KNOWN_EOS_ERRORS, UNSUPPORTED_PLATFORM_ERRORS
from anta.custom_types import REGEXP_EOS_BLACKLIST_CMDS, Revision
from anta.constants import EOS_BLACKLIST_CMDS, KNOWN_EOS_ERRORS, UNSUPPORTED_PLATFORM_ERRORS
from anta.custom_types import Revision
from anta.logger import anta_log_exception, exc_to_str
from anta.result_manager.models import AntaTestStatus, TestResult

Expand Down Expand Up @@ -432,7 +432,7 @@ def __init__(
inputs: dict[str, Any] | AntaTest.Input | None = None,
eos_data: list[dict[Any, Any] | str] | None = None,
) -> None:
"""AntaTest Constructor.
"""Initialize an AntaTest instance.
Parameters
----------
Expand Down Expand Up @@ -575,12 +575,12 @@ def blocked(self) -> bool:
"""Check if CLI commands contain a blocked keyword."""
state = False
for command in self.instance_commands:
for pattern in REGEXP_EOS_BLACKLIST_CMDS:
for pattern in EOS_BLACKLIST_CMDS:
if re.match(pattern, command.command):
self.logger.error(
"Command <%s> is blocked for security reason matching %s",
command.command,
REGEXP_EOS_BLACKLIST_CMDS,
EOS_BLACKLIST_CMDS,
)
self.result.is_error(f"<{command.command}> is blocked for security reason")
state = True
Expand Down
125 changes: 39 additions & 86 deletions anta/result_manager/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,12 +7,13 @@

import json
import logging
import warnings
from collections import defaultdict
from functools import cached_property
from itertools import chain
from typing import Any

from typing_extensions import deprecated

from anta.result_manager.models import AntaTestStatus, TestResult

from .models import CategoryStats, DeviceStats, TestStats
Expand All @@ -22,56 +23,40 @@

# pylint: disable=too-many-instance-attributes
class ResultManager:
"""Helper to manage Test Results and generate reports.
Examples
--------
Create Inventory:
inventory_anta = AntaInventory.parse(
filename='examples/inventory.yml',
username='ansible',
password='ansible',
)
Create Result Manager:
manager = ResultManager()
Run tests for all connected devices:
for device in inventory_anta.get_inventory().devices:
manager.add(
VerifyNTP(device=device).test()
)
manager.add(
VerifyEOSVersion(device=device).test(version='4.28.3M')
)
Print result in native format:
manager.results
[
TestResult(
name="pf1",
test="VerifyZeroTouch",
categories=["configuration"],
description="Verifies ZeroTouch is disabled",
result="success",
messages=[],
custom_field=None,
),
TestResult(
name="pf1",
test='VerifyNTP',
categories=["software"],
categories=['system'],
description='Verifies if NTP is synchronised.',
result='failure',
messages=["The device is not synchronized with the configured NTP server(s): 'NTP is disabled.'"],
custom_field=None,
),
]
"""Manager of ANTA Results.
The status of the class is initialized to "unset"
Then when adding a test with a status that is NOT 'error' the following
table shows the updated status:
| Current Status | Added test Status | Updated Status |
| -------------- | ------------------------------- | -------------- |
| unset | Any | Any |
| skipped | unset, skipped | skipped |
| skipped | success | success |
| skipped | failure | failure |
| success | unset, skipped, success | success |
| success | failure | failure |
| failure | unset, skipped success, failure | failure |
If the status of the added test is error, the status is untouched and the
`error_status` attribute is set to True.
Attributes
----------
results
dump
status
Status rerpesenting all the results.
error_status
Will be `True` if a test returned an error.
results_by_status
dump
json
device_stats
category_stats
test_stats
"""

_result_entries: list[TestResult]
Expand All @@ -84,26 +69,7 @@ class ResultManager:
_stats_in_sync: bool

def __init__(self) -> None:
"""Class constructor.
The status of the class is initialized to "unset"
Then when adding a test with a status that is NOT 'error' the following
table shows the updated status:
| Current Status | Added test Status | Updated Status |
| -------------- | ------------------------------- | -------------- |
| unset | Any | Any |
| skipped | unset, skipped | skipped |
| skipped | success | success |
| skipped | failure | failure |
| success | unset, skipped, success | success |
| success | failure | failure |
| failure | unset, skipped success, failure | failure |
If the status of the added test is error, the status is untouched and the
error_status is set to True.
"""
"""Initialize a ResultManager instance."""
self.reset()

def reset(self) -> None:
Expand Down Expand Up @@ -162,22 +128,9 @@ def test_stats(self) -> dict[str, TestStats]:
return dict(sorted(self._test_stats.items()))

@property
@deprecated("This property is deprecated, use `category_stats` instead. This will be removed in ANTA v2.0.0.", category=DeprecationWarning)
def sorted_category_stats(self) -> dict[str, CategoryStats]:
"""A property that returns the category_stats dictionary sorted by key name.
Deprecated
----------
This property is deprecated and will be removed in ANTA v2.0.0.
Use `category_stats` instead as it is now sorted by default.
TODO: Remove this property in ANTA v2.0.0.
"""
warnings.warn(
"sorted_category_stats is deprecated and will be removed in ANTA v2.0.0. Use category_stats instead as it is now sorted by default.",
DeprecationWarning,
stacklevel=2,
)
self._ensure_stats_in_sync()
"""A property that returns the category_stats dictionary sorted by key name."""
return self.category_stats

@cached_property
Expand Down
2 changes: 1 addition & 1 deletion anta/runner.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
# Copyright (c) 2023-2025 Arista Networks, Inc.
# Use of this source code is governed by the Apache License 2.0
# that can be found in the LICENSE file.
"""ANTA runner function."""
"""ANTA runner module."""

from __future__ import annotations

Expand Down
6 changes: 3 additions & 3 deletions docs/advanced_usages/as-python-lib.md
Original file line number Diff line number Diff line change
Expand Up @@ -14,8 +14,8 @@ ANTA is a Python library that can be used in user applications. This section des
A device is represented in ANTA as a instance of a subclass of the [AntaDevice](../api/device.md#anta.device.AntaDevice) abstract class.
There are few abstract methods that needs to be implemented by child classes:

- The [collect()](../api/device.md#anta.device.AntaDevice.collect) coroutine is in charge of collecting outputs of [AntaCommand](../api/models.md#anta.models.AntaCommand) instances.
- The [refresh()](../api/device.md#anta.device.AntaDevice.refresh) coroutine is in charge of updating attributes of the [AntaDevice](../api/device.md#anta.device.AntaDevice) instance. These attributes are used by [AntaInventory](../api/inventory.md#anta.inventory.AntaInventory) to filter out unreachable devices or by [AntaTest](../api/models.md#anta.models.AntaTest) to skip devices based on their hardware models.
- The [collect()](../api/device.md#anta.device.AntaDevice.collect) coroutine is in charge of collecting outputs of [AntaCommand](../api/commands.md#anta.models.AntaCommand) instances.
- The [refresh()](../api/device.md#anta.device.AntaDevice.refresh) coroutine is in charge of updating attributes of the [AntaDevice](../api/device.md#anta.device.AntaDevice) instance. These attributes are used by [AntaInventory](../api/inventory.md#anta.inventory.AntaInventory) to filter out unreachable devices or by [AntaTest](../api/tests/anta_test.md#anta.models.AntaTest) to skip devices based on their hardware models.

The [copy()](../api/device.md#anta.device.AntaDevice.copy) coroutine is used to copy files to and from the device. It does not need to be implemented if tests are not using it.

Expand All @@ -24,7 +24,7 @@ The [copy()](../api/device.md#anta.device.AntaDevice.copy) coroutine is used to
The [AsyncEOSDevice](../api/device.md#anta.device.AsyncEOSDevice) class is an implementation of [AntaDevice](../api/device.md#anta.device.AntaDevice) for Arista EOS.
It uses the [aio-eapi](https://github.com/jeremyschulman/aio-eapi) eAPI client and the [AsyncSSH](https://github.com/ronf/asyncssh) library.

- The [_collect()](../api/device.md#anta.device.AsyncEOSDevice._collect) coroutine collects [AntaCommand](../api/models.md#anta.models.AntaCommand) outputs using eAPI.
- The [_collect()](../api/device.md#anta.device.AsyncEOSDevice._collect) coroutine collects [AntaCommand](../api/commands.md#anta.models.AntaCommand) outputs using eAPI.
- The [refresh()](../api/device.md#anta.device.AsyncEOSDevice.refresh) coroutine tries to open a TCP connection on the eAPI port and update the `is_online` attribute accordingly. If the TCP connection succeeds, it sends a `show version` command to gather the hardware model of the device and updates the `established` and `hw_model` attributes.
- The [copy()](../api/device.md#anta.device.AsyncEOSDevice.copy) coroutine copies files to and from the device using the SCP protocol.

Expand Down
4 changes: 2 additions & 2 deletions docs/advanced_usages/caching.md
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,7 @@ The cache is initialized per `AntaDevice` and uses the following cache key desig

`<device_name>:<uid>`

The `uid` is an attribute of [AntaCommand](../api/models.md#anta.models.AntaCommand), which is a unique identifier generated from the command, version, revision and output format.
The `uid` is an attribute of [AntaCommand](../api/commands.md#anta.models.AntaCommand), which is a unique identifier generated from the command, version, revision and output format.

Each UID has its own asyncio lock. This design allows coroutines that need to access the cache for different UIDs to do so concurrently. The locks are managed by the `AntaCache.locks` dictionary.

Expand Down Expand Up @@ -62,7 +62,7 @@ There might be scenarios where caching is not wanted. You can disable caching in
This approach effectively disables caching for **ALL** commands sent to devices targeted by the `disable_cache` key.

3. For tests developers, caching can be disabled for a specific [`AntaCommand`](../api/models.md#anta.models.AntaCommand) or [`AntaTemplate`](../api/models.md#anta.models.AntaTemplate) by setting the `use_cache` attribute to `False`. That means the command output will always be collected on the device and therefore, never use caching.
3. For tests developers, caching can be disabled for a specific [`AntaCommand`](../api/commands.md#anta.models.AntaCommand) or [`AntaTemplate`](../api/commands.md#anta.models.AntaTemplate) by setting the `use_cache` attribute to `False`. That means the command output will always be collected on the device and therefore, never use caching.

### Disable caching in a child class of `AntaDevice`

Expand Down
Loading

0 comments on commit d402ed1

Please sign in to comment.