Repository: aristanetworks/anta (python)

Issue:
Bug: False positive on VerifyNTPAssociation
`show ntp associations | json` can return the following output:
```
DC05-Temp-Bleaf1#sho ntp associations | json
{
    "peers": {
        "10.10.200.22 (ntp.example.com)": {
            "condition": "sys.peer",
            "peerIpAddr": "10.10.200.22",
            "refid": "164.100.255.122",
            "stratumLevel": 3,
            "peerType": "unicast",
            "lastReceived": 1730124132.0,
            "pollInterval": 1024,
            "reachabilityHistory": [
                true,
                true,
                true,
                true,
                true,
                true,
                true,
                true
            ],
            "delay": 0.524,
            "offset": 5.545,
            "jitter": 3.8
        }
    }
}
```
The FQDN of the NTP server can be included in parentheses. The test logic should be updated accordingly.

Pull request description:
fix(anta.tests): Fix VerifyNTPAssociations to support IP-DNS peers
# Description

Update VerifyNTPAssociations test to support IP-DNS peers in the JSON output. We now check if the provided NTP server IP or DNS is matching either the peer key or `peerIpAddr` in the details.

Fixes #900 

# Checklist:

<!-- Delete not relevant items !-->

- [X] My code follows the style guidelines of this project
- [X] I have performed a self-review of my code
- [X] I have run pre-commit for code linting and typing (`pre-commit run`)
- [X] I have made corresponding changes to the documentation
- [X] I have added tests that prove my fix is effective or that my feature works
- [X] New and existing unit tests pass locally with my changes (`tox -e testenv`)


Patch:
diff --git a/anta/input_models/system.py b/anta/input_models/system.py
new file mode 100644
index 00000000..7600d287
--- /dev/null
+++ b/anta/input_models/system.py
@@ -0,0 +1,31 @@
+# Copyright (c) 2023-2024 Arista Networks, Inc.
+# Use of this source code is governed by the Apache License 2.0
+# that can be found in the LICENSE file.
+"""Module containing input models for system tests."""
+
+from __future__ import annotations
+
+from ipaddress import IPv4Address
+
+from pydantic import BaseModel, ConfigDict, Field
+
+from anta.custom_types import Hostname
+
+
+class NTPServer(BaseModel):
+    """Model for a NTP server."""
+
+    model_config = ConfigDict(extra="forbid")
+    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."""
+
+    def __str__(self) -> str:
+        """Representation of the NTPServer model."""
+        return f"{self.server_address} (Preferred: {self.preferred}, Stratum: {self.stratum})"


Test patch:
diff --git a/anta/tests/system.py b/anta/tests/system.py
index 775eecbc..f2ce4718 100644
--- a/anta/tests/system.py
+++ b/anta/tests/system.py
@@ -8,14 +8,12 @@
 from __future__ import annotations
 
 import re
-from ipaddress import IPv4Address
 from typing import TYPE_CHECKING, ClassVar
 
-from pydantic import BaseModel, Field
-
-from anta.custom_types import Hostname, PositiveInteger
+from anta.custom_types import PositiveInteger
+from anta.input_models.system import NTPServer
 from anta.models import AntaCommand, AntaTest
-from anta.tools import get_failed_logs, get_value
+from anta.tools import get_value
 
 if TYPE_CHECKING:
     from anta.models import AntaTemplate
@@ -323,55 +321,33 @@ class VerifyNTPAssociations(AntaTest):
 
         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."""
+        NTPServer: ClassVar[type[NTPServer]] = NTPServer
 
     @AntaTest.anta_test
     def test(self) -> None:
         """Main test function for VerifyNTPAssociations."""
-        failures: str = ""
+        self.result.is_success()
 
