Merge pull request #9 from jillesca/isis_fix

Fix: isis metrics when only one neighbor is present. Enhance: Error handling.

ncpeek/client.py

@@ -1,3 +1,4 @@
+import sys
 from typing import Optional, Union
 from dataclasses import dataclass
 from ncpeek.utils.text_utils import (
@@ -73,22 +74,19 @@ def _run(self) -> str:
 
     def _process_device(self, device: dict) -> dict:
         """Processes a single device operation and parsing its reply."""
-        try:
-            device = NetconfDevice(**device)
-            rpc = NetconfSession(
-                device=device,
-                netconf_filter=self._netconf_filter,
-                operation=self._operation,
-            )
-            data_dict = convert_xml_to_dict(xml_string=rpc.reply())
-            parser = get_parser(netconf_filter=self._filter_id)
-            parsed_data = parser.parse(
-                data_to_parse=data_dict,
-                device=device,
-                netconf_filter_id=self._filter_id,
-            )
-        except Exception as err:
-            parsed_data = [{"error": f"{err=}"}]
+        device = NetconfDevice(**device)
+        rpc = NetconfSession(
+            device=device,
+            netconf_filter=self._netconf_filter,
+            operation=self._operation,
+        )
+        data_dict = convert_xml_to_dict(xml_string=rpc.reply())
+        parser = get_parser(netconf_filter=self._filter_id)
+        parsed_data = parser.parse(
+            data_to_parse=data_dict,
+            device=device,
+            netconf_filter_id=self._filter_id,
+        )
         return parsed_data
 
 
@@ -99,6 +97,7 @@ def cli() -> None:
         print(client.execute_cli())
     except Exception as err:
         print(f"Error found: {err=}")
+        sys.exit(2)
 
 
 if __name__ == "__main__":

ncpeek/parsers/cisco_ios_xe_isis_oper.py

@@ -52,9 +52,13 @@ def _group_data_by_system_id(self, data: Dict) -> Dict:
             Dict: The grouped data.
         """
         grouped_data = defaultdict(list)
-        for neighbor in data["data"]["isis-oper-data"]["isis-instance"][
+        neighbors = data["data"]["isis-oper-data"]["isis-instance"][
             "isis-neighbor"
-        ]:
+        ]
+
+        if isinstance(neighbors, dict):
+            neighbors = [neighbors]
+        for neighbor in neighbors:
             grouped_data[neighbor["system-id"]].append(
                 self._create_neighbor_dict(neighbor)
             )
@@ -114,12 +118,14 @@ def _prepare_neighbor_data(self, grouped_data: Dict) -> List[Dict]:
             {
                 "system-id": system_id,
                 "local_interfaces_status": interfaces,
-                "neighbor_status": 1
-                if any(
-                    interface["isis_status"] == "isis-adj-up"
-                    for interface in interfaces
-                )
-                else 0,
+                "neighbor_status": (
+                    1
+                    if any(
+                        interface["isis_status"] == "isis-adj-up"
+                        for interface in interfaces
+                    )
+                    else 0
+                ),
             }
             for system_id, interfaces in grouped_data.items()
         ]

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "ncpeek"
-version = "0.1.3"
+version = "0.1.4"
 description = "ncpeek (short for netconf peek) is a netconf client that retrieves data using the ncclient library."
 authors = ["Jesus Illescas <[email protected]>"]
 readme = ["README.md"]

tests/test_parsers/test_cisco_ios_xe_isis_oper.py

@@ -9,9 +9,10 @@
     username="username",
 )
 parser = ISISStatsIOSXEParser(device=device)
-filter_id = "filter1"
 
-data_to_parse = {
+FILTER_ID = "filter1"
+
+DATA_TO_PARSE = {
     "data": {
         "isis-oper-data": {
             "isis-instance": {
@@ -48,7 +49,7 @@
 
 
 def test_parse():
-    parsed_data = parser.parse(data_to_parse, device, filter_id)
+    parsed_data = parser.parse(DATA_TO_PARSE, device, FILTER_ID)
     expected_data = [
         {
             "device": "device-1",
@@ -96,7 +97,7 @@ def test_parse():
 
 
 def test_group_data_by_system_id():
-    grouped_data = parser._group_data_by_system_id(data_to_parse)
+    grouped_data = parser._group_data_by_system_id(DATA_TO_PARSE)
     expected_data = {
         "00:00:00:00:00:0a": [
             {
@@ -126,3 +127,39 @@ def test_group_data_by_system_id():
     }
 
     assert grouped_data == expected_data
+
+
+SINGLE_ISIS_NEIGHBOR = {
+    "data": {
+        "isis-oper-data": {
+            "isis-instance": {
+                "isis-neighbor": {
+                    "holdtime": "22",
+                    "if-name": "GigabitEthernet3",
+                    "ipv4-address": "10.4.4.1",
+                    "level": "isis-level-1",
+                    "state": "isis-adj-up",
+                    "system-id": "00:00:00:00:00:0b",
+                },
+            },
+        },
+    }
+}
+
+
+def test_group_data_by_system_id_single_neighbor():
+    grouped_data = parser._group_data_by_system_id(SINGLE_ISIS_NEIGHBOR)
+
+    expected_data = {
+        "00:00:00:00:00:0b": [
+            {
+                "interface_name": "GigabitEthernet3",
+                "level": "isis-level-1",
+                "local_ipv4_address": "10.4.4.1",
+                "isis_status": "isis-adj-up",
+                "holdtime": "22",
+            }
+        ]
+    }
+
+    assert grouped_data == expected_data