Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

feat(anta.cli): Add anta get tests #843

Draft
wants to merge 3 commits into
base: main
Choose a base branch
from
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions anta/cli/get/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -17,3 +17,4 @@ def get() -> None:
get.add_command(commands.from_ansible)
get.add_command(commands.inventory)
get.add_command(commands.tags)
get.add_command(commands.tests)
33 changes: 33 additions & 0 deletions anta/cli/get/commands.py
Original file line number Diff line number Diff line change
Expand Up @@ -132,3 +132,36 @@ def tags(inventory: AntaInventory, **kwargs: Any) -> None:
tags.update(device.tags)
console.print("Tags found:")
console.print_json(json.dumps(sorted(tags), indent=2))


@click.command
def tests() -> None:
"""Show all builtin ANTA tests with an example output retrieved from each test documentation."""
console.print("Current builtin ANTA tests are:", style="white on blue")
import importlib
import inspect
import pkgutil

from numpydoc.docscrape import NumpyDocString

from anta.models import AntaTest

# TODO: with this method need to disable warning for unknown section in numpydoc
# Would be better to not use numpydoc
def explore_package(module_name, level=2):
loader = pkgutil.get_loader(module_name)
path = Path(loader.get_filename()).parent
for sub_module in pkgutil.walk_packages([str(path)]):
_, sub_module_name, _ = sub_module
qname = module_name + "." + sub_module_name
if sub_module.ispkg:
explore_package(qname, level=3)
else:
console.print(f"{qname}:")
qname_module = importlib.import_module(qname)
for name, obj in inspect.getmembers(qname_module):
if inspect.isclass(obj) and issubclass(obj, AntaTest) and obj != AntaTest:
doc = NumpyDocString(obj.__doc__)
print("\n".join(l[2 * level :] for l in doc["Examples"][level:-1]))

explore_package("anta.tests")
10 changes: 9 additions & 1 deletion anta/models.py
Original file line number Diff line number Diff line change
Expand Up @@ -506,11 +506,19 @@ def save_commands_data(self, eos_data: list[dict[str, Any] | str]) -> None:

def __init_subclass__(cls) -> None:
"""Verify that the mandatory class attributes are defined."""
mandatory_attributes = ["name", "description", "categories", "commands"]
mandatory_attributes = ["categories", "commands"]
for attr in mandatory_attributes:
if not hasattr(cls, attr):
msg = f"Class {cls.__module__}.{cls.__name__} is missing required class attribute {attr}"
raise NotImplementedError(msg)
# default_attributes = ["name", "description"]
if not hasattr(cls, "name"):
cls.name = cls.__name__
if not hasattr(cls, "description"):
if not cls.test.__doc__:
# No doctsring - raise
raise Exception("TODO")
cls.description = cls.test.__doc__.split(sep="\n")[0]

@property
def module(self) -> str:
Expand Down
3 changes: 1 addition & 2 deletions anta/reporter/csv_reporter.py
Original file line number Diff line number Diff line change
Expand Up @@ -57,8 +57,7 @@ def split_list_to_txt_list(cls, usr_list: list[str], delimiter: str = " - ") ->

@classmethod
def convert_to_list(cls, result: TestResult) -> list[str]:
"""
Convert a TestResult into a list of string for creating file content.
"""Convert a TestResult into a list of string for creating file content.

