Merge branch 'develop' into py313_tmp_fix

docs/requirements.txt

@@ -5,4 +5,4 @@ invoke==2.2.0
 jinja2==3.1.4
 MarkupSafe==2.1.5
 pytest==8.2.1
-ansible==9.6.0
+ansible==9.6.1

napalm/base/base.py

@@ -921,12 +921,23 @@ def get_ntp_servers(self) -> Dict[str, models.NTPServerDict]:
         """
         Returns the NTP servers configuration as dictionary.
         The keys of the dictionary represent the IP Addresses of the servers.
-        Inner dictionaries do not have yet any available keys.
+        Inner dictionaries MAY contain information regarding per-server configuration.
 
         Example::
 
             {
-                '192.168.0.1': {},
+                '192.168.0.1':
+                {
+                    'address': '192.168.0.1',
+                    'port': 123,
+                    'version': 4,
+                    'association_type': 'SERVER',
+                    'iburst': False,
+                    'prefer': False,
+                    'network_instance': 'default',
+                    'source_address': '192.0.2.1',
+                    'key_id': -1,
+                },
                 '17.72.148.53': {},
                 '37.187.56.220': {},
                 '162.158.20.18': {}

napalm/base/models.py

@@ -223,16 +223,22 @@
 
 NTPPeerDict = TypedDict(
     "NTPPeerDict",
-    {
-        # will populate it in the future wit potential keys
-    },
+    {},
     total=False,
 )
 
 NTPServerDict = TypedDict(
     "NTPServerDict",
     {
-        # will populate it in the future wit potential keys
+        "address": str,
+        "port": int,
+        "version": int,
+        "association_type": str,
+        "iburst": bool,
+        "prefer": bool,
+        "network_instance": str,
+        "source_address": str,
+        "key_id": int,
     },
     total=False,
 )

napalm/base/test/getters.py

@@ -311,7 +311,9 @@ def test_get_ntp_peers(self, test_case):
 
         for peer, peer_details in get_ntp_peers.items():
             assert isinstance(peer, str)
-            assert helpers.test_model(models.NTPPeerDict, peer_details)
+            assert helpers.test_model(
+                models.NTPPeerDict, peer_details, allow_subset=True
+            )
 
         return get_ntp_peers
 
@@ -323,7 +325,9 @@ def test_get_ntp_servers(self, test_case):
 
         for server, server_details in get_ntp_servers.items():
             assert isinstance(server, str)
-            assert helpers.test_model(models.NTPServerDict, server_details)
+            assert helpers.test_model(
+                models.NTPServerDict, server_details, allow_subset=True
+            )
 
         return get_ntp_servers
 

napalm/base/test/helpers.py

@@ -1,26 +1,31 @@
 """Several methods to help with the tests."""
 
 
-def test_model(model, data):
+def test_model(model, data, allow_subset=False):
     """Return if the dictionary `data` complies with the `model`."""
     # Access the underlying schema for a TypedDict directly
-    model = model.__annotations__
-    same_keys = set(model.keys()) == set(data.keys())
+    annotations = model.__annotations__
+    if allow_subset:
+        same_keys = set(data.keys()) <= set(annotations.keys())
+        source = data
+    else:
+        same_keys = set(annotations.keys()) == set(data.keys())
+        source = annotations
 
     if not same_keys:
         print(
             "model_keys: {}\ndata_keys: {}".format(
-                sorted(model.keys()), sorted(data.keys())
+                sorted(annotations.keys()), sorted(data.keys())
             )
         )
 
     correct_class = True
-    for key, instance_class in model.items():
-        correct_class = isinstance(data[key], instance_class) and correct_class
+    for key in source.keys():
+        correct_class = isinstance(data[key], annotations[key]) and correct_class
         if not correct_class:
             print(
                 "key: {}\nmodel_class: {}\ndata_class: {}".format(
-                    key, instance_class, data[key].__class__
+                    key, annotations[key], data[key].__class__
                 )
             )
 

napalm/eos/eos.py

@@ -1202,21 +1202,78 @@ def get_arp_table(self, vrf=""):
         return arp_table
 
     def get_ntp_servers(self):
-        commands = ["show running-config | section ntp"]
+        result = {}
 
-        raw_ntp_config = self._run_commands(commands, encoding="text")[0].get(
-            "output", ""
-        )
+        commands = ["show running-config | section ntp"]
 
-        ntp_config = napalm.base.helpers.textfsm_extractor(
-            self, "ntp_peers", raw_ntp_config
+        raw_ntp_config = (
+            self._run_commands(commands, encoding="text")[0]
+            .get("output", "")
+            .splitlines()
         )
 
-        return {
-            str(ntp_peer.get("ntppeer")): {}
-            for ntp_peer in ntp_config
-            if ntp_peer.get("ntppeer", "")
-        }
+        for server in raw_ntp_config:
+            details = {
+                "port": 123,
+                "version": 4,
+                "association_type": "SERVER",
+                "iburst": False,
+                "prefer": False,
+                "network_instance": "default",
+                "source_address": "",
+                "key_id": -1,
+            }
+            tokens = server.split()
+            if tokens[2] == "vrf":
+                details["network_instance"] = tokens[3]
+                server_ip = details["address"] = tokens[4]
+                idx = 5
+            else:
+                server_ip = details["address"] = tokens[2]
+                idx = 3
+            try:
+                parsed_address = napalm.base.helpers.ipaddress.ip_address(server_ip)
+                family = parsed_address.version
+            except ValueError:
+                # Assume family of 4, unless local-interface has no IPv4 addresses
+                family = 4
+            while idx < len(tokens):
+                if tokens[idx] == "iburst":
+                    details["iburst"] = True
+                    idx += 1
+
+                elif tokens[idx] == "key":
+                    details["key_id"] = int(tokens[idx + 1])
+                    idx += 2
+
+                elif tokens[idx] == "local-interface":
+                    interfaces = self.get_interfaces_ip()
+                    intf = tokens[idx + 1]
+                    if family == 6 and interfaces[intf]["ipv6"]:
+                        details["source_address"] = list(
+                            interfaces[intf]["ipv6"].keys()
+                        )[0]
+                    elif interfaces[intf]["ipv4"]:
+                        details["source_address"] = list(
+                            interfaces[intf]["ipv4"].keys()
+                        )[0]
+                    elif interfaces[intf]["ipv6"]:
+                        details["source_address"] = list(
+                            interfaces[intf]["ipv6"].keys()
+                        )[0]
+                    idx += 2
+
+                elif tokens[idx] == "version":
+                    details["version"] = int(tokens[idx + 1])
+                    idx += 2
+
+                elif tokens[idx] == "prefer":
+                    details["prefer"] = True
+                    idx += 1
+
+            result[server_ip] = details
+
+        return result
 
     def get_ntp_stats(self):
         ntp_stats = []
@@ -1687,7 +1744,7 @@ def traceroute(
         commands = []
 
         if vrf:
-            commands.append("routing-context vrf {vrf}".format(vrf=vrf))
+            commands.append("cli vrf {vrf}".format(vrf=vrf))
 
         if source:
             source_opt = "-s {source}".format(source=source)
@@ -2193,7 +2250,7 @@ def ping(
         commands = []
 
         if vrf:
-            commands.append("routing-context vrf {vrf}".format(vrf=vrf))
+            commands.append("cli vrf {vrf}".format(vrf=vrf))
 
         command = "ping {}".format(destination)
         command += " timeout {}".format(timeout)

napalm/ios/ios.py

@@ -22,7 +22,6 @@
 from netmiko._telnetlib import telnetlib
 import tempfile
 import uuid
-from collections import defaultdict
 
 from netmiko import FileTransfer, InLineTransfer
 
@@ -2076,7 +2075,7 @@ def get_bgp_neighbors(self):
         return bgp_neighbor_data
 
     def get_bgp_neighbors_detail(self, neighbor_address=""):
-        bgp_detail = defaultdict(lambda: defaultdict(lambda: []))
+        bgp_detail = {}
 
         raw_bgp_sum = self._send_command("show ip bgp all sum").strip()
 
@@ -2227,6 +2226,14 @@ def get_bgp_neighbors_detail(self, neighbor_address=""):
                         "export_policy": bgp_neigh_afi["policy_out"],
                     }
                 )
+
+            vrf_name = details["routing_table"]
+            if vrf_name not in bgp_detail.keys():
+                bgp_detail[vrf_name] = {}
+            remote_as = details["remote_as"]
+            if remote_as not in bgp_detail[vrf_name].keys():
+                bgp_detail[vrf_name][remote_as] = []
+
             bgp_detail[details["routing_table"]][details["remote_as"]].append(details)
         return bgp_detail
 

napalm/nxos/nxos.py

@@ -1204,27 +1204,32 @@ def get_arp_table(self, vrf: str = "") -> List[models.ARPTableDict]:
             )
         return arp_table
 
-    def _get_ntp_entity(
-        self, peer_type: str
-    ) -> Dict[str, Union[models.NTPPeerDict, models.NTPServerDict]]:
-        ntp_entities: Dict[str, Union[models.NTPPeerDict, models.NTPServerDict]] = {}
+    def _filter_ntp_table(self, peer_type: str) -> List[str]:
+        ret = []
         command = "show ntp peers"
         ntp_peers_table = self._get_command_table(command, "TABLE_peers", "ROW_peers")
-
         for ntp_peer in ntp_peers_table:
             if ntp_peer.get("serv_peer", "").strip() != peer_type:
                 continue
             peer_addr = napalm.base.helpers.ip(ntp_peer.get("PeerIPAddress").strip())
-            # Ignore the type of the following line until NTP data is modelled
-            ntp_entities[peer_addr] = {}  # type: ignore
-
-        return ntp_entities
+            ret.append(peer_addr)
+        return ret
 
     def get_ntp_peers(self) -> Dict[str, models.NTPPeerDict]:
-        return self._get_ntp_entity("Peer")
+        ntp_entities: Dict[str, models.NTPPeerDict] = {}
+        peers = self._filter_ntp_table("Peer")
+        for peer_addr in peers:
+            ntp_entities[peer_addr] = {}
+
+        return ntp_entities
 
     def get_ntp_servers(self) -> Dict[str, models.NTPServerDict]:
-        return self._get_ntp_entity("Server")
+        ntp_entities: Dict[str, models.NTPServerDict] = {}
+        peers = self._filter_ntp_table("Server")
+        for peer_addr in peers:
+            ntp_entities[peer_addr] = {}
+
+        return ntp_entities
 
     def get_ntp_stats(self) -> List[models.NTPStats]:
         ntp_stats: List[models.NTPStats] = []

test/eos/mocked_data/test_get_ntp_servers/details/expected_result.json

@@ -0,0 +1,24 @@
+{
+  "1.2.3.4": {
+    "port": 123,
+    "version": 4,
+    "association_type": "SERVER",
+    "iburst": true,
+    "prefer": true,
+    "network_instance": "FOO",
+    "source_address": "",
+    "key_id": -1,
+    "address": "1.2.3.4"
+  },
+  "4.3.2.1": {
+    "port": 123,
+    "version": 4,
+    "association_type": "SERVER",
+    "iburst": false,
+    "prefer": false,
+    "network_instance": "FOO",
+    "source_address": "172.20.20.2",
+    "key_id": -1,
+    "address": "4.3.2.1"
+  }
+}

test/eos/mocked_data/test_get_ntp_servers/details/show_ip_interface.json

@@ -0,0 +1,48 @@
+{
+  "interfaces": {
+      "Management0": {
+          "name": "Management0",
+          "lineProtocolStatus": "up",
+          "interfaceStatus": "connected",
+          "mtu": 1500,
+          "interfaceAddressBrief": {
+              "ipAddr": {
+                  "address": "172.20.20.2",
+                  "maskLen": 24
+              }
+          },
+          "ipv4Routable240": false,
+          "ipv4Routable0": false,
+          "enabled": true,
+          "description": "",
+          "interfaceAddress": {
+              "primaryIp": {
+                  "address": "172.20.20.2",
+                  "maskLen": 24
+              },
+              "secondaryIps": {},
+              "secondaryIpsOrderedList": [],
+              "virtualIp": {
+                  "address": "0.0.0.0",
+                  "maskLen": 0
+              },
+              "virtualSecondaryIps": {},
+              "virtualSecondaryIpsOrderedList": [],
+              "broadcastAddress": "255.255.255.255",
+              "dhcp": false
+          },
+          "proxyArp": false,
+          "proxyArpAllowDefault": false,
+          "localProxyArp": false,
+          "gratuitousArp": false,
+          "routedAddr": "00:1c:73:7b:8c:1d",
+          "isVrrpBackup": false,
+          "vrf": "default",
+          "urpf": "disable",
+          "addresslessForwarding": "isInvalid",
+          "directedBroadcastEnabled": false,
+          "maxMssIngress": 0,
+          "maxMssEgress": 0
+      }
+  }
+}