Merge branch 'develop' into develop

docs/requirements.txt

@@ -1,8 +1,8 @@
 urllib3==2.2.1 # https://github.com/readthedocs/readthedocs.org/issues/10290
-sphinx==7.2.6
+sphinx==7.3.7
 sphinx-rtd-theme==2.0.0
 invoke==2.2.0
-jinja2==3.1.3
+jinja2==3.1.4
 MarkupSafe==2.1.5
 pytest==8.2.1
-ansible==9.4.0
+ansible==9.6.0

napalm/base/base.py

@@ -639,7 +639,7 @@ def get_lldp_neighbors_detail(
 
     def get_bgp_config(
         self, group: str = "", neighbor: str = ""
-    ) -> models.BPGConfigGroupDict:
+    ) -> models.BGPConfigGroupDict:
         """
         Returns a dictionary containing the BGP configuration.
         Can return either the whole config, either the config only for a group or neighbor.

napalm/base/models.py

@@ -153,8 +153,8 @@
     "AFDict", {"sent_prefixes": int, "accepted_prefixes": int, "received_prefixes": int}
 )
 
-BPGConfigGroupDict = TypedDict(
-    "BPGConfigGroupDict",
+BGPConfigGroupDict = TypedDict(
+    "BGPConfigGroupDict",
     {
         "type": str,
         "description": str,
@@ -188,8 +188,8 @@
     },
 )
 
-BGPStateAdressFamilyDict = TypedDict(
-    "BGPStateAdressFamilyDict",
+BGPStateAddressFamilyDict = TypedDict(
+    "BGPStateAddressFamilyDict",
     {"received_prefixes": int, "accepted_prefixes": int, "sent_prefixes": int},
 )
 
@@ -203,7 +203,7 @@
         "is_enabled": bool,
         "description": str,
         "uptime": int,
-        "address_family": Dict[str, BGPStateAdressFamilyDict],
+        "address_family": Dict[str, BGPStateAddressFamilyDict],
     },
 )
 

napalm/base/test/base.py

@@ -323,7 +323,7 @@ def test_get_bgp_config(self):
         result = len(get_bgp_config) > 0
 
         for bgp_group in get_bgp_config.values():
-            result = result and self._test_model(models.BPGConfigGroupDict, bgp_group)
+            result = result and self._test_model(models.BGPConfigGroupDict, bgp_group)
             for bgp_neighbor in bgp_group.get("neighbors", {}).values():
                 result = result and self._test_model(
                     models.BGPConfigNeighborDict, bgp_neighbor

napalm/base/test/getters.py

@@ -249,7 +249,7 @@ def test_get_bgp_config(self, test_case):
         assert get_bgp_config == {} or len(get_bgp_config) > 0
 
         for bgp_group in get_bgp_config.values():
-            assert helpers.test_model(models.BPGConfigGroupDict, bgp_group)
+            assert helpers.test_model(models.BGPConfigGroupDict, bgp_group)
             for bgp_neighbor in bgp_group.get("neighbors", {}).values():
                 assert helpers.test_model(models.BGPConfigNeighborDict, bgp_neighbor)
 

napalm/eos/eos.py

@@ -34,11 +34,12 @@
 import pyeapi
 from pyeapi.eapilib import ConnectionError, EapiConnection
 from netmiko import ConfigInvalidException
+from typing import Dict
 
 # NAPALM base
 import napalm.base.helpers
 from napalm.base.netmiko_helpers import netmiko_args
-from napalm.base.base import NetworkDriver
+from napalm.base.base import NetworkDriver, models
 from napalm.base.utils import string_parsers
 from napalm.base.exceptions import (
     CommitError,
@@ -672,140 +673,84 @@ def get_interfaces_counters(self):
             )
         return interface_counters
 
-    def get_bgp_neighbors(self):
-        def get_re_group(res, key, default=None):
-            """Small helper to retrieve data from re match groups"""
-            try:
-                return res.group(key)
-            except KeyError:
-                return default
-
-        NEIGHBOR_FILTER = "vrf all | include IPv[46] (Unicast|6PE):.*[0-9]+ | grep -v ' IPv[46] Unicast:/.' | remote AS |^Local AS|Desc|BGP state |remote router ID"  # noqa
-        output_summary_cmds = self._run_commands(
-            ["show ipv6 bgp summary vrf all", "show ip bgp summary vrf all"],
-            encoding="json",
-        )
-        output_neighbor_cmds = self._run_commands(
+    def get_bgp_neighbors(self) -> Dict[str, models.BGPStateNeighborsPerVRFDict]:
+        cmd_outputs = self._run_commands(
             [
-                "show ip bgp neighbors " + NEIGHBOR_FILTER,
-                "show ipv6 bgp peers " + NEIGHBOR_FILTER,
+                "show ip bgp summary vrf all",
+                "show ipv6 bgp summary vrf all",
+                "show ip bgp neighbors vrf all",
+                "show ipv6 bgp peers vrf all",
             ],
-            encoding="text",
+            encoding="json",
         )
 
-        bgp_counters = defaultdict(lambda: dict(peers={}))
-        for summary in output_summary_cmds:
-            """
-            Json output looks as follows
-            "vrfs": {
-                "default": {
-                    "routerId": 1,
-                    "asn": 1,
-                    "peers": {
-                        "1.1.1.1": {
-                            "msgSent": 1,
-                            "inMsgQueue": 0,
-                            "prefixReceived": 3926,
-                            "upDownTime": 1449501378.418644,
-                            "version": 4,
-                            "msgReceived": 59616,
-                            "prefixAccepted": 3926,
-                            "peerState": "Established",
-                            "outMsgQueue": 0,
-                            "underMaintenance": false,
-                            "asn": 1
-                        }
-                    }
-                }
-            }
-            """
-            for vrf, vrf_data in summary["vrfs"].items():
-                bgp_counters[vrf]["router_id"] = vrf_data["routerId"]
-                for peer, peer_data in vrf_data["peers"].items():
-                    if peer_data["peerState"] == "Idle":
+        bgp_counters = defaultdict(
+            lambda: models.BGPStateNeighborsPerVRFDict(
+                peers=models.BGPStateNeighborDict()  # type: ignore
+            )  # type: ignore
+        )  # type: ignore
+        # Iterate IPv4 and IPv6 neighbor details
+        for cmd in cmd_outputs[2:]:
+            for vrf_name, vrf_data in cmd["vrfs"].items():
+                vrf = bgp_counters[vrf_name]
+                for peer in vrf_data["peerList"]:
+                    peer_ip = napalm.base.helpers.ip(peer["peerAddress"])
+                    v4_summary = cmd_outputs[0]["vrfs"][vrf_name]["peers"].get(
+                        peer_ip, {}
+                    )
+                    v6_summary = cmd_outputs[1]["vrfs"][vrf_name]["peers"].get(
+                        peer_ip, {}
+                    )
+                    local_as = napalm.base.helpers.as_number(peer["localAsn"])
+                    remote_as = napalm.base.helpers.as_number(peer["asn"])
+                    remote_id = napalm.base.helpers.ip(peer["routerId"])
+                    if peer["state"] == "Idle":
                         is_enabled = (
                             True
-                            if peer_data["peerStateIdleReason"] != "Admin"
+                            if peer["idleReason"] != "Administratively shut down"
                             else False
                         )
                     else:
                         is_enabled = True
-                    peer_info = {
-                        "is_up": peer_data["peerState"] == "Established",
-                        "is_enabled": is_enabled,
-                        "uptime": int(time.time() - float(peer_data["upDownTime"])),
-                        "description": peer_data.get("description", ""),
-                    }
-                    bgp_counters[vrf]["peers"][napalm.base.helpers.ip(peer)] = peer_info
-        lines = []
-        [lines.extend(x["output"].splitlines()) for x in output_neighbor_cmds]
-        while lines:
-            """
-            Raw output from the command looks like the following:
-
-              BGP neighbor is 1.1.1.1, remote AS 1, external link
-                Description: Very info such descriptive
-                BGP version 4, remote router ID 1.1.1.1, VRF my_vrf
-                BGP state is Idle, Administratively shut down
-                 IPv4 Unicast:         683        78
-                 IPv6 Unicast:           0         0
-              Local AS is 2, local router ID 2.2.2.2
-            """
-            neighbor_info = re.match(self._RE_BGP_INFO, lines.pop(0))
-            # this line can be either description or rid info
-            next_line = lines.pop(0)
-            desc = re.match(self._RE_BGP_DESC, next_line)
-            if desc is None:
-                rid_info = re.match(self._RE_BGP_RID_INFO, next_line)
-                desc = ""
-            else:
-                rid_info = re.match(self._RE_BGP_RID_INFO, lines.pop(0))
-                desc = desc.group("description")
-            lines.pop(0)
-            v4_stats = re.match(self._RE_BGP_PREFIX, lines.pop(0))
-            v6_stats = re.match(self._RE_BGP_PREFIX, lines.pop(0))
-            local_as = re.match(self._RE_BGP_LOCAL, lines.pop(0))
-            data = {
-                "remote_as": napalm.base.helpers.as_number(neighbor_info.group("as")),
-                "remote_id": napalm.base.helpers.ip(
-                    get_re_group(rid_info, "rid", "0.0.0.0")
-                ),
-                "local_as": napalm.base.helpers.as_number(local_as.group("as")),
-                "description": str(desc),
-                "address_family": {
-                    "ipv4": {
-                        "sent_prefixes": int(get_re_group(v4_stats, "sent", -1)),
-                        "received_prefixes": int(
-                            get_re_group(v4_stats, "received", -1)
+                    is_up = peer["state"] == "Established"
+                    description = peer.get("description", "")
+                    uptime = int(peer.get("establishedTime", -1))
+                    v4: models.BGPStateAddressFamilyDict = {
+                        "received_prefixes": peer["prefixesReceived"],
+                        "accepted_prefixes": (
+                            v4_summary["prefixAccepted"] if v4_summary else 0
                         ),
-                        "accepted_prefixes": -1,
-                    },
-                    "ipv6": {
-                        "sent_prefixes": int(get_re_group(v6_stats, "sent", -1)),
-                        "received_prefixes": int(
-                            get_re_group(v6_stats, "received", -1)
+                        "sent_prefixes": peer["prefixesSent"],
+                    }
+                    v6: models.BGPStateAddressFamilyDict = {
+                        "received_prefixes": peer["v6PrefixesReceived"],
+                        "accepted_prefixes": (
+                            v6_summary["prefixAccepted"] if v6_summary else 0
                         ),
-                        "accepted_prefixes": -1,
-                    },
-                },
-            }
-            peer_addr = napalm.base.helpers.ip(neighbor_info.group("neighbor"))
-            vrf = rid_info.group("vrf")
-            if peer_addr not in bgp_counters[vrf]["peers"]:
-                bgp_counters[vrf]["peers"][peer_addr] = {
-                    "is_up": False,  # if not found, means it was not found in the oper stats
-                    # i.e. neighbor down,
-                    "uptime": 0,
-                    "is_enabled": True,
-                }
-            if (
-                "description" in bgp_counters[vrf]["peers"][peer_addr]
-                and not data["description"]
-            ):
-                data["description"] = bgp_counters[vrf]["peers"][peer_addr][
-                    "description"
-                ]
-            bgp_counters[vrf]["peers"][peer_addr].update(data)
+                        "sent_prefixes": peer["v6PrefixesSent"],
+                    }
+                    peer_data: models.BGPStateNeighborDict = {
+                        "local_as": local_as,
+                        "remote_as": remote_as,
+                        "remote_id": remote_id,
+                        "is_up": is_up,
+                        "is_enabled": is_enabled,
+                        "description": description,
+                        "uptime": uptime,
+                        "address_family": {
+                            "ipv4": v4,
+                            "ipv6": v6,
+                        },
+                    }
+                    vrf["peers"][peer_ip] = peer_data
+
+        # Iterate IPv4 and IPv6 summary details for router-id assignment
+        for cmd in cmd_outputs[:2]:
+            for vrf_name, vrf_data in cmd["vrfs"].items():
+                bgp_counters[vrf_name]["router_id"] = napalm.base.helpers.ip(
+                    vrf_data["routerId"]
+                )
+
         if "default" in bgp_counters:
             bgp_counters["global"] = bgp_counters.pop("default")
         return dict(bgp_counters)

requirements-dev.txt

@@ -1,5 +1,5 @@
-black==24.3.0
-coveralls==3.3.1
+black==24.4.2
+coveralls==4.0.1
 ddt==1.7.2
 flake8-import-order==0.18.2
 pytest==8.2.1
@@ -8,10 +8,10 @@ pytest-json-report==1.5.0
 pyflakes==3.2.0
 pylama==8.4.1
 mock==5.1.0
-mypy==1.9.0
+mypy==1.10.0
 types-PyYAML==6.0.12.20240311
-types-requests==2.31.0.20240406
-types-six==1.16.21.20240311
-types-setuptools==69.5.0.20240519
+types-requests==2.32.0.20240521
+types-six==1.16.21.20240513
+types-setuptools==69.5.0.20240522
 ttp==0.9.5
 ttp_templates==0.3.6