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/2] 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/2] 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." + ], + }, + }, ]