Parameters
----------
Expand Down
4 changes: 0 additions & 4 deletions anta/tests/aaa.py
Original file line number Diff line number Diff line change
Expand Up @@ -35,8 +35,6 @@ class VerifyTacacsSourceIntf(AntaTest):
```
"""

name = "VerifyTacacsSourceIntf"
description = "Verifies TACACS source-interface for a specified VRF."
categories: ClassVar[list[str]] = ["aaa"]
commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaCommand(command="show tacacs", revision=1)]

Expand Down Expand Up @@ -81,8 +79,6 @@ class VerifyTacacsServers(AntaTest):
```
"""

name = "VerifyTacacsServers"
description = "Verifies TACACS servers are configured for a specified VRF."
categories: ClassVar[list[str]] = ["aaa"]
commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaCommand(command="show tacacs", revision=1)]

Expand Down
9 changes: 3 additions & 6 deletions anta/tests/avt.py
Original file line number Diff line number Diff line change
Expand Up @@ -18,8 +18,7 @@


class VerifyAVTPathHealth(AntaTest):
"""
Verifies the status of all Adaptive Virtual Topology (AVT) paths for all VRFs.
"""Verifies the status of all Adaptive Virtual Topology (AVT) paths for all VRFs.

Expected Results
----------------
Expand Down Expand Up @@ -73,8 +72,7 @@ def test(self) -> None:


class VerifyAVTSpecificPath(AntaTest):
"""
Verifies the status and type of an Adaptive Virtual Topology (AVT) path for a specified VRF.
"""Verifies the status and type of an Adaptive Virtual Topology (AVT) path for a specified VRF.

Expected Results
----------------
Expand Down Expand Up @@ -191,8 +189,7 @@ def test(self) -> None:


class VerifyAVTRole(AntaTest):
"""
Verifies the Adaptive Virtual Topology (AVT) role of a device.
"""Verifies the Adaptive Virtual Topology (AVT) role of a device.

Expected Results
----------------
Expand Down
9 changes: 3 additions & 6 deletions anta/tests/flow_tracking.py
Original file line number Diff line number Diff line change
Expand Up @@ -16,8 +16,7 @@


def validate_record_export(record_export: dict[str, str], tracker_info: dict[str, str]) -> str:
"""
Validate the record export configuration against the tracker info.
"""Validate the record export configuration against the tracker info.

Parameters
----------
Expand All @@ -40,8 +39,7 @@ def validate_record_export(record_export: dict[str, str], tracker_info: dict[str


def validate_exporters(exporters: list[dict[str, str]], tracker_info: dict[str, str]) -> str:
"""
Validate the exporter configurations against the tracker info.
"""Validate the exporter configurations against the tracker info.

Parameters
----------
Expand Down Expand Up @@ -73,8 +71,7 @@ def validate_exporters(exporters: list[dict[str, str]], tracker_info: dict[str,


class VerifyHardwareFlowTrackerStatus(AntaTest):
"""
Verifies if hardware flow tracking is running and an input tracker is active.
"""Verifies if hardware flow tracking is running and an input tracker is active.

This test optionally verifies the tracker interval/timeout and exporter configuration.

Expand Down
6 changes: 2 additions & 4 deletions anta/tests/path_selection.py
Original file line number Diff line number Diff line change
Expand Up @@ -18,8 +18,7 @@


class VerifyPathsHealth(AntaTest):
"""
Verifies the path and telemetry state of all paths under router path-selection.
"""Verifies the path and telemetry state of all paths under router path-selection.

The expected states are 'IPsec established', 'Resolved' for path and 'active' for telemetry.

Expand Down Expand Up @@ -73,8 +72,7 @@ def test(self) -> None:


class VerifySpecificPath(AntaTest):
"""
Verifies the path and telemetry state of a specific path for an IPv4 peer under router path-selection.
"""Verifies the path and telemetry state of a specific path for an IPv4 peer under router path-selection.

The expected states are 'IPsec established', 'Resolved' for path and 'active' for telemetry.

Expand Down
20 changes: 7 additions & 13 deletions anta/tests/routing/isis.py
Original file line number Diff line number Diff line change
Expand Up @@ -446,8 +446,7 @@ def test(self) -> None:


class VerifyISISSegmentRoutingDataplane(AntaTest):
"""
Verify dataplane of a list of ISIS-SR instances.
"""Verify dataplane of a list of ISIS-SR instances.

Expected Results
----------------
Expand Down Expand Up @@ -530,8 +529,7 @@ def test(self) -> None:


class VerifyISISSegmentRoutingTunnels(AntaTest):
"""
Verify ISIS-SR tunnels computed by device.
"""Verify ISIS-SR tunnels computed by device.

Expected Results
----------------
Expand All @@ -543,7 +541,7 @@ class VerifyISISSegmentRoutingTunnels(AntaTest):
--------
```yaml
anta.tests.routing:
isis:
isis:
- VerifyISISSegmentRoutingTunnels:
entries:
# Check only endpoint
Expand Down Expand Up @@ -638,8 +636,7 @@ def test(self) -> None:
self.result.is_failure("\n".join(failure_message))

def _check_tunnel_type(self, via_input: VerifyISISSegmentRoutingTunnels.Input.Entry.Vias, eos_entry: dict[str, Any]) -> bool:
"""
Check if the tunnel type specified in `via_input` matches any of the tunnel types in `eos_entry`.
"""Check if the tunnel type specified in `via_input` matches any of the tunnel types in `eos_entry`.

Parameters
----------
Expand All @@ -666,8 +663,7 @@ def _check_tunnel_type(self, via_input: VerifyISISSegmentRoutingTunnels.Input.En
return True

def _check_tunnel_nexthop(self, via_input: VerifyISISSegmentRoutingTunnels.Input.Entry.Vias, eos_entry: dict[str, Any]) -> bool:
"""
Check if the tunnel nexthop matches the given input.
"""Check if the tunnel nexthop matches the given input.

Parameters
----------
Expand All @@ -694,8 +690,7 @@ def _check_tunnel_nexthop(self, via_input: VerifyISISSegmentRoutingTunnels.Input
return True

def _check_tunnel_interface(self, via_input: VerifyISISSegmentRoutingTunnels.Input.Entry.Vias, eos_entry: dict[str, Any]) -> bool:
"""
Check if the tunnel interface exists in the given EOS entry.
"""Check if the tunnel interface exists in the given EOS entry.

Parameters
----------
Expand All @@ -722,8 +717,7 @@ def _check_tunnel_interface(self, via_input: VerifyISISSegmentRoutingTunnels.Inp
return True

def _check_tunnel_id(self, via_input: VerifyISISSegmentRoutingTunnels.Input.Entry.Vias, eos_entry: dict[str, Any]) -> bool:
"""
Check if the tunnel ID matches any of the tunnel IDs in the EOS entry's vias.
"""Check if the tunnel ID matches any of the tunnel IDs in the EOS entry's vias.

Parameters
----------
Expand Down
9 changes: 3 additions & 6 deletions anta/tests/security.py
Original file line number Diff line number Diff line change
Expand Up @@ -669,8 +669,7 @@ def test(self) -> None:


class VerifyIPSecConnHealth(AntaTest):
"""
Verifies all IPv4 security connections.
"""Verifies all IPv4 security connections.

Expected Results
----------------
Expand Down Expand Up @@ -716,8 +715,7 @@ def test(self) -> None:


class VerifySpecificIPSecConn(AntaTest):
"""
Verifies the state of IPv4 security connections for a specified peer.
"""Verifies the state of IPv4 security connections for a specified peer.

It optionally allows for the verification of a specific path for a peer by providing source and destination addresses.
If these addresses are not provided, it will verify all paths for the specified peer.
Expand Down Expand Up @@ -831,8 +829,7 @@ def test(self) -> None:


class VerifyHardwareEntropy(AntaTest):
"""
Verifies hardware entropy generation is enabled on device.
"""Verifies hardware entropy generation is enabled on device.

Expected Results
----------------
Expand Down
6 changes: 2 additions & 4 deletions anta/tests/stun.py
Original file line number Diff line number Diff line change
Expand Up @@ -18,8 +18,7 @@


class VerifyStunClient(AntaTest):
"""
Verifies the configuration of the STUN client, specifically the IPv4 source address and port.
"""Verifies the configuration of the STUN client, specifically the IPv4 source address and port.

Optionally, it can also verify the public address and port.

Expand Down Expand Up @@ -118,8 +117,7 @@ def test(self) -> None:


class VerifyStunServer(AntaTest):
"""
Verifies the STUN server status is enabled and running.
"""Verifies the STUN server status is enabled and running.

Expected Results
----------------
Expand Down
4 changes: 2 additions & 2 deletions anta/tests/system.py
Original file line number Diff line number Diff line change
Expand Up @@ -112,8 +112,8 @@ class VerifyCoredump(AntaTest):
* Success: The test will pass if there are NO core dump(s) in /var/core.
* Failure: The test will fail if there are core dump(s) in /var/core.

Info
----
Notes
-----
* This test will NOT check for minidump(s) generated by certain agents in /var/core/minidump.

Examples
Expand Down
4 changes: 2 additions & 2 deletions anta/tests/vxlan.py
Original file line number Diff line number Diff line change
Expand Up @@ -23,8 +23,8 @@
class VerifyVxlan1Interface(AntaTest):
"""Verifies if the Vxlan1 interface is configured and 'up/up'.

Warning
-------
Warnings
--------
The name of this test has been updated from 'VerifyVxlan' for better representation.

Expected Results
Expand Down
3 changes: 1 addition & 2 deletions anta/tools.py
Original file line number Diff line number Diff line change
Expand Up @@ -93,8 +93,7 @@ def get_dict_superset(
*,
required: bool = False,
) -> Any:
"""
Get the first dictionary from a list of dictionaries that is a superset of the input dict.
"""Get the first dictionary from a list of dictionaries that is a superset of the input dict.

Returns the supplied default value or None if there is no match and "required" is False.

Expand Down
3 changes: 1 addition & 2 deletions asynceapi/aio_portcheck.py
Original file line number Diff line number Diff line change
Expand Up @@ -34,8 +34,7 @@


async def port_check_url(url: URL, timeout: int = 5) -> bool:
"""
Open the port designated by the URL given the timeout in seconds.
"""Open the port designated by the URL given the timeout in seconds.

Parameters
----------
Expand Down
Loading
Loading