-        if not (peer_details := get_value(self.instance_commands[0].json_output, "peers")):
-            self.result.is_failure("None of NTP peers are not configured.")
+        if not (peers := get_value(self.instance_commands[0].json_output, "peers")):
+            self.result.is_failure("No NTP peers 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
+            # We check `peerIpAddr` in the peer details - covering IPv4Address input, or the peer key - covering Hostname input.
+            matching_peer = next((peer for peer, peer_details in peers.items() if (server_address in {peer_details["peerIpAddr"], peer})), None)
 
-            # 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")}
+            if not matching_peer:
+                self.result.is_failure(f"{ntp_server} - Not configured")
+                continue
 
-            # 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"
+            # Collecting the expected/actual NTP peer details.
+            exp_condition = "sys.peer" if ntp_server.preferred else "candidate"
+            exp_stratum = ntp_server.stratum
+            act_condition = get_value(peers[matching_peer], "condition")
+            act_stratum = get_value(peers[matching_peer], "stratumLevel")
 
-        # Check if there are any failures.
-        if not failures:
-            self.result.is_success()
-        else:
-            self.result.is_failure(failures)
+            if act_condition != exp_condition or act_stratum != exp_stratum:
+                self.result.is_failure(f"{ntp_server} - Bad association; Condition: {act_condition}, Stratum: {act_stratum}")
diff --git a/docs/api/tests.system.md b/docs/api/tests.system.md
index 5dcfbc02..e474295c 100644
--- a/docs/api/tests.system.md
+++ b/docs/api/tests.system.md
@@ -7,6 +7,8 @@ anta_title: ANTA catalog for System tests
   ~ that can be found in the LICENSE file.
   -->
 
+# Tests
+
 ::: anta.tests.system
     options:
       show_root_heading: false
@@ -18,3 +20,15 @@ anta_title: ANTA catalog for System tests
       filters:
         - "!test"
         - "!render"
+
+# Input models
+
+::: anta.input_models.system
+
+    options:
+      show_root_heading: false
+      show_root_toc_entry: false
+      show_bases: false
+      anta_hide_test_module_description: true
+      show_labels: true
+      filters: ["!^__str__"]
diff --git a/tests/units/anta_tests/test_system.py b/tests/units/anta_tests/test_system.py
index 1eda8a1d..532bd233 100644
--- a/tests/units/anta_tests/test_system.py
+++ b/tests/units/anta_tests/test_system.py
@@ -346,6 +346,39 @@ poll interval unknown
         },
         "expected": {"result": "success"},
     },
+    {
+        "name": "success-ip-dns",
+        "test": VerifyNTPAssociations,
+        "eos_data": [
+            {
+                "peers": {
+                    "1.1.1.1 (1.ntp.networks.com)": {
+                        "condition": "sys.peer",
+                        "peerIpAddr": "1.1.1.1",
+                        "stratumLevel": 1,
+                    },
+                    "2.2.2.2 (2.ntp.networks.com)": {
+                        "condition": "candidate",
+                        "peerIpAddr": "2.2.2.2",
+                        "stratumLevel": 2,
+                    },
+                    "3.3.3.3 (3.ntp.networks.com)": {
+                        "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": "failure",
         "test": VerifyNTPAssociations,
@@ -380,9 +413,9 @@ poll interval unknown
         "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."
+                "1.1.1.1 (Preferred: True, Stratum: 1) - Bad association; Condition: candidate, Stratum: 2",
+                "2.2.2.2 (Preferred: False, Stratum: 2) - Bad association; Condition: sys.peer, Stratum: 2",
+                "3.3.3.3 (Preferred: False, Stratum: 2) - Bad association; Condition: sys.peer, Stratum: 3",
             ],
         },
     },
@@ -399,7 +432,7 @@ poll interval unknown
         },
         "expected": {
             "result": "failure",
-            "messages": ["None of NTP peers are not configured."],
+            "messages": ["No NTP peers configured"],
         },
     },
     {
@@ -430,7 +463,7 @@ poll interval unknown
         },
         "expected": {
             "result": "failure",
-            "messages": ["NTP peer 3.3.3.3 is not configured."],
+            "messages": ["3.3.3.3 (Preferred: False, Stratum: 1) - Not configured"],
         },
     },
     {
@@ -457,8 +490,9 @@ poll interval unknown
         "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."
+                "1.1.1.1 (Preferred: True, Stratum: 1) - Bad association; Condition: candidate, Stratum: 1",
+                "2.2.2.2 (Preferred: False, Stratum: 1) - Not configured",
+                "3.3.3.3 (Preferred: False, Stratum: 1) - Not configured",
             ],
         },
     },