From 5f325803d5c25ada43a62a81fbff7c474a497395 Mon Sep 17 00:00:00 2001 From: "pre-commit-ci[bot]" <66853113+pre-commit-ci[bot]@users.noreply.github.com> Date: Tue, 3 Sep 2024 09:28:00 +0200 Subject: [PATCH 1/8] ci: pre-commit autoupdate (#812) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit [pre-commit.ci] pre-commit autoupdate updates: - [github.com/pycqa/pylint: v3.2.6 → v3.2.7](https://github.com/pycqa/pylint/compare/v3.2.6...v3.2.7) Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com> --- .pre-commit-config.yaml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index fbf8c3fd2..440265441 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -52,7 +52,7 @@ repos: name: Run Ruff formatter - repo: https://github.com/pycqa/pylint - rev: "v3.2.6" + rev: "v3.2.7" hooks: - id: pylint name: Check code style with pylint From e7da54a43d9bd47177205dadc0d14e7894f4fc45 Mon Sep 17 00:00:00 2001 From: vitthalmagadum <122079046+vitthalmagadum@users.noreply.github.com> Date: Wed, 4 Sep 2024 06:11:21 +0530 Subject: [PATCH 2/8] feat(anta): Added the test case to verify NTP associations functionality (#757) --- anta/tests/system.py | 96 +++++++++++++- examples/tests.yaml | 9 ++ tests/units/anta_tests/test_system.py | 183 ++++++++++++++++++++++++++ 3 files changed, 287 insertions(+), 1 deletion(-) diff --git a/anta/tests/system.py b/anta/tests/system.py index 49d2dd25d..486e5e1ed 100644 --- a/anta/tests/system.py +++ b/anta/tests/system.py @@ -8,10 +8,14 @@ from __future__ import annotations import re +from ipaddress import IPv4Address from typing import TYPE_CHECKING, ClassVar -from anta.custom_types import PositiveInteger +from pydantic import BaseModel, Field + +from anta.custom_types import Hostname, PositiveInteger from anta.models import AntaCommand, AntaTest +from anta.tools import get_failed_logs, get_value if TYPE_CHECKING: from anta.models import AntaTemplate @@ -299,3 +303,93 @@ def test(self) -> None: else: data = command_output.split("\n")[0] self.result.is_failure(f"The device is not synchronized with the configured NTP server(s): '{data}'") + + +class VerifyNTPAssociations(AntaTest): + """Verifies the Network Time Protocol (NTP) associations. + + Expected Results + ---------------- + * Success: The test will pass if the Primary NTP server (marked as preferred) has the condition 'sys.peer' and + all other NTP servers have the condition 'candidate'. + * Failure: The test will fail if the Primary NTP server (marked as preferred) does not have the condition 'sys.peer' or + if any other NTP server does not have the condition 'candidate'. + + Examples + -------- + ```yaml + anta.tests.system: + - VerifyNTPAssociations: + ntp_servers: + - server_address: 1.1.1.1 + preferred: True + stratum: 1 + - server_address: 2.2.2.2 + stratum: 2 + - server_address: 3.3.3.3 + stratum: 2 + ``` + """ + + name = "VerifyNTPAssociations" + description = "Verifies the Network Time Protocol (NTP) associations." + categories: ClassVar[list[str]] = ["system"] + commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaCommand(command="show ntp associations")] + + class Input(AntaTest.Input): + """Input model for the VerifyNTPAssociations test.""" + + ntp_servers: list[NTPServer] + """List of NTP servers.""" + + class NTPServer(BaseModel): + """Model for a NTP server.""" + + server_address: Hostname | IPv4Address + """The NTP server address as an IPv4 address or hostname. The NTP server name defined in the running configuration + of the device may change during DNS resolution, which is not handled in ANTA. Please provide the DNS-resolved server name. + For example, 'ntp.example.com' in the configuration might resolve to 'ntp3.example.com' in the device output.""" + preferred: bool = False + """Optional preferred for NTP server. If not provided, it defaults to `False`.""" + stratum: int = Field(ge=0, le=16) + """NTP stratum level (0 to 15) where 0 is the reference clock and 16 indicates unsynchronized. + Values should be between 0 and 15 for valid synchronization and 16 represents an out-of-sync state.""" + + @AntaTest.anta_test + def test(self) -> None: + """Main test function for VerifyNTPAssociations.""" + failures: str = "" + + if not (peer_details := get_value(self.instance_commands[0].json_output, "peers")): + self.result.is_failure("None of NTP peers are not configured.") + return + + # Iterate over each NTP server. + for ntp_server in self.inputs.ntp_servers: + server_address = str(ntp_server.server_address) + preferred = ntp_server.preferred + stratum = ntp_server.stratum + + # Check if NTP server details exists. + if (peer_detail := get_value(peer_details, server_address, separator="..")) is None: + failures += f"NTP peer {server_address} is not configured.\n" + continue + + # Collecting the expected NTP peer details. + expected_peer_details = {"condition": "candidate", "stratum": stratum} + if preferred: + expected_peer_details["condition"] = "sys.peer" + + # Collecting the actual NTP peer details. + actual_peer_details = {"condition": get_value(peer_detail, "condition"), "stratum": get_value(peer_detail, "stratumLevel")} + + # Collecting failures logs if any. + failure_logs = get_failed_logs(expected_peer_details, actual_peer_details) + if failure_logs: + failures += f"For NTP peer {server_address}:{failure_logs}\n" + + # Check if there are any failures. + if not failures: + self.result.is_success() + else: + self.result.is_failure(failures) diff --git a/examples/tests.yaml b/examples/tests.yaml index c5f87fae7..1ad8e28db 100644 --- a/examples/tests.yaml +++ b/examples/tests.yaml @@ -445,6 +445,15 @@ anta.tests.system: - VerifyMemoryUtilization: - VerifyFileSystemUtilization: - VerifyNTP: + - VerifyNTPAssociations: + ntp_servers: + - server_address: 1.1.1.1 + preferred: True + stratum: 1 + - server_address: 2.2.2.2 + stratum: 1 + - server_address: 3.3.3.3 + stratum: 1 anta.tests.vlan: - VerifyVlanInternalPolicy: diff --git a/tests/units/anta_tests/test_system.py b/tests/units/anta_tests/test_system.py index 6965461d6..54849b734 100644 --- a/tests/units/anta_tests/test_system.py +++ b/tests/units/anta_tests/test_system.py @@ -14,6 +14,7 @@ VerifyFileSystemUtilization, VerifyMemoryUtilization, VerifyNTP, + VerifyNTPAssociations, VerifyReloadCause, VerifyUptime, ) @@ -286,4 +287,186 @@ "inputs": None, "expected": {"result": "failure", "messages": ["The device is not synchronized with the configured NTP server(s): 'unsynchronised'"]}, }, + { + "name": "success", + "test": VerifyNTPAssociations, + "eos_data": [ + { + "peers": { + "1.1.1.1": { + "condition": "sys.peer", + "peerIpAddr": "1.1.1.1", + "stratumLevel": 1, + }, + "2.2.2.2": { + "condition": "candidate", + "peerIpAddr": "2.2.2.2", + "stratumLevel": 2, + }, + "3.3.3.3": { + "condition": "candidate", + "peerIpAddr": "3.3.3.3", + "stratumLevel": 2, + }, + } + } + ], + "inputs": { + "ntp_servers": [ + {"server_address": "1.1.1.1", "preferred": True, "stratum": 1}, + {"server_address": "2.2.2.2", "stratum": 2}, + {"server_address": "3.3.3.3", "stratum": 2}, + ] + }, + "expected": {"result": "success"}, + }, + { + "name": "success-pool-name", + "test": VerifyNTPAssociations, + "eos_data": [ + { + "peers": { + "1.ntp.networks.com": { + "condition": "sys.peer", + "peerIpAddr": "1.1.1.1", + "stratumLevel": 1, + }, + "2.ntp.networks.com": { + "condition": "candidate", + "peerIpAddr": "2.2.2.2", + "stratumLevel": 2, + }, + "3.ntp.networks.com": { + "condition": "candidate", + "peerIpAddr": "3.3.3.3", + "stratumLevel": 2, + }, + } + } + ], + "inputs": { + "ntp_servers": [ + {"server_address": "1.ntp.networks.com", "preferred": True, "stratum": 1}, + {"server_address": "2.ntp.networks.com", "stratum": 2}, + {"server_address": "3.ntp.networks.com", "stratum": 2}, + ] + }, + "expected": {"result": "success"}, + }, + { + "name": "failure", + "test": VerifyNTPAssociations, + "eos_data": [ + { + "peers": { + "1.1.1.1": { + "condition": "candidate", + "peerIpAddr": "1.1.1.1", + "stratumLevel": 2, + }, + "2.2.2.2": { + "condition": "sys.peer", + "peerIpAddr": "2.2.2.2", + "stratumLevel": 2, + }, + "3.3.3.3": { + "condition": "sys.peer", + "peerIpAddr": "3.3.3.3", + "stratumLevel": 3, + }, + } + } + ], + "inputs": { + "ntp_servers": [ + {"server_address": "1.1.1.1", "preferred": True, "stratum": 1}, + {"server_address": "2.2.2.2", "stratum": 2}, + {"server_address": "3.3.3.3", "stratum": 2}, + ] + }, + "expected": { + "result": "failure", + "messages": [ + "For NTP peer 1.1.1.1:\nExpected `sys.peer` as the condition, but found `candidate` instead.\nExpected `1` as the stratum, but found `2` instead.\n" + "For NTP peer 2.2.2.2:\nExpected `candidate` as the condition, but found `sys.peer` instead.\n" + "For NTP peer 3.3.3.3:\nExpected `candidate` as the condition, but found `sys.peer` instead.\nExpected `2` as the stratum, but found `3` instead." + ], + }, + }, + { + "name": "failure-no-peers", + "test": VerifyNTPAssociations, + "eos_data": [{"peers": {}}], + "inputs": { + "ntp_servers": [ + {"server_address": "1.1.1.1", "preferred": True, "stratum": 1}, + {"server_address": "2.2.2.2", "stratum": 1}, + {"server_address": "3.3.3.3", "stratum": 1}, + ] + }, + "expected": { + "result": "failure", + "messages": ["None of NTP peers are not configured."], + }, + }, + { + "name": "failure-one-peer-not-found", + "test": VerifyNTPAssociations, + "eos_data": [ + { + "peers": { + "1.1.1.1": { + "condition": "sys.peer", + "peerIpAddr": "1.1.1.1", + "stratumLevel": 1, + }, + "2.2.2.2": { + "condition": "candidate", + "peerIpAddr": "2.2.2.2", + "stratumLevel": 1, + }, + } + } + ], + "inputs": { + "ntp_servers": [ + {"server_address": "1.1.1.1", "preferred": True, "stratum": 1}, + {"server_address": "2.2.2.2", "stratum": 1}, + {"server_address": "3.3.3.3", "stratum": 1}, + ] + }, + "expected": { + "result": "failure", + "messages": ["NTP peer 3.3.3.3 is not configured."], + }, + }, + { + "name": "failure-with-two-peers-not-found", + "test": VerifyNTPAssociations, + "eos_data": [ + { + "peers": { + "1.1.1.1": { + "condition": "candidate", + "peerIpAddr": "1.1.1.1", + "stratumLevel": 1, + } + } + } + ], + "inputs": { + "ntp_servers": [ + {"server_address": "1.1.1.1", "preferred": True, "stratum": 1}, + {"server_address": "2.2.2.2", "stratum": 1}, + {"server_address": "3.3.3.3", "stratum": 1}, + ] + }, + "expected": { + "result": "failure", + "messages": [ + "For NTP peer 1.1.1.1:\nExpected `sys.peer` as the condition, but found `candidate` instead.\n" + "NTP peer 2.2.2.2 is not configured.\nNTP peer 3.3.3.3 is not configured." + ], + }, + }, ] From 522ae9d710c76bf9f4b5a0e5eff078867bee003a Mon Sep 17 00:00:00 2001 From: vitthalmagadum <122079046+vitthalmagadum@users.noreply.github.com> Date: Thu, 5 Sep 2024 02:04:14 +0530 Subject: [PATCH 3/8] feat(anta): Added strict check to VerifyBGPPeerMPCaps test to verify only mentioned multiprotocol capabilities of a BGP peer should be listed (#783) * issue_781 Added optionak check to verify multiprotocol capability * issue_781 handling review comments: added helper function for capabilities * issue_781 Handling review comments: updated strict check * issue_781: Handling review comments: updated docsting * issue_781 handling review comments : updated strict check in testcase --------- Co-authored-by: VitthalMagadum Co-authored-by: Carl Baillargeon --- anta/tests/routing/bgp.py | 20 ++- examples/tests.yaml | 1 + tests/units/anta_tests/routing/test_bgp.py | 146 +++++++++++++++++++++ 3 files changed, 164 insertions(+), 3 deletions(-) diff --git a/anta/tests/routing/bgp.py b/anta/tests/routing/bgp.py index 6a7002356..70d2a6fcb 100644 --- a/anta/tests/routing/bgp.py +++ b/anta/tests/routing/bgp.py @@ -685,6 +685,8 @@ def test(self) -> None: class VerifyBGPPeerMPCaps(AntaTest): """Verifies the multiprotocol capabilities of a BGP peer in a specified VRF. + Supports `strict: True` to verify that only the specified capabilities are configured, requiring an exact match. + Expected Results ---------------- * Success: The test will pass if the BGP peer's multiprotocol capabilities are advertised, received, and enabled in the specified VRF. @@ -699,6 +701,7 @@ class VerifyBGPPeerMPCaps(AntaTest): bgp_peers: - peer_address: 172.30.11.1 vrf: default + strict: False capabilities: - ipv4Unicast ``` @@ -722,6 +725,8 @@ class BgpPeer(BaseModel): """IPv4 address of a BGP peer.""" vrf: str = "default" """Optional VRF for BGP peer. If not provided, it defaults to `default`.""" + strict: bool = False + """If True, requires exact matching of provided capabilities. Defaults to False.""" capabilities: list[MultiProtocolCaps] """List of multiprotocol capabilities to be verified.""" @@ -730,14 +735,14 @@ def test(self) -> None: """Main test function for VerifyBGPPeerMPCaps.""" failures: dict[str, Any] = {"bgp_peers": {}} - # Iterate over each bgp peer + # Iterate over each bgp peer. for bgp_peer in self.inputs.bgp_peers: peer = str(bgp_peer.peer_address) vrf = bgp_peer.vrf capabilities = bgp_peer.capabilities failure: dict[str, dict[str, dict[str, Any]]] = {"bgp_peers": {peer: {vrf: {}}}} - # Check if BGP output exists + # Check if BGP output exists. if ( not (bgp_output := get_value(self.instance_commands[0].json_output, f"vrfs.{vrf}.peerList")) or (bgp_output := get_item(bgp_output, "peerAddress", peer)) is None @@ -746,8 +751,17 @@ def test(self) -> None: failures = deep_update(failures, failure) continue - # Check each capability + # Fetching the capabilities output. bgp_output = get_value(bgp_output, "neighborCapabilities.multiprotocolCaps") + + if bgp_peer.strict and sorted(capabilities) != sorted(bgp_output): + failure["bgp_peers"][peer][vrf] = { + "status": f"Expected only `{', '.join(capabilities)}` capabilities should be listed but found `{', '.join(bgp_output)}` instead." + } + failures = deep_update(failures, failure) + continue + + # Check each capability for capability in capabilities: capability_output = bgp_output.get(capability) diff --git a/examples/tests.yaml b/examples/tests.yaml index 1ad8e28db..f5a5ca46b 100644 --- a/examples/tests.yaml +++ b/examples/tests.yaml @@ -546,6 +546,7 @@ anta.tests.routing: bgp_peers: - peer_address: 172.30.11.1 vrf: default + strict: False capabilities: - ipv4Unicast - VerifyBGPPeerASNCap: diff --git a/tests/units/anta_tests/routing/test_bgp.py b/tests/units/anta_tests/routing/test_bgp.py index 47db8e60b..b76939bd5 100644 --- a/tests/units/anta_tests/routing/test_bgp.py +++ b/tests/units/anta_tests/routing/test_bgp.py @@ -2200,6 +2200,152 @@ ], }, }, + { + "name": "success-strict", + "test": VerifyBGPPeerMPCaps, + "eos_data": [ + { + "vrfs": { + "default": { + "peerList": [ + { + "peerAddress": "172.30.11.1", + "neighborCapabilities": { + "multiprotocolCaps": { + "ipv4Unicast": { + "advertised": True, + "received": True, + "enabled": True, + }, + "ipv4MplsLabels": { + "advertised": True, + "received": True, + "enabled": True, + }, + } + }, + } + ] + }, + "MGMT": { + "peerList": [ + { + "peerAddress": "172.30.11.10", + "neighborCapabilities": { + "multiprotocolCaps": { + "ipv4Unicast": { + "advertised": True, + "received": True, + "enabled": True, + }, + "ipv4MplsVpn": { + "advertised": True, + "received": True, + "enabled": True, + }, + } + }, + } + ] + }, + } + } + ], + "inputs": { + "bgp_peers": [ + { + "peer_address": "172.30.11.1", + "vrf": "default", + "strict": True, + "capabilities": ["Ipv4 Unicast", "ipv4 Mpls labels"], + }, + { + "peer_address": "172.30.11.10", + "vrf": "MGMT", + "strict": True, + "capabilities": ["ipv4 Unicast", "ipv4 MplsVpn"], + }, + ] + }, + "expected": {"result": "success"}, + }, + { + "name": "failure-srict", + "test": VerifyBGPPeerMPCaps, + "eos_data": [ + { + "vrfs": { + "default": { + "peerList": [ + { + "peerAddress": "172.30.11.1", + "neighborCapabilities": { + "multiprotocolCaps": { + "ipv4Unicast": { + "advertised": True, + "received": True, + "enabled": True, + }, + "ipv4MplsLabels": { + "advertised": True, + "received": True, + "enabled": True, + }, + } + }, + } + ] + }, + "MGMT": { + "peerList": [ + { + "peerAddress": "172.30.11.10", + "neighborCapabilities": { + "multiprotocolCaps": { + "ipv4Unicast": { + "advertised": True, + "received": True, + "enabled": True, + }, + "ipv4MplsVpn": { + "advertised": False, + "received": True, + "enabled": True, + }, + } + }, + } + ] + }, + } + } + ], + "inputs": { + "bgp_peers": [ + { + "peer_address": "172.30.11.1", + "vrf": "default", + "strict": True, + "capabilities": ["Ipv4 Unicast"], + }, + { + "peer_address": "172.30.11.10", + "vrf": "MGMT", + "strict": True, + "capabilities": ["ipv4MplsVpn", "L2vpnEVPN"], + }, + ] + }, + "expected": { + "result": "failure", + "messages": [ + "Following BGP peer multiprotocol capabilities are not found or not ok:\n{'bgp_peers': {'172.30.11.1': " + "{'default': {'status': 'Expected only `ipv4Unicast` capabilities should be listed but found `ipv4Unicast, ipv4MplsLabels` instead.'}}," + " '172.30.11.10': {'MGMT': {'status': 'Expected only `ipv4MplsVpn, l2VpnEvpn` capabilities should be listed but found `ipv4Unicast, " + "ipv4MplsVpn` instead.'}}}}" + ], + }, + }, { "name": "success", "test": VerifyBGPPeerASNCap, From abdaea152c515a1da8bdff17c6c753852566a443 Mon Sep 17 00:00:00 2001 From: vitthalmagadum <122079046+vitthalmagadum@users.noreply.github.com> Date: Thu, 5 Sep 2024 02:20:42 +0530 Subject: [PATCH 4/8] refactor(anta): Update VerifySnmpContact , VerifySnmpLocation tests to have a more human readable format for the test result failures messages (#806) --- anta/tests/snmp.py | 13 +++++++++++-- tests/units/anta_tests/test_snmp.py | 28 ++++++++++++++++++++++++++++ 2 files changed, 39 insertions(+), 2 deletions(-) diff --git a/anta/tests/snmp.py b/anta/tests/snmp.py index ac98bfd2f..c7329b6d7 100644 --- a/anta/tests/snmp.py +++ b/anta/tests/snmp.py @@ -11,6 +11,7 @@ from anta.custom_types import PositiveInteger from anta.models import AntaCommand, AntaTest +from anta.tools import get_value if TYPE_CHECKING: from anta.models import AntaTemplate @@ -183,8 +184,12 @@ class Input(AntaTest.Input): @AntaTest.anta_test def test(self) -> None: """Main test function for VerifySnmpLocation.""" - location = self.instance_commands[0].json_output["location"]["location"] + # Verifies the SNMP location is configured. + if not (location := get_value(self.instance_commands[0].json_output, "location.location")): + self.result.is_failure("SNMP location is not configured.") + return + # Verifies the expected SNMP location. if location != self.inputs.location: self.result.is_failure(f"Expected `{self.inputs.location}` as the location, but found `{location}` instead.") else: @@ -222,8 +227,12 @@ class Input(AntaTest.Input): @AntaTest.anta_test def test(self) -> None: """Main test function for VerifySnmpContact.""" - contact = self.instance_commands[0].json_output["contact"]["contact"] + # Verifies the SNMP contact is configured. + if not (contact := get_value(self.instance_commands[0].json_output, "contact.contact")): + self.result.is_failure("SNMP contact is not configured.") + return + # Verifies the expected SNMP contact. if contact != self.inputs.contact: self.result.is_failure(f"Expected `{self.inputs.contact}` as the contact, but found `{contact}` instead.") else: diff --git a/tests/units/anta_tests/test_snmp.py b/tests/units/anta_tests/test_snmp.py index b4d31521e..64c44382e 100644 --- a/tests/units/anta_tests/test_snmp.py +++ b/tests/units/anta_tests/test_snmp.py @@ -99,6 +99,20 @@ "messages": ["Expected `New York` as the location, but found `Europe` instead."], }, }, + { + "name": "failure-details-not-configured", + "test": VerifySnmpLocation, + "eos_data": [ + { + "location": {"location": ""}, + } + ], + "inputs": {"location": "New York"}, + "expected": { + "result": "failure", + "messages": ["SNMP location is not configured."], + }, + }, { "name": "success", "test": VerifySnmpContact, @@ -124,4 +138,18 @@ "messages": ["Expected `Bob@example.com` as the contact, but found `Jon@example.com` instead."], }, }, + { + "name": "failure-details-not-configured", + "test": VerifySnmpContact, + "eos_data": [ + { + "contact": {"contact": ""}, + } + ], + "inputs": {"contact": "Bob@example.com"}, + "expected": { + "result": "failure", + "messages": ["SNMP contact is not configured."], + }, + }, ] From 9f433ce562569799a06328f33d416191c84fcea4 Mon Sep 17 00:00:00 2001 From: "pre-commit-ci[bot]" <66853113+pre-commit-ci[bot]@users.noreply.github.com> Date: Tue, 10 Sep 2024 09:52:01 +0200 Subject: [PATCH 5/8] ci: pre-commit autoupdate (#824) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit [pre-commit.ci] pre-commit autoupdate updates: - [github.com/astral-sh/ruff-pre-commit: v0.6.3 → v0.6.4](https://github.com/astral-sh/ruff-pre-commit/compare/v0.6.3...v0.6.4) Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com> --- .pre-commit-config.yaml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 440265441..9da8faaad 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -43,7 +43,7 @@ repos: - '' - repo: https://github.com/astral-sh/ruff-pre-commit - rev: v0.6.3 + rev: v0.6.4 hooks: - id: ruff name: Run Ruff linter From e32821d612177e8506b1fdb0989d0213d9bb5c03 Mon Sep 17 00:00:00 2001 From: "dependabot[bot]" <49699333+dependabot[bot]@users.noreply.github.com> Date: Wed, 11 Sep 2024 15:20:11 +0200 Subject: [PATCH 6/8] doc: Make API documentation great again (#797) * chore: update griffe requirement from <1.0.0,>=0.46 to >=0.46,<2.0.0 Updates the requirements on [griffe](https://github.com/mkdocstrings/griffe) to permit the latest version. - [Release notes](https://github.com/mkdocstrings/griffe/releases) - [Changelog](https://github.com/mkdocstrings/griffe/blob/main/CHANGELOG.md) - [Commits](https://github.com/mkdocstrings/griffe/compare/0.46.0...1.1.0) --- updated-dependencies: - dependency-name: griffe dependency-type: direct:production ... Signed-off-by: dependabot[bot] * Doc: Fix documentation * Refactor: Address PR comments * Doc: Address more PR comments * chore: Add python handler min version for mkdocstring * Apply suggestions from code review * Update anta/models.py * Update anta/reporter/csv_reporter.py * doc: Adjust pyproject.toml as per ruff doc * doc: Reading the doc better * Fix doc * doc: Better ruff config * doc: Fix css issue * Update pyproject.toml --------- Signed-off-by: dependabot[bot] Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com> Co-authored-by: gmuloc Co-authored-by: Carl Baillargeon --- anta/catalog.py | 90 ++++++++---- anta/cli/get/utils.py | 27 ++-- anta/cli/nrfu/utils.py | 6 +- anta/cli/utils.py | 3 +- anta/custom_types.py | 14 +- anta/decorators.py | 24 ++-- anta/device.py | 135 ++++++++++++------ anta/inventory/__init__.py | 82 +++++++---- anta/inventory/models.py | 37 +++-- anta/logger.py | 15 +- anta/models.py | 123 ++++++++++------ anta/reporter/__init__.py | 61 +++++--- anta/reporter/csv_reporter.py | 40 ++++-- anta/reporter/md_reporter.py | 36 +++-- anta/result_manager/__init__.py | 121 +++++++++------- anta/result_manager/models.py | 39 +++-- anta/runner.py | 59 +++++--- anta/tests/flow_tracking.py | 24 ++-- anta/tests/logging.py | 9 +- anta/tests/routing/bgp.py | 87 ++++++----- anta/tests/routing/isis.py | 33 +++-- anta/tests/routing/ospf.py | 18 ++- anta/tools.py | 31 ++-- docs/advanced_usages/as-python-lib.md | 8 +- docs/advanced_usages/caching.md | 2 +- docs/api/device.md | 14 +- docs/api/models.md | 7 +- docs/api/result_manager.md | 2 - docs/api/result_manager_models.md | 2 - docs/cli/debug.md | 2 +- docs/cli/nrfu.md | 2 +- docs/getting-started.md | 2 +- docs/stylesheets/extra.material.css | 14 +- .../{anta_test.html => anta_test.html.jinja} | 10 +- .../material/{class.html => class.html.jinja} | 4 +- .../{docstring.html => docstring.html.jinja} | 0 mkdocs.yml | 4 +- pyproject.toml | 31 ++-- 38 files changed, 766 insertions(+), 452 deletions(-) rename docs/templates/python/material/{anta_test.html => anta_test.html.jinja} (96%) rename docs/templates/python/material/{class.html => class.html.jinja} (91%) rename docs/templates/python/material/{docstring.html => docstring.html.jinja} (100%) diff --git a/anta/catalog.py b/anta/catalog.py index 7ed4bc718..46b90d3e6 100644 --- a/anta/catalog.py +++ b/anta/catalog.py @@ -39,8 +39,12 @@ class AntaTestDefinition(BaseModel): """Define a test with its associated inputs. - test: An AntaTest concrete subclass - inputs: The associated AntaTest.Input subclass instance + Attributes + ---------- + test + An AntaTest concrete subclass. + inputs + The associated AntaTest.Input subclass instance. """ model_config = ConfigDict(frozen=True) @@ -60,6 +64,7 @@ def serialize_model(self) -> dict[str, AntaTest.Input]: Returns ------- + dict A dictionary representing the model. """ return {self.test.__name__: self.inputs} @@ -132,14 +137,14 @@ def check_inputs(self) -> AntaTestDefinition: class AntaCatalogFile(RootModel[dict[ImportString[Any], list[AntaTestDefinition]]]): # pylint: disable=too-few-public-methods """Represents an ANTA Test Catalog File. - Example: + Example ------- - A valid test catalog file must have the following structure: - ``` - : - - : - - ``` + A valid test catalog file must have the following structure: + ``` + : + - : + + ``` """ @@ -149,16 +154,16 @@ class AntaCatalogFile(RootModel[dict[ImportString[Any], list[AntaTestDefinition] def flatten_modules(data: dict[str, Any], package: str | None = None) -> dict[ModuleType, list[Any]]: """Allow the user to provide a data structure with nested Python modules. - Example: + Example ------- - ``` - anta.tests.routing: - generic: - - - bgp: - - - ``` - `anta.tests.routing.generic` and `anta.tests.routing.bgp` are importable Python modules. + ``` + anta.tests.routing: + generic: + - + bgp: + - + ``` + `anta.tests.routing.generic` and `anta.tests.routing.bgp` are importable Python modules. """ modules: dict[ModuleType, list[Any]] = {} @@ -234,6 +239,7 @@ def yaml(self) -> str: Returns ------- + str The YAML representation string of this model. """ # TODO: Pydantic and YAML serialization/deserialization is not supported natively. @@ -247,6 +253,7 @@ def to_json(self) -> str: Returns ------- + str The JSON representation string of this model. """ return self.model_dump_json(serialize_as_any=True, exclude_unset=True, indent=2) @@ -267,8 +274,10 @@ def __init__( Parameters ---------- - tests: A list of AntaTestDefinition instances. - filename: The path from which the catalog is loaded. + tests + A list of AntaTestDefinition instances. + filename + The path from which the catalog is loaded. """ self._tests: list[AntaTestDefinition] = [] @@ -314,8 +323,10 @@ def parse(filename: str | Path, file_format: Literal["yaml", "json"] = "yaml") - Parameters ---------- - filename: Path to test catalog YAML or JSON fil - file_format: Format of the file, either 'yaml' or 'json' + filename + Path to test catalog YAML or JSON file. + file_format + Format of the file, either 'yaml' or 'json'. """ if file_format not in ["yaml", "json"]: @@ -343,7 +354,8 @@ def from_dict(data: RawCatalogInput, filename: str | Path | None = None) -> Anta Parameters ---------- - data: Python dictionary used to instantiate the AntaCatalog instance + data + Python dictionary used to instantiate the AntaCatalog instance. filename: value to be set as AntaCatalog instance attribute """ @@ -377,7 +389,8 @@ def from_list(data: ListAntaTestTuples) -> AntaCatalog: Parameters ---------- - data: Python list used to instantiate the AntaCatalog instance + data + Python list used to instantiate the AntaCatalog instance. """ tests: list[AntaTestDefinition] = [] @@ -394,10 +407,12 @@ def merge_catalogs(cls, catalogs: list[AntaCatalog]) -> AntaCatalog: Parameters ---------- - catalogs: A list of AntaCatalog instances to merge. + catalogs + A list of AntaCatalog instances to merge. Returns ------- + AntaCatalog A new AntaCatalog instance containing the tests of all the input catalogs. """ combined_tests = list(chain(*(catalog.tests for catalog in catalogs))) @@ -406,12 +421,18 @@ def merge_catalogs(cls, catalogs: list[AntaCatalog]) -> AntaCatalog: def merge(self, catalog: AntaCatalog) -> AntaCatalog: """Merge two AntaCatalog instances. + Warning + ------- + This method is deprecated and will be removed in ANTA v2.0. Use `AntaCatalog.merge_catalogs()` instead. + Parameters ---------- - catalog: AntaCatalog instance to merge to this instance. + catalog + AntaCatalog instance to merge to this instance. Returns ------- + AntaCatalog A new AntaCatalog instance containing the tests of the two instances. """ # TODO: Use a decorator to deprecate this method instead. See https://github.com/aristanetworks/anta/issues/754 @@ -427,6 +448,7 @@ def dump(self) -> AntaCatalogFile: Returns ------- + AntaCatalogFile An AntaCatalogFile instance containing tests of this AntaCatalog instance. """ root: dict[ImportString[Any], list[AntaTestDefinition]] = {} @@ -441,7 +463,9 @@ def build_indexes(self, filtered_tests: set[str] | None = None) -> None: If a `filtered_tests` set is provided, only the tests in this set will be indexed. This method populates two attributes: + - tag_to_tests: A dictionary mapping each tag to a set of tests that contain it. + - tests_without_tags: A set of tests that do not have any tags. Once the indexes are built, the `indexes_built` attribute is set to True. @@ -466,17 +490,21 @@ def get_tests_by_tags(self, tags: set[str], *, strict: bool = False) -> set[Anta Parameters ---------- - tags: The tags to filter tests by. If empty, return all tests without tags. - strict: If True, returns only tests that contain all specified tags (intersection). - If False, returns tests that contain any of the specified tags (union). + tags + The tags to filter tests by. If empty, return all tests without tags. + strict + If True, returns only tests that contain all specified tags (intersection). + If False, returns tests that contain any of the specified tags (union). Returns ------- - set[AntaTestDefinition]: A set of tests that match the given tags. + set[AntaTestDefinition] + A set of tests that match the given tags. Raises ------ - ValueError: If the indexes have not been built prior to method call. + ValueError + If the indexes have not been built prior to method call. """ if not self.indexes_built: msg = "Indexes have not been built yet. Call build_indexes() first." diff --git a/anta/cli/get/utils.py b/anta/cli/get/utils.py index ba4d886d5..8f11676db 100644 --- a/anta/cli/get/utils.py +++ b/anta/cli/get/utils.py @@ -84,18 +84,24 @@ def get_cv_token(cvp_ip: str, cvp_username: str, cvp_password: str, *, verify_ce Parameters ---------- - cvp_ip: IP address of CloudVision. - cvp_username: Username to connect to CloudVision. - cvp_password: Password to connect to CloudVision. - verify_cert: Enable or disable certificate verification when connecting to CloudVision. + cvp_ip + IP address of CloudVision. + cvp_username + Username to connect to CloudVision. + cvp_password + Password to connect to CloudVision. + verify_cert + Enable or disable certificate verification when connecting to CloudVision. Returns ------- - token(str): The token to use in further API calls to CloudVision. + str + The token to use in further API calls to CloudVision. Raises ------ - requests.ssl.SSLError: If the certificate verification fails + requests.ssl.SSLError + If the certificate verification fails. """ # use CVP REST API to generate a token @@ -163,9 +169,12 @@ def create_inventory_from_ansible(inventory: Path, output: Path, ansible_group: Parameters ---------- - inventory: Ansible Inventory file to read - output: ANTA inventory file to generate. - ansible_group: Ansible group from where to extract data. + inventory + Ansible Inventory file to read. + output + ANTA inventory file to generate. + ansible_group + Ansible group from where to extract data. """ try: diff --git a/anta/cli/nrfu/utils.py b/anta/cli/nrfu/utils.py index 748578dec..947c08901 100644 --- a/anta/cli/nrfu/utils.py +++ b/anta/cli/nrfu/utils.py @@ -147,8 +147,10 @@ def save_markdown_report(ctx: click.Context, md_output: pathlib.Path) -> None: Parameters ---------- - ctx: Click context containing the result manager. - md_output: Path to save the markdown report. + ctx + Click context containing the result manager. + md_output + Path to save the markdown report. """ try: MDReportGenerator.generate(results=_get_result_manager(ctx), md_filename=md_output) diff --git a/anta/cli/utils.py b/anta/cli/utils.py index 2f6e7d302..19ffb113f 100644 --- a/anta/cli/utils.py +++ b/anta/cli/utils.py @@ -62,7 +62,8 @@ def exit_with_code(ctx: click.Context) -> None: Parameters ---------- - ctx: Click Context + ctx + Click Context. """ if ctx.obj.get("ignore_status"): diff --git a/anta/custom_types.py b/anta/custom_types.py index 322fa4aca..6747e7663 100644 --- a/anta/custom_types.py +++ b/anta/custom_types.py @@ -66,9 +66,9 @@ def interface_case_sensitivity(v: str) -> str: Examples -------- - - ethernet -> Ethernet - - vlan -> Vlan - - loopback -> Loopback + - ethernet -> Ethernet + - vlan -> Vlan + - loopback -> Loopback """ if isinstance(v, str) and v != "" and not v[0].isupper(): @@ -81,10 +81,10 @@ def bgp_multiprotocol_capabilities_abbreviations(value: str) -> str: Examples -------- - - IPv4 Unicast - - L2vpnEVPN - - ipv4 MPLS Labels - - ipv4Mplsvpn + - IPv4 Unicast + - L2vpnEVPN + - ipv4 MPLS Labels + - ipv4Mplsvpn """ patterns = { diff --git a/anta/decorators.py b/anta/decorators.py index c9f8b6d28..f5608ef26 100644 --- a/anta/decorators.py +++ b/anta/decorators.py @@ -22,11 +22,13 @@ def deprecated_test(new_tests: list[str] | None = None) -> Callable[[F], F]: Parameters ---------- - new_tests: A list of new test classes that should replace the deprecated test. + new_tests + A list of new test classes that should replace the deprecated test. Returns ------- - Callable[[F], F]: A decorator that can be used to wrap test functions. + Callable[[F], F] + A decorator that can be used to wrap test functions. """ @@ -35,11 +37,13 @@ def decorator(function: F) -> F: Parameters ---------- - function: The test function to be decorated. + function + The test function to be decorated. Returns ------- - F: The decorated function. + F + The decorated function. """ @@ -66,11 +70,13 @@ def skip_on_platforms(platforms: list[str]) -> Callable[[F], F]: Parameters ---------- - platforms: List of hardware models on which the test should be skipped. + platforms + List of hardware models on which the test should be skipped. Returns ------- - Callable[[F], F]: A decorator that can be used to wrap test functions. + Callable[[F], F] + A decorator that can be used to wrap test functions. """ @@ -79,11 +85,13 @@ def decorator(function: F) -> F: Parameters ---------- - function: The test function to be decorated. + function + The test function to be decorated. Returns ------- - F: The decorated function. + F + The decorated function. """ diff --git a/anta/device.py b/anta/device.py index 087f3b57b..74b81d91e 100644 --- a/anta/device.py +++ b/anta/device.py @@ -42,13 +42,20 @@ class AntaDevice(ABC): Attributes ---------- - name: Device name - is_online: True if the device IP is reachable and a port can be open. - established: True if remote command execution succeeds. - hw_model: Hardware model of the device. - tags: Tags for this device. - cache: In-memory cache from aiocache library for this device (None if cache is disabled). - cache_locks: Dictionary mapping keys to asyncio locks to guarantee exclusive access to the cache if not disabled. + name : str + Device name. + is_online : bool + True if the device IP is reachable and a port can be open. + established : bool + True if remote command execution succeeds. + hw_model : str + Hardware model of the device. + tags : set[str] + Tags for this device. + cache : Cache | None + In-memory cache from aiocache library for this device (None if cache is disabled). + cache_locks : dict + Dictionary mapping keys to asyncio locks to guarantee exclusive access to the cache if not disabled. """ @@ -57,9 +64,12 @@ def __init__(self, name: str, tags: set[str] | None = None, *, disable_cache: bo Parameters ---------- - name: Device name. - tags: Tags for this device. - disable_cache: Disable caching for all commands for this device. + name + Device name. + tags + Tags for this device. + disable_cache + Disable caching for all commands for this device. """ self.name: str = name @@ -132,8 +142,10 @@ async def _collect(self, command: AntaCommand, *, collection_id: str | None = No Parameters ---------- - command: The command to collect. - collection_id: An identifier used to build the eAPI request ID. + command + The command to collect. + collection_id + An identifier used to build the eAPI request ID. """ async def collect(self, command: AntaCommand, *, collection_id: str | None = None) -> None: @@ -149,8 +161,10 @@ async def collect(self, command: AntaCommand, *, collection_id: str | None = Non Parameters ---------- - command: The command to collect. - collection_id: An identifier used to build the eAPI request ID. + command + The command to collect. + collection_id + An identifier used to build the eAPI request ID. """ # Need to ignore pylint no-member as Cache is a proxy class and pylint is not smart enough # https://github.com/pylint-dev/pylint/issues/7258 @@ -172,8 +186,10 @@ async def collect_commands(self, commands: list[AntaCommand], *, collection_id: Parameters ---------- - commands: The commands to collect. - collection_id: An identifier used to build the eAPI request ID. + commands + The commands to collect. + collection_id + An identifier used to build the eAPI request ID. """ await asyncio.gather(*(self.collect(command=command, collection_id=collection_id) for command in commands)) @@ -182,9 +198,12 @@ async def refresh(self) -> None: """Update attributes of an AntaDevice instance. This coroutine must update the following attributes of AntaDevice: - - `is_online`: When the device IP is reachable and a port can be open - - `established`: When a command execution succeeds - - `hw_model`: The hardware model of the device + + - `is_online`: When the device IP is reachable and a port can be open. + + - `established`: When a command execution succeeds. + + - `hw_model`: The hardware model of the device. """ async def copy(self, sources: list[Path], destination: Path, direction: Literal["to", "from"] = "from") -> None: @@ -194,9 +213,12 @@ async def copy(self, sources: list[Path], destination: Path, direction: Literal[ Parameters ---------- - sources: List of files to copy to or from the device. - destination: Local or remote destination when copying the files. Can be a folder. - direction: Defines if this coroutine copies files to or from the device. + sources + List of files to copy to or from the device. + destination + Local or remote destination when copying the files. Can be a folder. + direction + Defines if this coroutine copies files to or from the device. """ _ = (sources, destination, direction) @@ -209,11 +231,16 @@ class AsyncEOSDevice(AntaDevice): Attributes ---------- - name: Device name - is_online: True if the device IP is reachable and a port can be open - established: True if remote command execution succeeds - hw_model: Hardware model of the device - tags: Tags for this device + name : str + Device name. + is_online : bool + True if the device IP is reachable and a port can be open. + established : bool + True if remote command execution succeeds. + hw_model : str + Hardware model of the device. + tags : set[str] + Tags for this device. """ @@ -239,19 +266,32 @@ def __init__( Parameters ---------- - host: Device FQDN or IP. - username: Username to connect to eAPI and SSH. - password: Password to connect to eAPI and SSH. - name: Device name. - enable: Collect commands using privileged mode. - enable_password: Password used to gain privileged access on EOS. - port: eAPI port. Defaults to 80 is proto is 'http' or 443 if proto is 'https'. - ssh_port: SSH port. - tags: Tags for this device. - timeout: Timeout value in seconds for outgoing API calls. - insecure: Disable SSH Host Key validation. - proto: eAPI protocol. Value can be 'http' or 'https'. - disable_cache: Disable caching for all commands for this device. + host + Device FQDN or IP. + username + Username to connect to eAPI and SSH. + password + Password to connect to eAPI and SSH. + name + Device name. + enable + Collect commands using privileged mode. + enable_password + Password used to gain privileged access on EOS. + port + eAPI port. Defaults to 80 is proto is 'http' or 443 if proto is 'https'. + ssh_port + SSH port. + tags + Tags for this device. + timeout + Timeout value in seconds for outgoing API calls. + insecure + Disable SSH Host Key validation. + proto + eAPI protocol. Value can be 'http' or 'https'. + disable_cache + Disable caching for all commands for this device. """ if host is None: @@ -315,8 +355,10 @@ async def _collect(self, command: AntaCommand, *, collection_id: str | None = No Parameters ---------- - command: The command to collect. - collection_id: An identifier used to build the eAPI request ID. + command + The command to collect. + collection_id + An identifier used to build the eAPI request ID. """ commands: list[dict[str, str | int]] = [] if self.enable and self._enable_password is not None: @@ -407,9 +449,12 @@ async def copy(self, sources: list[Path], destination: Path, direction: Literal[ Parameters ---------- - sources: List of files to copy to or from the device. - destination: Local or remote destination when copying the files. Can be a folder. - direction: Defines if this coroutine copies files to or from the device. + sources + List of files to copy to or from the device. + destination + Local or remote destination when copying the files. Can be a folder. + direction + Defines if this coroutine copies files to or from the device. """ async with asyncssh.connect( diff --git a/anta/inventory/__init__.py b/anta/inventory/__init__.py index 46609676a..29450be62 100644 --- a/anta/inventory/__init__.py +++ b/anta/inventory/__init__.py @@ -46,8 +46,10 @@ def _update_disable_cache(kwargs: dict[str, Any], *, inventory_disable_cache: bo Parameters ---------- - inventory_disable_cache: The value of disable_cache in the inventory - kwargs: The kwargs to instantiate the device + inventory_disable_cache + The value of disable_cache in the inventory. + kwargs + The kwargs to instantiate the device. """ updated_kwargs = kwargs.copy() @@ -64,9 +66,12 @@ def _parse_hosts( Parameters ---------- - inventory_input: AntaInventoryInput used to parse the devices - inventory: AntaInventory to add the parsed devices to - **kwargs: Additional keyword arguments to pass to the device constructor + inventory_input + AntaInventoryInput used to parse the devices. + inventory + AntaInventory to add the parsed devices to. + **kwargs + Additional keyword arguments to pass to the device constructor. """ if inventory_input.hosts is None: @@ -93,13 +98,17 @@ def _parse_networks( Parameters ---------- - inventory_input: AntaInventoryInput used to parse the devices - inventory: AntaInventory to add the parsed devices to - **kwargs: Additional keyword arguments to pass to the device constructor + inventory_input + AntaInventoryInput used to parse the devices. + inventory + AntaInventory to add the parsed devices to. + **kwargs + Additional keyword arguments to pass to the device constructor. Raises ------ - InventoryIncorrectSchemaError: Inventory file is not following AntaInventory Schema. + InventoryIncorrectSchemaError + Inventory file is not following AntaInventory Schema. """ if inventory_input.networks is None: @@ -126,13 +135,17 @@ def _parse_ranges( Parameters ---------- - inventory_input: AntaInventoryInput used to parse the devices - inventory: AntaInventory to add the parsed devices to - **kwargs: Additional keyword arguments to pass to the device constructor + inventory_input + AntaInventoryInput used to parse the devices. + inventory + AntaInventory to add the parsed devices to. + **kwargs + Additional keyword arguments to pass to the device constructor. Raises ------ - InventoryIncorrectSchemaError: Inventory file is not following AntaInventory Schema. + InventoryIncorrectSchemaError + Inventory file is not following AntaInventory Schema. """ if inventory_input.ranges is None: @@ -177,19 +190,29 @@ def parse( Parameters ---------- - filename: Path to device inventory YAML file. - username: Username to use to connect to devices. - password: Password to use to connect to devices. - enable_password: Enable password to use if required. - timeout: Timeout value in seconds for outgoing API calls. - enable: Whether or not the commands need to be run in enable mode towards the devices. - insecure: Disable SSH Host Key validation. - disable_cache: Disable cache globally. + filename + Path to device inventory YAML file. + username + Username to use to connect to devices. + password + Password to use to connect to devices. + enable_password + Enable password to use if required. + timeout + Timeout value in seconds for outgoing API calls. + enable + Whether or not the commands need to be run in enable mode towards the devices. + insecure + Disable SSH Host Key validation. + disable_cache + Disable cache globally. Raises ------ - InventoryRootKeyError: Root key of inventory is missing. - InventoryIncorrectSchemaError: Inventory file is not following AntaInventory Schema. + InventoryRootKeyError + Root key of inventory is missing. + InventoryIncorrectSchemaError + Inventory file is not following AntaInventory Schema. """ inventory = AntaInventory() @@ -256,12 +279,16 @@ def get_inventory(self, *, established_only: bool = False, tags: set[str] | None Parameters ---------- - established_only: Whether or not to include only established devices. - tags: Tags to filter devices. - devices: Names to filter devices. + established_only + Whether or not to include only established devices. + tags + Tags to filter devices. + devices + Names to filter devices. Returns ------- + AntaInventory An inventory with filtered AntaDevice objects. """ @@ -295,7 +322,8 @@ def add_device(self, device: AntaDevice) -> None: Parameters ---------- - device: Device object to be added + device + Device object to be added. """ self[device.name] = device diff --git a/anta/inventory/models.py b/anta/inventory/models.py index 5796ef700..2eea701f6 100644 --- a/anta/inventory/models.py +++ b/anta/inventory/models.py @@ -21,11 +21,16 @@ class AntaInventoryHost(BaseModel): Attributes ---------- - host: IP Address or FQDN of the device. - port: Custom eAPI port to use. - name: Custom name of the device. - tags: Tags of the device. - disable_cache: Disable cache for this device. + host : Hostname | IPvAnyAddress + IP Address or FQDN of the device. + port : Port | None + Custom eAPI port to use. + name : str | None + Custom name of the device. + tags : set[str] + Tags of the device. + disable_cache : bool + Disable cache for this device. """ @@ -43,9 +48,12 @@ class AntaInventoryNetwork(BaseModel): Attributes ---------- - network: Subnet to use for scanning. - tags: Tags of the devices in this network. - disable_cache: Disable cache for all devices in this network. + network : IPvAnyNetwork + Subnet to use for scanning. + tags : set[str] + Tags of the devices in this network. + disable_cache : bool + Disable cache for all devices in this network. """ @@ -61,10 +69,14 @@ class AntaInventoryRange(BaseModel): Attributes ---------- - start: IPv4 or IPv6 address for the beginning of the range. - stop: IPv4 or IPv6 address for the end of the range. - tags: Tags of the devices in this IP range. - disable_cache: Disable cache for all devices in this IP range. + start : IPvAnyAddress + IPv4 or IPv6 address for the beginning of the range. + stop : IPvAnyAddress + IPv4 or IPv6 address for the end of the range. + tags : set[str] + Tags of the devices in this IP range. + disable_cache : bool + Disable cache for all devices in this IP range. """ @@ -90,6 +102,7 @@ def yaml(self) -> str: Returns ------- + str The YAML representation string of this model. """ # TODO: Pydantic and YAML serialization/deserialization is not supported natively. diff --git a/anta/logger.py b/anta/logger.py index b64fbe7b4..54733fb73 100644 --- a/anta/logger.py +++ b/anta/logger.py @@ -51,8 +51,10 @@ def setup_logging(level: LogLevel = Log.INFO, file: Path | None = None) -> None: Parameters ---------- - level: ANTA logging level - file: Send logs to a file + level + ANTA logging level + file + Send logs to a file """ # Init root logger @@ -106,9 +108,12 @@ def anta_log_exception(exception: BaseException, message: str | None = None, cal Parameters ---------- - exception: The Exception being logged. - message: An optional message. - calling_logger: A logger to which the exception should be logged. If not present, the logger in this file is used. + exception + The Exception being logged. + message + An optional message. + calling_logger + A logger to which the exception should be logged. If not present, the logger in this file is used. """ if calling_logger is None: diff --git a/anta/models.py b/anta/models.py index e2cf49857..9a695bcd6 100644 --- a/anta/models.py +++ b/anta/models.py @@ -48,11 +48,16 @@ class AntaTemplate: Attributes ---------- - 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. + 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. """ # pylint: disable=too-few-public-methods @@ -97,18 +102,20 @@ def render(self, **params: str | int | bool) -> AntaCommand: Parameters ---------- - params: dictionary of variables with string values to render the Python f-string + 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. + AntaTemplateRenderError + If a parameter is missing to render the AntaTemplate instance. """ try: command = self.template.format(**params) @@ -141,15 +148,24 @@ class AntaCommand(BaseModel): Attributes ---------- - command: Device command - version: eAPI version - valid values are 1 or "latest". - revision: eAPI revision of the command. Valid values are 1 to 99. Revision has precedence over version. - ofmt: eAPI output - json or text. - output: Output of the command. Only defined if there was no errors. - template: AntaTemplate object used to render this command. - errors: If the command execution fails, eAPI returns a list of strings detailing the error(s). - params: Pydantic Model containing the variables values used to render the template. - use_cache: Enable or disable caching for this AntaCommand if the AntaDevice supports it. + command + Device command. + version + eAPI version - valid values are 1 or "latest". + revision + eAPI revision of the command. Valid values are 1 to 99. Revision has precedence over version. + ofmt + eAPI output - json or text. + output + Output of the command. Only defined if there was no errors. + template + AntaTemplate object used to render this command. + errors + If the command execution fails, eAPI returns a list of strings detailing the error(s). + params + Pydantic Model containing the variables values used to render the template. + use_cache + Enable or disable caching for this AntaCommand if the AntaDevice supports it. """ @@ -214,9 +230,9 @@ def requires_privileges(self) -> bool: Raises ------ - RuntimeError - If the command has not been collected and has not returned an error. - AntaDevice.collect() must be called before this property. + RuntimeError + If the command has not been collected and has not returned an error. + AntaDevice.collect() must be called before this property. """ if not self.collected and not self.error: msg = f"Command '{self.command}' has not been collected and has not returned an error. Call AntaDevice.collect()." @@ -229,9 +245,9 @@ def supported(self) -> bool: Raises ------ - RuntimeError - If the command has not been collected and has not returned an error. - AntaDevice.collect() must be called before this property. + RuntimeError + If the command has not been collected and has not returned an error. + AntaDevice.collect() must be called before this property. """ if not self.collected and not self.error: msg = f"Command '{self.command}' has not been collected and has not returned an error. Call AntaDevice.collect()." @@ -247,8 +263,10 @@ def __init__(self, template: AntaTemplate, key: str) -> None: Parameters ---------- - template: The AntaTemplate instance that failed to render - key: Key that has not been provided to render the template + template + The AntaTemplate instance that failed to render. + key + Key that has not been provided to render the template. """ self.template = template @@ -297,11 +315,16 @@ def test(self) -> None: 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 + 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. """ # Mandatory class attributes @@ -332,7 +355,8 @@ class Input(BaseModel): Attributes ---------- - result_overwrite: Define fields to overwrite in the TestResult object + result_overwrite + Define fields to overwrite in the TestResult object. """ model_config = ConfigDict(extra="forbid") @@ -351,9 +375,12 @@ class ResultOverwrite(BaseModel): Attributes ---------- - description: overwrite TestResult.description - categories: overwrite TestResult.categories - custom_field: a free string that will be included in the TestResult object + description + Overwrite `TestResult.description`. + categories + Overwrite `TestResult.categories`. + custom_field + A free string that will be included in the TestResult object. """ @@ -367,7 +394,8 @@ class Filters(BaseModel): Attributes ---------- - tags: Tag of devices on which to run the test. + tags + Tag of devices on which to run the test. """ model_config = ConfigDict(extra="forbid") @@ -383,10 +411,13 @@ def __init__( 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. + 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 @@ -558,14 +589,18 @@ async def wrapper( 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. + 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 ------- - result: TestResult instance attribute populated with error status if any + TestResult + The TestResult instance attribute populated with error status if any. """ if self.result.result != "unset": diff --git a/anta/reporter/__init__.py b/anta/reporter/__init__.py index e74aaec5f..01baf3a6e 100644 --- a/anta/reporter/__init__.py +++ b/anta/reporter/__init__.py @@ -45,12 +45,15 @@ def _split_list_to_txt_list(self, usr_list: list[str], delimiter: str | None = N Parameters ---------- - usr_list (list[str]): List of string to concatenate - delimiter (str, optional): A delimiter to use to start string. Defaults to None. + usr_list : list[str] + List of string to concatenate. + delimiter : str, optional + A delimiter to use to start string. Defaults to None. Returns ------- - str: Multi-lines string + str + Multi-lines string. """ if delimiter is not None: @@ -64,11 +67,14 @@ def _build_headers(self, headers: list[str], table: Table) -> Table: Parameters ---------- - headers: List of headers. - table: A rich Table instance. + headers + List of headers. + table + A rich Table instance. Returns ------- + Table A rich `Table` instance with headers. """ @@ -84,11 +90,11 @@ def _color_result(self, status: AntaTestStatus) -> str: Parameters ---------- - status: AntaTestStatus enum to color. + status: AntaTestStatus enum to color. Returns ------- - The colored string. + The colored string. """ color = RICH_COLOR_THEME.get(str(status), "") @@ -101,11 +107,14 @@ def report_all(self, manager: ResultManager, title: str = "All tests results") - Parameters ---------- - manager: A ResultManager instance. - title: Title for the report. Defaults to 'All tests results'. + manager + A ResultManager instance. + title + Title for the report. Defaults to 'All tests results'. Returns ------- + Table A fully populated rich `Table` """ @@ -135,12 +144,16 @@ def report_summary_tests( Parameters ---------- - manager: A ResultManager instance. - tests: List of test names to include. None to select all tests. - title: Title of the report. + manager + A ResultManager instance. + tests + List of test names to include. None to select all tests. + title + Title of the report. Returns ------- + Table A fully populated rich `Table`. """ table = Table(title=title, show_lines=True) @@ -177,12 +190,16 @@ def report_summary_devices( Parameters ---------- - manager: A ResultManager instance. - devices: List of device names to include. None to select all devices. - title: Title of the report. + manager + A ResultManager instance. + devices + List of device names to include. None to select all devices. + title + Title of the report. Returns ------- + Table A fully populated rich `Table`. """ table = Table(title=title, show_lines=True) @@ -225,6 +242,9 @@ def render(self, data: list[dict[str, Any]], *, trim_blocks: bool = True, lstrip Report is built based on a J2 template provided by user. Data structure sent to template is: + Example + ------- + ``` >>> print(ResultManager.json) [ { @@ -236,15 +256,20 @@ def render(self, data: list[dict[str, Any]], *, trim_blocks: bool = True, lstrip description: ..., } ] + ``` Parameters ---------- - data: List of results from ResultManager.results - trim_blocks: enable trim_blocks for J2 rendering. - lstrip_blocks: enable lstrip_blocks for J2 rendering. + data + List of results from `ResultManager.results`. + trim_blocks + enable trim_blocks for J2 rendering. + lstrip_blocks + enable lstrip_blocks for J2 rendering. Returns ------- + str Rendered template """ diff --git a/anta/reporter/csv_reporter.py b/anta/reporter/csv_reporter.py index 221cbec81..570da9e6b 100644 --- a/anta/reporter/csv_reporter.py +++ b/anta/reporter/csv_reporter.py @@ -42,12 +42,15 @@ def split_list_to_txt_list(cls, usr_list: list[str], delimiter: str = " - ") -> Parameters ---------- - usr_list: List of string to concatenate - delimiter: A delimiter to use to start string. Defaults to None. + usr_list + List of string to concatenate. + delimiter + A delimiter to use to start string. Defaults to None. Returns ------- - str: Multi-lines string + str + Multi-lines string. """ return f"{delimiter}".join(f"{line}" for line in usr_list) @@ -57,9 +60,15 @@ def convert_to_list(cls, result: TestResult) -> list[str]: """ Convert a TestResult into a list of string for creating file content. - Args: - ---- - results: A TestResult to convert into list. + Parameters + ---------- + results + A TestResult to convert into list. + + Returns + ------- + list[str] + TestResult converted into a list. """ message = cls.split_list_to_txt_list(result.messages) if len(result.messages) > 0 else "" categories = cls.split_list_to_txt_list(result.categories) if len(result.categories) > 0 else "None" @@ -76,14 +85,17 @@ def convert_to_list(cls, result: TestResult) -> list[str]: def generate(cls, results: ResultManager, csv_filename: pathlib.Path) -> None: """Build CSV flle with tests results. - Parameter - --------- - results: A ResultManager instance. - csv_filename: File path where to save CSV data. - - Raise - ----- - OSError if any is raised while writing the CSV file. + Parameters + ---------- + results + A ResultManager instance. + csv_filename + File path where to save CSV data. + + Raises + ------ + OSError + if any is raised while writing the CSV file. """ headers = [ cls.Headers.device, diff --git a/anta/reporter/md_reporter.py b/anta/reporter/md_reporter.py index 7b97fb176..f4eadb2b5 100644 --- a/anta/reporter/md_reporter.py +++ b/anta/reporter/md_reporter.py @@ -41,8 +41,10 @@ def generate(cls, results: ResultManager, md_filename: Path) -> None: Parameters ---------- - results: The ResultsManager instance containing all test results. - md_filename: The path to the markdown file to write the report into. + results + The ResultsManager instance containing all test results. + md_filename + The path to the markdown file to write the report into. """ try: with md_filename.open("w", encoding="utf-8") as mdfile: @@ -74,8 +76,10 @@ def __init__(self, mdfile: TextIOWrapper, results: ResultManager) -> None: Parameters ---------- - mdfile: An open file object to write the markdown data into. - results: The ResultsManager instance containing all test results. + mdfile + An open file object to write the markdown data into. + results + The ResultsManager instance containing all test results. """ self.mdfile = mdfile self.results = results @@ -102,12 +106,13 @@ def generate_heading_name(self) -> str: Returns ------- - str: Formatted header name. + str + Formatted header name. Example ------- - - `ANTAReport` will become ANTA Report. - - `TestResultsSummary` will become Test Results Summary. + - `ANTAReport` will become ANTA Report. + - `TestResultsSummary` will become Test Results Summary. """ class_name = self.__class__.__name__ @@ -124,8 +129,10 @@ def write_table(self, table_heading: list[str], *, last_table: bool = False) -> Parameters ---------- - table_heading: List of strings to join for the table heading. - last_table: Flag to determine if it's the last table of the markdown file to avoid unnecessary new line. Defaults to False. + table_heading + List of strings to join for the table heading. + last_table + Flag to determine if it's the last table of the markdown file to avoid unnecessary new line. Defaults to False. """ self.mdfile.write("\n".join(table_heading) + "\n") for row in self.generate_rows(): @@ -140,11 +147,12 @@ def write_heading(self, heading_level: int) -> None: Parameters ---------- - heading_level: The level of the heading (1-6). + heading_level + The level of the heading (1-6). Example ------- - ## Test Results Summary + ## Test Results Summary """ # Ensure the heading level is within the valid range of 1 to 6 heading_level = max(1, min(heading_level, 6)) @@ -157,11 +165,13 @@ def safe_markdown(self, text: str | None) -> str: Parameters ---------- - text: The text to escape markdown characters from. + text + The text to escape markdown characters from. Returns ------- - str: The text with escaped markdown characters. + str + The text with escaped markdown characters. """ # Custom field from a TestResult object can be None if text is None: diff --git a/anta/result_manager/__init__.py b/anta/result_manager/__init__.py index 95da45684..b1fd9c2d4 100644 --- a/anta/result_manager/__init__.py +++ b/anta/result_manager/__init__.py @@ -21,52 +21,52 @@ class ResultManager: Examples -------- - Create Inventory: + Create Inventory: - inventory_anta = AntaInventory.parse( - filename='examples/inventory.yml', - username='ansible', - password='ansible', + 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') ) - 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, - ), - ] + 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, + ), + ] """ def __init__(self) -> None: @@ -143,7 +143,8 @@ def _update_status(self, test_status: AntaTestStatus) -> None: Parameters ---------- - test_status: AntaTestStatus to update the ResultManager status. + test_status + AntaTestStatus to update the ResultManager status. """ if test_status == "error": self.error_status = True @@ -158,7 +159,8 @@ def _update_stats(self, result: TestResult) -> None: Parameters ---------- - result: TestResult to update the statistics. + result + TestResult to update the statistics. """ result.categories = [ " ".join(word.upper() if word.lower() in ACRONYM_CATEGORIES else word.title() for word in category.split()) for category in result.categories @@ -194,7 +196,8 @@ def add(self, result: TestResult) -> None: Parameters ---------- - result: TestResult to add to the ResultManager instance. + result + TestResult to add to the ResultManager instance. """ self._result_entries.append(result) self._update_status(result.result) @@ -210,12 +213,15 @@ def get_results(self, status: set[AntaTestStatus] | None = None, sort_by: list[s Parameters ---------- - status: Optional set of AntaTestStatus enum members to filter the results. - sort_by: Optional list of TestResult fields to sort the results. + status + Optional set of AntaTestStatus enum members to filter the results. + sort_by + Optional list of TestResult fields to sort the results. Returns ------- - List of TestResult. + list[TestResult] + List of results. """ # Return all results if no status is provided, otherwise return results for multiple statuses results = self._result_entries if status is None else list(chain.from_iterable(self.results_by_status.get(status, []) for status in status)) @@ -236,10 +242,12 @@ def get_total_results(self, status: set[AntaTestStatus] | None = None) -> int: Parameters ---------- - status: Optional set of AntaTestStatus enum members to filter the results. + status + Optional set of AntaTestStatus enum members to filter the results. Returns ------- + int Total number of results. """ if status is None: @@ -258,10 +266,12 @@ def filter(self, hide: set[AntaTestStatus]) -> ResultManager: Parameters ---------- - hide: Set of AntaTestStatus enum members to select tests to hide based on their status. + hide + Set of AntaTestStatus enum members to select tests to hide based on their status. Returns ------- + ResultManager A filtered `ResultManager`. """ possible_statuses = set(AntaTestStatus) @@ -274,10 +284,12 @@ def filter_by_tests(self, tests: set[str]) -> ResultManager: Parameters ---------- - tests: Set of test names to filter the results. + tests + Set of test names to filter the results. Returns ------- + ResultManager A filtered `ResultManager`. """ manager = ResultManager() @@ -289,10 +301,11 @@ def filter_by_devices(self, devices: set[str]) -> ResultManager: Parameters ---------- - devices: Set of device names to filter the results. + devices: Set of device names to filter the results. Returns ------- + ResultManager A filtered `ResultManager`. """ manager = ResultManager() @@ -304,6 +317,7 @@ def get_tests(self) -> set[str]: Returns ------- + set[str] Set of test names. """ return {str(result.test) for result in self._result_entries} @@ -313,6 +327,7 @@ def get_devices(self) -> set[str]: Returns ------- + set[str] Set of device names. """ return {str(result.name) for result in self._result_entries} diff --git a/anta/result_manager/models.py b/anta/result_manager/models.py index 2bb2aed2e..32975816c 100644 --- a/anta/result_manager/models.py +++ b/anta/result_manager/models.py @@ -33,13 +33,20 @@ class TestResult(BaseModel): Attributes ---------- - name: Name of the device where the test was run. - test: Name of the test run on the device. - categories: List of categories the TestResult belongs to. Defaults to the AntaTest categories. - description: Description of the TestResult. Defaults to the AntaTest description. - result: Result of the test. Must be one of the AntaTestStatus Enum values: unset, success, failure, error or skipped. - messages: Messages to report after the test, if any. - custom_field: Custom field to store a string for flexibility in integrating with ANTA. + name : str + Name of the device where the test was run. + test : str + Name of the test run on the device. + categories : list[str] + List of categories the TestResult belongs to. Defaults to the AntaTest categories. + description : str + Description of the TestResult. Defaults to the AntaTest description. + result : AntaTestStatus + Result of the test. Must be one of the AntaTestStatus Enum values: unset, success, failure, error or skipped. + messages : list[str] + Messages to report after the test, if any. + custom_field : str | None + Custom field to store a string for flexibility in integrating with ANTA. """ @@ -56,7 +63,8 @@ def is_success(self, message: str | None = None) -> None: Parameters ---------- - message: Optional message related to the test + message + Optional message related to the test. """ self._set_status(AntaTestStatus.SUCCESS, message) @@ -66,7 +74,8 @@ def is_failure(self, message: str | None = None) -> None: Parameters ---------- - message: Optional message related to the test + message + Optional message related to the test. """ self._set_status(AntaTestStatus.FAILURE, message) @@ -76,7 +85,8 @@ def is_skipped(self, message: str | None = None) -> None: Parameters ---------- - message: Optional message related to the test + message + Optional message related to the test. """ self._set_status(AntaTestStatus.SKIPPED, message) @@ -86,7 +96,8 @@ def is_error(self, message: str | None = None) -> None: Parameters ---------- - message: Optional message related to the test + message + Optional message related to the test. """ self._set_status(AntaTestStatus.ERROR, message) @@ -96,8 +107,10 @@ def _set_status(self, status: AntaTestStatus, message: str | None = None) -> Non Parameters ---------- - status: status of the test - message: optional message + status + Status of the test. + message + Optional message. """ self.result = status diff --git a/anta/runner.py b/anta/runner.py index df4c70cc4..e07cba94f 100644 --- a/anta/runner.py +++ b/anta/runner.py @@ -40,7 +40,8 @@ def adjust_rlimit_nofile() -> tuple[int, int]: Returns ------- - tuple[int, int]: The new soft and hard limits for open file descriptors. + tuple[int, int] + The new soft and hard limits for open file descriptors. """ try: nofile = int(os.environ.get("ANTA_NOFILE", DEFAULT_NOFILE)) @@ -61,7 +62,8 @@ def log_cache_statistics(devices: list[AntaDevice]) -> None: Parameters ---------- - devices: List of devices in the inventory. + devices + List of devices in the inventory. """ for device in devices: if device.cache_statistics is not None: @@ -80,13 +82,17 @@ async def setup_inventory(inventory: AntaInventory, tags: set[str] | None, devic Parameters ---------- - inventory: AntaInventory object that includes the device(s). - tags: Tags to filter devices from the inventory. - devices: Devices on which to run tests. None means all devices. + inventory + AntaInventory object that includes the device(s). + tags + Tags to filter devices from the inventory. + devices + Devices on which to run tests. None means all devices. Returns ------- - AntaInventory | None: The filtered inventory or None if there are no devices to run tests on. + AntaInventory | None + The filtered inventory or None if there are no devices to run tests on. """ if len(inventory) == 0: logger.info("The inventory is empty, exiting") @@ -118,13 +124,18 @@ def prepare_tests( Parameters ---------- - inventory: AntaInventory object that includes the device(s). - catalog: AntaCatalog object that includes the list of tests. - tests: Tests to run against devices. None means all tests. - tags: Tags to filter devices from the inventory. + inventory + AntaInventory object that includes the device(s). + catalog + AntaCatalog object that includes the list of tests. + tests + Tests to run against devices. None means all tests. + tags + Tags to filter devices from the inventory. Returns ------- + defaultdict[AntaDevice, set[AntaTestDefinition]] | None A mapping of devices to the tests to run or None if there are no tests to run. """ # Build indexes for the catalog. If `tests` is set, filter the indexes based on these tests @@ -162,10 +173,12 @@ def get_coroutines(selected_tests: defaultdict[AntaDevice, set[AntaTestDefinitio Parameters ---------- - selected_tests: A mapping of devices to the tests to run. The selected tests are generated by the `prepare_tests` function. + selected_tests + A mapping of devices to the tests to run. The selected tests are generated by the `prepare_tests` function. Returns ------- + list[Coroutine[Any, Any, TestResult]] The list of coroutines to run. """ coros = [] @@ -207,14 +220,22 @@ async def main( # noqa: PLR0913 Parameters ---------- - manager: ResultManager object to populate with the test results. - inventory: AntaInventory object that includes the device(s). - catalog: AntaCatalog object that includes the list of tests. - devices: Devices on which to run tests. None means all devices. These may come from the `--device / -d` CLI option in NRFU. - tests: Tests to run against devices. None means all tests. These may come from the `--test / -t` CLI option in NRFU. - tags: Tags to filter devices from the inventory. These may come from the `--tags` CLI option in NRFU. - established_only: Include only established device(s). - dry_run: Build the list of coroutine to run and stop before test execution. + manager + ResultManager object to populate with the test results. + inventory + AntaInventory object that includes the device(s). + catalog + AntaCatalog object that includes the list of tests. + devices + Devices on which to run tests. None means all devices. These may come from the `--device / -d` CLI option in NRFU. + tests + Tests to run against devices. None means all tests. These may come from the `--test / -t` CLI option in NRFU. + tags + Tags to filter devices from the inventory. These may come from the `--tags` CLI option in NRFU. + established_only + Include only established device(s). + dry_run + Build the list of coroutine to run and stop before test execution. """ # Adjust the maximum number of open file descriptors for the ANTA process limits = adjust_rlimit_nofile() diff --git a/anta/tests/flow_tracking.py b/anta/tests/flow_tracking.py index bab8860e6..5336cf14d 100644 --- a/anta/tests/flow_tracking.py +++ b/anta/tests/flow_tracking.py @@ -19,13 +19,17 @@ def validate_record_export(record_export: dict[str, str], tracker_info: dict[str """ Validate the record export configuration against the tracker info. - Args: - record_export (dict): The expected record export configuration. - tracker_info (dict): The actual tracker info from the command output. + Parameters + ---------- + record_export + The expected record export configuration. + tracker_info + The actual tracker info from the command output. Returns ------- - str : A failure message if the record export configuration does not match, otherwise blank string. + str + A failure message if the record export configuration does not match, otherwise blank string. """ failed_log = "" actual_export = {"inactive timeout": tracker_info.get("inactiveTimeout"), "interval": tracker_info.get("activeInterval")} @@ -39,13 +43,17 @@ def validate_exporters(exporters: list[dict[str, str]], tracker_info: dict[str, """ Validate the exporter configurations against the tracker info. - Args: - exporters (list[dict]): The list of expected exporter configurations. - tracker_info (dict): The actual tracker info from the command output. + Parameters + ---------- + exporters + The list of expected exporter configurations. + tracker_info + The actual tracker info from the command output. Returns ------- - str: Failure message if any exporter configuration does not match. + str + Failure message if any exporter configuration does not match. """ failed_log = "" for exporter in exporters: diff --git a/anta/tests/logging.py b/anta/tests/logging.py index b520fc1e1..c5202cce1 100644 --- a/anta/tests/logging.py +++ b/anta/tests/logging.py @@ -27,12 +27,15 @@ def _get_logging_states(logger: logging.Logger, command_output: str) -> str: Parameters ---------- - logger: The logger object. - command_output: The `show logging` output. + logger + The logger object. + command_output + The `show logging` output. Returns ------- - str: The operational logging states. + str + The operational logging states. """ log_states = command_output.partition("\n\nExternal configuration:")[0] diff --git a/anta/tests/routing/bgp.py b/anta/tests/routing/bgp.py index 70d2a6fcb..3477fc8b2 100644 --- a/anta/tests/routing/bgp.py +++ b/anta/tests/routing/bgp.py @@ -26,31 +26,38 @@ def _add_bgp_failures(failures: dict[tuple[str, str | None], dict[str, Any]], af Parameters ---------- - failures: The dictionary to which the failure will be added. - afi: The address family identifier. - vrf: The VRF name. - safi: The subsequent address family identifier. - issue: A description of the issue. Can be of any type. - - Example: + failures + The dictionary to which the failure will be added. + afi + The address family identifier. + vrf + The VRF name. + safi + The subsequent address family identifier. + issue + A description of the issue. Can be of any type. + + Example ------- The `failures` dictionary will have the following structure: - { - ('afi1', 'safi1'): { - 'afi': 'afi1', - 'safi': 'safi1', - 'vrfs': { - 'vrf1': issue1, - 'vrf2': issue2 - } - }, - ('afi2', None): { - 'afi': 'afi2', - 'vrfs': { - 'vrf1': issue3 - } + ``` + { + ('afi1', 'safi1'): { + 'afi': 'afi1', + 'safi': 'safi1', + 'vrfs': { + 'vrf1': issue1, + 'vrf2': issue2 + } + }, + ('afi2', None): { + 'afi': 'afi2', + 'vrfs': { + 'vrf1': issue3 } } + } + ``` """ key = (afi, safi) @@ -65,21 +72,27 @@ def _check_peer_issues(peer_data: dict[str, Any] | None) -> dict[str, Any]: Parameters ---------- - peer_data: The BGP peer data dictionary nested in the `show bgp summary` command. + peer_data + The BGP peer data dictionary nested in the `show bgp summary` command. Returns ------- - dict: Dictionary with keys indicating issues or an empty dictionary if no issues. + dict + Dictionary with keys indicating issues or an empty dictionary if no issues. Raises ------ - ValueError: If any of the required keys ("peerState", "inMsgQueue", "outMsgQueue") are missing in `peer_data`, i.e. invalid BGP peer data. + ValueError + If any of the required keys ("peerState", "inMsgQueue", "outMsgQueue") are missing in `peer_data`, i.e. invalid BGP peer data. - Example: + Example ------- - {"peerNotFound": True} - {"peerState": "Idle", "inMsgQueue": 2, "outMsgQueue": 0} - {} + This can for instance return + ``` + {"peerNotFound": True} + {"peerState": "Idle", "inMsgQueue": 2, "outMsgQueue": 0} + {} + ``` """ if peer_data is None: @@ -106,15 +119,21 @@ def _add_bgp_routes_failure( Parameters ---------- - bgp_routes: The list of expected routes. - bgp_output: The BGP output from the device. - peer: The IP address of the BGP peer. - vrf: The name of the VRF for which the routes need to be verified. - route_type: The type of BGP routes. Defaults to 'advertised_routes'. + bgp_routes + The list of expected routes. + bgp_output + The BGP output from the device. + peer + The IP address of the BGP peer. + vrf + The name of the VRF for which the routes need to be verified. + route_type + The type of BGP routes. Defaults to 'advertised_routes'. Returns ------- - dict[str, dict[str, dict[str, dict[str, list[str]]]]]: A dictionary containing the missing routes and invalid or inactive routes. + dict[str, dict[str, dict[str, dict[str, list[str]]]]] + A dictionary containing the missing routes and invalid or inactive routes. """ # Prepare the failure routes dictionary diff --git a/anta/tests/routing/isis.py b/anta/tests/routing/isis.py index dee472571..344605d3a 100644 --- a/anta/tests/routing/isis.py +++ b/anta/tests/routing/isis.py @@ -20,13 +20,15 @@ def _count_isis_neighbor(isis_neighbor_json: dict[str, Any]) -> int: """Count the number of isis neighbors. - Args - ---- - isis_neighbor_json: The JSON output of the `show isis neighbors` command. + Parameters + ---------- + isis_neighbor_json + The JSON output of the `show isis neighbors` command. Returns ------- - int: The number of isis neighbors. + int + The number of isis neighbors. """ count = 0 @@ -39,13 +41,15 @@ def _count_isis_neighbor(isis_neighbor_json: dict[str, Any]) -> int: def _get_not_full_isis_neighbors(isis_neighbor_json: dict[str, Any]) -> list[dict[str, Any]]: """Return the isis neighbors whose adjacency state is not `up`. - Args - ---- - isis_neighbor_json: The JSON output of the `show isis neighbors` command. + Parameters + ---------- + isis_neighbor_json + The JSON output of the `show isis neighbors` command. Returns ------- - list[dict[str, Any]]: A list of isis neighbors whose adjacency state is not `UP`. + list[dict[str, Any]] + A list of isis neighbors whose adjacency state is not `UP`. """ return [ @@ -66,14 +70,17 @@ def _get_not_full_isis_neighbors(isis_neighbor_json: dict[str, Any]) -> list[dic def _get_full_isis_neighbors(isis_neighbor_json: dict[str, Any], neighbor_state: Literal["up", "down"] = "up") -> list[dict[str, Any]]: """Return the isis neighbors whose adjacency state is `up`. - Args - ---- - isis_neighbor_json: The JSON output of the `show isis neighbors` command. - neighbor_state: Value of the neihbor state we are looking for. Default up + Parameters + ---------- + isis_neighbor_json + The JSON output of the `show isis neighbors` command. + neighbor_state + Value of the neihbor state we are looking for. Defaults to `up`. Returns ------- - list[dict[str, Any]]: A list of isis neighbors whose adjacency state is not `UP`. + list[dict[str, Any]] + A list of isis neighbors whose adjacency state is not `UP`. """ return [ diff --git a/anta/tests/routing/ospf.py b/anta/tests/routing/ospf.py index 342ada2f4..3ffd81d53 100644 --- a/anta/tests/routing/ospf.py +++ b/anta/tests/routing/ospf.py @@ -20,11 +20,13 @@ def _count_ospf_neighbor(ospf_neighbor_json: dict[str, Any]) -> int: Parameters ---------- - ospf_neighbor_json: The JSON output of the `show ip ospf neighbor` command. + ospf_neighbor_json + The JSON output of the `show ip ospf neighbor` command. Returns ------- - int: The number of OSPF neighbors. + int + The number of OSPF neighbors. """ count = 0 @@ -39,11 +41,13 @@ def _get_not_full_ospf_neighbors(ospf_neighbor_json: dict[str, Any]) -> list[dic Parameters ---------- - ospf_neighbor_json: The JSON output of the `show ip ospf neighbor` command. + ospf_neighbor_json + The JSON output of the `show ip ospf neighbor` command. Returns ------- - list[dict[str, Any]]: A list of OSPF neighbors whose adjacency state is not `full`. + list[dict[str, Any]] + A list of OSPF neighbors whose adjacency state is not `full`. """ return [ @@ -65,11 +69,13 @@ def _get_ospf_max_lsa_info(ospf_process_json: dict[str, Any]) -> list[dict[str, Parameters ---------- - ospf_process_json: OSPF process information in JSON format. + ospf_process_json + OSPF process information in JSON format. Returns ------- - list[dict[str, Any]]: A list of dictionaries containing OSPF LSAs information. + list[dict[str, Any]] + A list of dictionaries containing OSPF LSAs information. """ return [ diff --git a/anta/tools.py b/anta/tools.py index 55748b492..00aad5afe 100644 --- a/anta/tools.py +++ b/anta/tools.py @@ -34,12 +34,15 @@ def get_failed_logs(expected_output: dict[Any, Any], actual_output: dict[Any, An Parameters ---------- - expected_output (dict): Expected output of a test. - actual_output (dict): Actual output of a test + expected_output + Expected output of a test. + actual_output + Actual output of a test Returns ------- - str: Failed log of a test. + str + Failed log of a test. """ failed_logs = [] @@ -65,12 +68,15 @@ def custom_division(numerator: float, denominator: float) -> int | float: Parameters ---------- - numerator: The numerator. - denominator: The denominator. + numerator + The numerator. + denominator + The denominator. Returns ------- - Union[int, float]: The result of the division. + Union[int, float] + The result of the division. """ result = numerator / denominator return int(result) if result.is_integer() else result @@ -304,11 +310,13 @@ def cprofile(sort_by: str = "cumtime") -> Callable[[F], F]: Parameters ---------- - sort_by (str): The criterion to sort the profiling results. Default is 'cumtime'. + sort_by + The criterion to sort the profiling results. Default is 'cumtime'. Returns ------- - Callable: The decorated function with conditional profiling. + Callable + The decorated function with conditional profiling. """ def decorator(func: F) -> F: @@ -320,11 +328,14 @@ async def wrapper(*args: Any, **kwargs: Any) -> Any: Parameters ---------- - *args: Arbitrary positional arguments. - **kwargs: Arbitrary keyword arguments. + *args + Arbitrary positional arguments. + **kwargs + Arbitrary keyword arguments. Returns ------- + Any The result of the function call. """ cprofile_file = os.environ.get("ANTA_CPROFILE") diff --git a/docs/advanced_usages/as-python-lib.md b/docs/advanced_usages/as-python-lib.md index f8d67348b..08bb818c1 100644 --- a/docs/advanced_usages/as-python-lib.md +++ b/docs/advanced_usages/as-python-lib.md @@ -11,11 +11,11 @@ ANTA is a Python library that can be used in user applications. This section des ## [AntaDevice](../api/device.md#anta.device.AntaDevice) Abstract Class -A device is represented in ANTA as a instance of a subclass of the [AntaDevice](../api/device.md### ::: anta.device.AntaDevice) abstract class. +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 [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 [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. @@ -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/models.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. @@ -35,7 +35,7 @@ The [AntaInventory](../api/inventory.md#anta.inventory.AntaInventory) class is a [AntaInventory](../api/inventory.md#anta.inventory.AntaInventory) provides methods to interact with the ANTA inventory: -- The [add_device()](../api/inventory.md#anta.inventory.AntaInventory.add_device) method adds an [AntaDevice](../api/device.md### ::: anta.device.AntaDevice) instance to the inventory. Adding an entry to [AntaInventory](../api/inventory.md#anta.inventory.AntaInventory) with a key different from the device name is not allowed. +- The [add_device()](../api/inventory.md#anta.inventory.AntaInventory.add_device) method adds an [AntaDevice](../api/device.md#anta.device.AntaDevice) instance to the inventory. Adding an entry to [AntaInventory](../api/inventory.md#anta.inventory.AntaInventory) with a key different from the device name is not allowed. - The [get_inventory()](../api/inventory.md#anta.inventory.AntaInventory.get_inventory) returns a new [AntaInventory](../api/inventory.md#anta.inventory.AntaInventory) instance with filtered out devices based on the method inputs. - The [connect_inventory()](../api/inventory.md#anta.inventory.AntaInventory.connect_inventory) coroutine will execute the [refresh()](../api/device.md#anta.device.AntaDevice.refresh) coroutines of all the devices in the inventory. - The [parse()](../api/inventory.md#anta.inventory.AntaInventory.parse) static method creates an [AntaInventory](../api/inventory.md#anta.inventory.AntaInventory) instance from a YAML file and returns it. The devices are [AsyncEOSDevice](../api/device.md#anta.device.AsyncEOSDevice) instances. diff --git a/docs/advanced_usages/caching.md b/docs/advanced_usages/caching.md index ce4a7877c..7de310de7 100644 --- a/docs/advanced_usages/caching.md +++ b/docs/advanced_usages/caching.md @@ -47,7 +47,7 @@ There might be scenarios where caching is not wanted. You can disable caching in ```bash anta --disable-cache --username arista --password arista nrfu table ``` -2. Caching can be disabled per device, network or range by setting the `disable_cache` key to `True` when defining the ANTA [Inventory](../usage-inventory-catalog.md#create-an-inventory-file) file: +2. Caching can be disabled per device, network or range by setting the `disable_cache` key to `True` when defining the ANTA [Inventory](../usage-inventory-catalog.md#device-inventory) file: ```yaml anta_inventory: hosts: diff --git a/docs/api/device.md b/docs/api/device.md index 03cff192e..9401f59af 100644 --- a/docs/api/device.md +++ b/docs/api/device.md @@ -6,20 +6,18 @@ # AntaDevice base class -## UML representation - ![](../imgs/uml/anta.device.AntaDevice.jpeg) -### ::: anta.device.AntaDevice +## ::: anta.device.AntaDevice options: - filters: ["!^_[^_]", "!__(eq|rich_repr)__"] + filters: ["!^_[^_]", "!__(eq|rich_repr)__", "_collect"] # Async EOS device class -## UML representation - ![](../imgs/uml/anta.device.AsyncEOSDevice.jpeg) -### ::: anta.device.AsyncEOSDevice + + +## ::: anta.device.AsyncEOSDevice options: - filters: ["!^_[^_]", "!__(eq|rich_repr)__"] + filters: ["!^_[^_]", "!__(eq|rich_repr)__", "_collect"] diff --git a/docs/api/models.md b/docs/api/models.md index b0c1e916f..3175fce54 100644 --- a/docs/api/models.md +++ b/docs/api/models.md @@ -6,8 +6,6 @@ # Test definition -## UML Diagram - ![](../imgs/uml/anta.models.AntaTest.jpeg) ### ::: anta.models.AntaTest @@ -16,9 +14,8 @@ # Command definition -## UML Diagram - ![](../imgs/uml/anta.models.AntaCommand.jpeg) + ### ::: anta.models.AntaCommand !!! warning @@ -30,8 +27,6 @@ # Template definition -## UML Diagram - ![](../imgs/uml/anta.models.AntaTemplate.jpeg) ### ::: anta.models.AntaTemplate diff --git a/docs/api/result_manager.md b/docs/api/result_manager.md index 72e05aaf4..dca0a19dd 100644 --- a/docs/api/result_manager.md +++ b/docs/api/result_manager.md @@ -6,8 +6,6 @@ # Result Manager definition -## UML Diagram - ![](../imgs/uml/anta.result_manager.ResultManager.jpeg) ### ::: anta.result_manager.ResultManager diff --git a/docs/api/result_manager_models.md b/docs/api/result_manager_models.md index 096bd036b..d0ccc7983 100644 --- a/docs/api/result_manager_models.md +++ b/docs/api/result_manager_models.md @@ -6,8 +6,6 @@ # Test Result model -## UML Diagram - ![](../imgs/uml/anta.result_manager.models.TestResult.jpeg) ### ::: anta.result_manager.models.TestResult diff --git a/docs/cli/debug.md b/docs/cli/debug.md index 376dffb14..b0b8a164f 100644 --- a/docs/cli/debug.md +++ b/docs/cli/debug.md @@ -14,7 +14,7 @@ The ANTA CLI includes a set of debugging tools, making it easier to build and te These tools are especially helpful in building the tests, as they give a visual access to the output received from the eAPI. They also facilitate the extraction of output content for use in unit tests, as described in our [contribution guide](../contribution.md). !!! warning - The `debug` tools require a device from your inventory. Thus, you MUST use a valid [ANTA Inventory](../usage-inventory-catalog.md#create-an-inventory-file). + The `debug` tools require a device from your inventory. Thus, you MUST use a valid [ANTA Inventory](../usage-inventory-catalog.md#device-inventory). ## Executing an EOS command diff --git a/docs/cli/nrfu.md b/docs/cli/nrfu.md index 0de782551..579fbdeef 100644 --- a/docs/cli/nrfu.md +++ b/docs/cli/nrfu.md @@ -228,7 +228,7 @@ The template `./custom_template.j2` is a simple Jinja2 template: {% endfor %} ``` -The Jinja2 template has access to all `TestResult` elements and their values, as described in this [documentation](../api/result_manager_models.md#testresult-entry). +The Jinja2 template has access to all `TestResult` elements and their values, as described in this [documentation](../api/result_manager_models.md#anta.result_manager.models.TestResult). You can also save the report result to a file using the `--output` option: diff --git a/docs/getting-started.md b/docs/getting-started.md index 39b270ce1..c166ebe78 100644 --- a/docs/getting-started.md +++ b/docs/getting-started.md @@ -72,7 +72,7 @@ anta_inventory: tags: ['fabric', 'leaf'] ``` -> You can read more details about how to build your inventory [here](usage-inventory-catalog.md#create-an-inventory-file) +> You can read more details about how to build your inventory [here](usage-inventory-catalog.md#device-inventory) ## Test Catalog diff --git a/docs/stylesheets/extra.material.css b/docs/stylesheets/extra.material.css index 1724da961..44e7c1350 100644 --- a/docs/stylesheets/extra.material.css +++ b/docs/stylesheets/extra.material.css @@ -126,10 +126,8 @@ line-height: 1em; font-size: 1.3rem; margin: 1em 0; - /* font-weight: 700; */ letter-spacing: -.01em; color: var(--md-default-fg-color--light); - text-transform: capitalize; font-style: normal; font-weight: bold; } @@ -142,19 +140,15 @@ line-height: 1em; color: var(--md-default-fg-color--light); font-style: italic; - text-transform: capitalize; } .md-typeset h5, .md-typeset h6 { font-size: 0.9rem; margin: 1em 0; - /* font-weight: 700; */ letter-spacing: -.01em; - /* line-height: 2em; */ color: var(--md-default-fg-color--light); font-style: italic; - text-transform: capitalize; text-decoration: underline; } @@ -163,17 +157,13 @@ padding: .6rem .8rem; color: var(--md-default-fg-color); vertical-align: top; - /* background-color: var(--md-accent-bg-color); */ text-align: left; - /* min-width: 100%; */ - /* display: table; */ } .md-typeset table:not([class]) td { /* padding: .9375em 1.25em; */ border-collapse: collapse; vertical-align: center; text-align: left; - /* border-bottom: 1px solid var(--md-default-fg-color--light); */ } .md-typeset code { padding: 0 .2941176471em; @@ -250,3 +240,7 @@ div.doc-contents { padding-left: 25px; border-left: .05rem solid var(--md-typeset-table-color); } +h5.doc-heading { + /* Avoid to capitalize h5 headers for mkdocstrings */ + text-transform: none; +} diff --git a/docs/templates/python/material/anta_test.html b/docs/templates/python/material/anta_test.html.jinja similarity index 96% rename from docs/templates/python/material/anta_test.html rename to docs/templates/python/material/anta_test.html.jinja index ade0ba691..a40d86a24 100644 --- a/docs/templates/python/material/anta_test.html +++ b/docs/templates/python/material/anta_test.html.jinja @@ -31,7 +31,7 @@ {% endif %} {% with heading_level = heading_level + extra_level %} {% for attribute in attributes|order_members(config.members_order, members_list) %} - {% if members_list is not none or attribute.is_public(check_name=False) %} + {% if members_list is not none or attribute.is_public %} {% include attribute|get_template with context %} {% endif %} {% endfor %} @@ -60,7 +60,7 @@ {% include "attributes_table.html" with context %} {% set obj = old_obj %} {% else %} - {% if members_list is not none or class.is_public(check_name=False) %} + {% if members_list is not none or class.is_public %} {% include class|get_template with context %} {% endif %} {% endif %} @@ -82,7 +82,7 @@ {% with heading_level = heading_level + extra_level %} {% for function in functions|order_members(config.members_order, members_list) %} {% if not (obj.kind.value == "class" and function.name == "__init__" and config.merge_init_into_class) %} - {% if members_list is not none or function.is_public(check_name=False) %} + {% if members_list is not none or function.is_public %} {% include function|get_template with context %} {% endif %} {% endif %} @@ -104,7 +104,7 @@ {% endif %} {% with heading_level = heading_level + extra_level %} {% for module in modules|order_members(config.members_order.alphabetical, members_list) %} - {% if members_list is not none or module.is_public(check_name=False) %} + {% if members_list is not none or module.is_public %} {% include module|get_template with context %} {% endif %} {% endfor %} @@ -129,7 +129,7 @@ {% if not (obj.is_class and child.name == "__init__" and config.merge_init_into_class) %} - {% if members_list is not none or child.is_public(check_name=False) %} + {% if members_list is not none or child.is_public %} {% if child.is_attribute %} {% with attribute = child %} {% include attribute|get_template with context %} diff --git a/docs/templates/python/material/class.html b/docs/templates/python/material/class.html.jinja similarity index 91% rename from docs/templates/python/material/class.html rename to docs/templates/python/material/class.html.jinja index 940103b4f..1c1173ce4 100644 --- a/docs/templates/python/material/class.html +++ b/docs/templates/python/material/class.html.jinja @@ -1,4 +1,4 @@ -{% extends "_base/class.html" %} +{% extends "_base/class.html.jinja" %} {% set anta_test = namespace(found=false) %} {% for base in class.bases %} {% set basestr = base | string %} @@ -10,7 +10,7 @@ {% if anta_test.found %} {% set root = False %} {% set heading_level = heading_level + 1 %} - {% include "anta_test.html" with context %} + {% include "anta_test.html.jinja" with context %} {# render source after children - TODO make add flag to respect disabling it.. though do we want to disable?#}
Source code in diff --git a/docs/templates/python/material/docstring.html b/docs/templates/python/material/docstring.html.jinja similarity index 100% rename from docs/templates/python/material/docstring.html rename to docs/templates/python/material/docstring.html.jinja diff --git a/mkdocs.yml b/mkdocs.yml index 291fb2bda..e206ba218 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -153,7 +153,7 @@ markdown_extensions: separator: "-" # permalink: "#" permalink: true - # baselevel: 3 + baselevel: 2 - pymdownx.highlight - pymdownx.snippets: base_path: @@ -193,7 +193,7 @@ nav: - Configuration: api/tests.configuration.md - Connectivity: api/tests.connectivity.md - Field Notices: api/tests.field_notices.md - - Flow Tracking: api/test.flow_tracking.md + - Flow Tracking: api/tests.flow_tracking.md - GreenT: api/tests.greent.md - Hardware: api/tests.hardware.md - Interfaces: api/tests.interfaces.md diff --git a/pyproject.toml b/pyproject.toml index 7202d4839..cfcbdb6d1 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -81,18 +81,19 @@ dev = [ "yamllint>=1.32.0", ] doc = [ - "fontawesome_markdown", - "griffe >=0.46,<1.0.0", + "fontawesome_markdown>=0.2.6", + "griffe >=1.2.0", "mike==2.1.3", - "mkdocs-autorefs>=0.4.1", + "mkdocs>=1.6.1", + "mkdocs-autorefs>=1.2.0", "mkdocs-bootswatch>=1.1", - "mkdocs-git-revision-date-localized-plugin>=1.1.0", + "mkdocs-git-revision-date-localized-plugin>=1.2.8", "mkdocs-git-revision-date-plugin>=0.3.2", - "mkdocs-material-extensions>=1.0.3", - "mkdocs-material>=8.3.9", - "mkdocs>=1.3.1", - "mkdocstrings[python]>=0.20.0", - "mkdocs-glightbox>=0.4.0" + "mkdocs-glightbox>=0.4.0", + "mkdocs-material-extensions>=1.3.1", + "mkdocs-material>=9.5.34", + "mkdocstrings[python]>=0.26.0", + "mkdocstrings-python>=1.11.0" ] [project.urls] @@ -326,11 +327,17 @@ target-version = "py39" [tool.ruff.lint] # select all cause we like being suffering -select = ["ALL"] +select = ["ALL", + # By enabling a convention for docstrings, ruff automatically ignore some rules that need to be + # added back if we want them. + # https://docs.astral.sh/ruff/faq/#does-ruff-support-numpy-or-google-style-docstrings + # TODO: Augment the numpy convention rules to make sure we add all the params + # Uncomment below D417 + "D415", + # "D417", +] ignore = [ "ANN101", # Missing type annotation for `self` in method - we know what self is.. - "D203", # Ignoring conflicting D* warnings - one-blank-line-before-class - "D213", # Ignoring conflicting D* warnings - multi-line-summary-second-line "COM812", # Ignoring conflicting rules that may cause conflicts when used with the formatter "ISC001", # Ignoring conflicting rules that may cause conflicts when used with the formatter "TD002", # We don't have require authors in TODO From 02e8491a15089cc64d1a786a321f56bfaa593f27 Mon Sep 17 00:00:00 2001 From: vitthalmagadum <122079046+vitthalmagadum@users.noreply.github.com> Date: Wed, 11 Sep 2024 19:59:09 +0530 Subject: [PATCH 7/8] feat(anta): Added test case to verify Link Aggregation Control Protocol (LACP) functionality (#764) --- anta/custom_types.py | 8 ++ anta/tests/interfaces.py | 106 +++++++++++++++++- examples/tests.yaml | 6 ++ tests/units/anta_tests/test_interfaces.py | 124 ++++++++++++++++++++++ tests/units/test_custom_types.py | 19 ++++ 5 files changed, 262 insertions(+), 1 deletion(-) diff --git a/anta/custom_types.py b/anta/custom_types.py index 6747e7663..c1e1f6428 100644 --- a/anta/custom_types.py +++ b/anta/custom_types.py @@ -21,6 +21,8 @@ """Match EOS interface types like Ethernet1/1, Vlan1, Loopback1, etc.""" REGEXP_TYPE_VXLAN_SRC_INTERFACE = r"^(Loopback)([0-9]|[1-9][0-9]{1,2}|[1-7][0-9]{3}|8[01][0-9]{2}|819[01])$" """Match Vxlan source interface like Loopback10.""" +REGEX_TYPE_PORTCHANNEL = r"^Port-Channel[0-9]{1,6}$" +"""Match Port Channel interface like Port-Channel5.""" REGEXP_TYPE_HOSTNAME = r"^(([a-zA-Z0-9]|[a-zA-Z0-9][a-zA-Z0-9\-]*[a-zA-Z0-9])\.)*([A-Za-z0-9]|[A-Za-z0-9][A-Za-z0-9\-]*[A-Za-z0-9])$" """Match hostname like `my-hostname`, `my-hostname-1`, `my-hostname-1-2`.""" @@ -135,6 +137,12 @@ def validate_regex(value: str) -> str: BeforeValidator(interface_autocomplete), BeforeValidator(interface_case_sensitivity), ] +PortChannelInterface = Annotated[ + str, + Field(pattern=REGEX_TYPE_PORTCHANNEL), + BeforeValidator(interface_autocomplete), + BeforeValidator(interface_case_sensitivity), +] Afi = Literal["ipv4", "ipv6", "vpn-ipv4", "vpn-ipv6", "evpn", "rt-membership", "path-selection", "link-state"] Safi = Literal["unicast", "multicast", "labeled-unicast", "sr-te"] EncryptionAlgorithm = Literal["RSA", "ECDSA"] diff --git a/anta/tests/interfaces.py b/anta/tests/interfaces.py index dfbf15aa6..9ff1cf357 100644 --- a/anta/tests/interfaces.py +++ b/anta/tests/interfaces.py @@ -15,7 +15,7 @@ from pydantic_extra_types.mac_address import MacAddress from anta import GITHUB_SUGGESTION -from anta.custom_types import EthernetInterface, Interface, Percent, PositiveInteger +from anta.custom_types import EthernetInterface, Interface, Percent, PortChannelInterface, PositiveInteger from anta.decorators import skip_on_platforms from anta.models import AntaCommand, AntaTemplate, AntaTest from anta.tools import custom_division, get_failed_logs, get_item, get_value @@ -883,3 +883,107 @@ def test(self) -> None: output["speed"] = f"{custom_division(output['speed'], BPS_GBPS_CONVERSIONS)}Gbps" failed_log = get_failed_logs(expected_interface_output, actual_interface_output) self.result.is_failure(f"For interface {intf}:{failed_log}\n") + + +class VerifyLACPInterfacesStatus(AntaTest): + """Verifies the Link Aggregation Control Protocol (LACP) status of the provided interfaces. + + - Verifies that the interface is a member of the LACP port channel. + - Ensures that the synchronization is established. + - Ensures the interfaces are in the correct state for collecting and distributing traffic. + - Validates that LACP settings, such as timeouts, are correctly configured. (i.e The long timeout mode, also known as "slow" mode, is the default setting.) + + Expected Results + ---------------- + * Success: The test will pass if the provided interfaces are bundled in port channel and all specified parameters are correct. + * Failure: The test will fail if any interface is not bundled in port channel or any of specified parameter is not correct. + + Examples + -------- + ```yaml + anta.tests.interfaces: + - VerifyLACPInterfacesStatus: + interfaces: + - name: Ethernet1 + portchannel: Port-Channel100 + ``` + """ + + name = "VerifyLACPInterfacesStatus" + description = "Verifies the Link Aggregation Control Protocol(LACP) status of the provided interfaces." + categories: ClassVar[list[str]] = ["interfaces"] + commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaTemplate(template="show lacp interface {interface}", revision=1)] + + class Input(AntaTest.Input): + """Input model for the VerifyLACPInterfacesStatus test.""" + + interfaces: list[LACPInterface] + """List of LACP member interface.""" + + class LACPInterface(BaseModel): + """Model for an LACP member interface.""" + + name: EthernetInterface + """Ethernet interface to validate.""" + portchannel: PortChannelInterface + """Port Channel in which the interface is bundled.""" + + def render(self, template: AntaTemplate) -> list[AntaCommand]: + """Render the template for each interface in the input list.""" + return [template.render(interface=interface.name) for interface in self.inputs.interfaces] + + @AntaTest.anta_test + def test(self) -> None: + """Main test function for VerifyLACPInterfacesStatus.""" + self.result.is_success() + + # Member port verification parameters. + member_port_details = ["activity", "aggregation", "synchronization", "collecting", "distributing", "timeout"] + + # Iterating over command output for different interfaces + for command, input_entry in zip(self.instance_commands, self.inputs.interfaces): + interface = input_entry.name + portchannel = input_entry.portchannel + + # Verify if a PortChannel is configured with the provided interface + if not (interface_details := get_value(command.json_output, f"portChannels.{portchannel}.interfaces.{interface}")): + self.result.is_failure(f"Interface '{interface}' is not configured to be a member of LACP '{portchannel}'.") + continue + + # Verify the interface is bundled in port channel. + actor_port_status = interface_details.get("actorPortStatus") + if actor_port_status != "bundled": + message = f"For Interface {interface}:\nExpected `bundled` as the local port status, but found `{actor_port_status}` instead.\n" + self.result.is_failure(message) + continue + + # Collecting actor and partner port details + actor_port_details = interface_details.get("actorPortState", {}) + partner_port_details = interface_details.get("partnerPortState", {}) + + # Collecting actual interface details + actual_interface_output = { + "actor_port_details": {param: actor_port_details.get(param, "NotFound") for param in member_port_details}, + "partner_port_details": {param: partner_port_details.get(param, "NotFound") for param in member_port_details}, + } + + # Forming expected interface details + expected_details = {param: param != "timeout" for param in member_port_details} + expected_interface_output = {"actor_port_details": expected_details, "partner_port_details": expected_details} + + # Forming failure message + if actual_interface_output != expected_interface_output: + message = f"For Interface {interface}:\n" + actor_port_failed_log = get_failed_logs( + expected_interface_output.get("actor_port_details", {}), actual_interface_output.get("actor_port_details", {}) + ) + partner_port_failed_log = get_failed_logs( + expected_interface_output.get("partner_port_details", {}), actual_interface_output.get("partner_port_details", {}) + ) + + if actor_port_failed_log: + message += f"Actor port details:{actor_port_failed_log}\n" + if partner_port_failed_log: + message += f"Partner port details:{partner_port_failed_log}\n" + + self.result.is_failure(message) diff --git a/examples/tests.yaml b/examples/tests.yaml index f5a5ca46b..bb7d3b0d6 100644 --- a/examples/tests.yaml +++ b/examples/tests.yaml @@ -221,6 +221,12 @@ anta.tests.interfaces: - name: Eth2 auto: False speed: 2.5 + - VerifyLACPInterfacesStatus: + interfaces: + - name: Ethernet5 + portchannel: Port-Channel5 + - name: Ethernet6 + portchannel: Port-Channel5 anta.tests.lanz: - VerifyLANZ: diff --git a/tests/units/anta_tests/test_interfaces.py b/tests/units/anta_tests/test_interfaces.py index b8cf493da..c38ac89f2 100644 --- a/tests/units/anta_tests/test_interfaces.py +++ b/tests/units/anta_tests/test_interfaces.py @@ -21,6 +21,7 @@ VerifyIpVirtualRouterMac, VerifyL2MTU, VerifyL3MTU, + VerifyLACPInterfacesStatus, VerifyLoopbackCount, VerifyPortChannels, VerifyStormControlDrops, @@ -2441,4 +2442,127 @@ ], }, }, + { + "name": "success", + "test": VerifyLACPInterfacesStatus, + "eos_data": [ + { + "portChannels": { + "Port-Channel5": { + "interfaces": { + "Ethernet5": { + "actorPortStatus": "bundled", + "partnerPortState": { + "activity": True, + "timeout": False, + "aggregation": True, + "synchronization": True, + "collecting": True, + "distributing": True, + }, + "actorPortState": { + "activity": True, + "timeout": False, + "aggregation": True, + "synchronization": True, + "collecting": True, + "distributing": True, + }, + } + } + } + }, + "interface": "Ethernet5", + "orphanPorts": {}, + } + ], + "inputs": {"interfaces": [{"name": "Ethernet5", "portchannel": "Port-Channel5"}]}, + "expected": {"result": "success"}, + }, + { + "name": "failure-not-bundled", + "test": VerifyLACPInterfacesStatus, + "eos_data": [ + { + "portChannels": { + "Port-Channel5": { + "interfaces": { + "Ethernet5": { + "actorPortStatus": "No Aggregate", + } + } + } + }, + "interface": "Ethernet5", + "orphanPorts": {}, + } + ], + "inputs": {"interfaces": [{"name": "Ethernet5", "portchannel": "Po5"}]}, + "expected": { + "result": "failure", + "messages": ["For Interface Ethernet5:\nExpected `bundled` as the local port status, but found `No Aggregate` instead.\n"], + }, + }, + { + "name": "failure-no-details-found", + "test": VerifyLACPInterfacesStatus, + "eos_data": [ + { + "portChannels": {"Port-Channel5": {"interfaces": {}}}, + } + ], + "inputs": {"interfaces": [{"name": "Ethernet5", "portchannel": "Po 5"}]}, + "expected": { + "result": "failure", + "messages": ["Interface 'Ethernet5' is not configured to be a member of LACP 'Port-Channel5'."], + }, + }, + { + "name": "failure-lacp-params", + "test": VerifyLACPInterfacesStatus, + "eos_data": [ + { + "portChannels": { + "Port-Channel5": { + "interfaces": { + "Ethernet5": { + "actorPortStatus": "bundled", + "partnerPortState": { + "activity": False, + "timeout": False, + "aggregation": False, + "synchronization": False, + "collecting": True, + "distributing": True, + }, + "actorPortState": { + "activity": False, + "timeout": False, + "aggregation": False, + "synchronization": False, + "collecting": True, + "distributing": True, + }, + } + } + } + }, + "interface": "Ethernet5", + "orphanPorts": {}, + } + ], + "inputs": {"interfaces": [{"name": "Ethernet5", "portchannel": "port-channel 5"}]}, + "expected": { + "result": "failure", + "messages": [ + "For Interface Ethernet5:\n" + "Actor port details:\nExpected `True` as the activity, but found `False` instead." + "\nExpected `True` as the aggregation, but found `False` instead." + "\nExpected `True` as the synchronization, but found `False` instead." + "\nPartner port details:\nExpected `True` as the activity, but found `False` instead.\n" + "Expected `True` as the aggregation, but found `False` instead.\n" + "Expected `True` as the synchronization, but found `False` instead.\n" + ], + }, + }, ] diff --git a/tests/units/test_custom_types.py b/tests/units/test_custom_types.py index 8119849a6..e3dc09d25 100644 --- a/tests/units/test_custom_types.py +++ b/tests/units/test_custom_types.py @@ -17,6 +17,7 @@ from anta.custom_types import ( REGEX_BGP_IPV4_MPLS_VPN, REGEX_BGP_IPV4_UNICAST, + REGEX_TYPE_PORTCHANNEL, REGEXP_BGP_IPV4_MPLS_LABELS, REGEXP_BGP_L2VPN_AFI, REGEXP_EOS_BLACKLIST_CMDS, @@ -140,6 +141,22 @@ def test_regexp_type_vxlan_src_interface() -> None: assert re.match(REGEXP_TYPE_VXLAN_SRC_INTERFACE, "Loopback9000") is None +def test_regexp_type_portchannel() -> None: + """Test REGEX_TYPE_PORTCHANNEL.""" + # Test strings that should match the pattern + assert re.match(REGEX_TYPE_PORTCHANNEL, "Port-Channel5") is not None + assert re.match(REGEX_TYPE_PORTCHANNEL, "Port-Channel100") is not None + assert re.match(REGEX_TYPE_PORTCHANNEL, "Port-Channel999") is not None + assert re.match(REGEX_TYPE_PORTCHANNEL, "Port-Channel1000") is not None + + # Test strings that should not match the pattern + assert re.match(REGEX_TYPE_PORTCHANNEL, "Port-Channel") is None + assert re.match(REGEX_TYPE_PORTCHANNEL, "Port_Channel") is None + assert re.match(REGEX_TYPE_PORTCHANNEL, "Port_Channel1000") is None + assert re.match(REGEX_TYPE_PORTCHANNEL, "Port_Channel5/1") is None + assert re.match(REGEX_TYPE_PORTCHANNEL, "Port-Channel-100") is None + + def test_regexp_type_hostname() -> None: """Test REGEXP_TYPE_HOSTNAME.""" # Test strings that should match the pattern @@ -200,6 +217,8 @@ def test_interface_autocomplete_success() -> None: assert interface_autocomplete("eth2") == "Ethernet2" assert interface_autocomplete("po3") == "Port-Channel3" assert interface_autocomplete("lo4") == "Loopback4" + assert interface_autocomplete("Po1000") == "Port-Channel1000" + assert interface_autocomplete("Po 1000") == "Port-Channel1000" def test_interface_autocomplete_no_alias() -> None: From cff671ee9a9f8a5d861add8f598b54bc67fab153 Mon Sep 17 00:00:00 2001 From: "pre-commit-ci[bot]" <66853113+pre-commit-ci[bot]@users.noreply.github.com> Date: Mon, 16 Sep 2024 19:02:58 +0200 Subject: [PATCH 8/8] ci: update ruff-pre-commit to v0.6.5 (#830) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit [pre-commit.ci] pre-commit autoupdate updates: - [github.com/astral-sh/ruff-pre-commit: v0.6.4 → v0.6.5](https://github.com/astral-sh/ruff-pre-commit/compare/v0.6.4...v0.6.5) Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com> --- .pre-commit-config.yaml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 9da8faaad..003c2e5c5 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -43,7 +43,7 @@ repos: - '' - repo: https://github.com/astral-sh/ruff-pre-commit - rev: v0.6.4 + rev: v0.6.5 hooks: - id: ruff name: Run Ruff linter