handle duplicate keys in juniper json output

inventory_provider/bgp.py → inventory_provider/juniper.py

@@ -15,3 +15,17 @@ def neighbors(router, routing_instances=["IAS"], group_expression=None):
                                 continue
                             for n in g["neighbor"]:
                                 yield n
+
+
+def interfaces(router):
+    for ifc_info in router["interfaces"]["interface-information"]:
+        for ifc_list in ifc_info["logical-interface"]:
+            for ifc in ifc_list:
+                yield {
+                    "name": ifc["name"][0]["data"],
+                    "status": "%s/%s" % (
+                        ifc["admin-status"][0]["data"],
+                        ifc["oper-status"][0]["data"]),
+                    "description": ifc["description"][0]["data"],
+                    "type": "logical"
+                }
\ No newline at end of file

inventory_provider/router_interfaces.py

@@ -132,7 +132,7 @@ def load_routers(config_file):
     for line in config_file:
         m = re.match(r'^([a-z\d]+\.[a-z\d]{3,4}\.[a-z\d]{2}\.(geant|eumedconnect)\d*\.net)\s*=([^,]+)\s*,(.*)\s*$', line)
         if not m:
-            logging.warning("malformed config file line: '%s'" + line.strip())
+            logging.warning("malformed config file line: '%s'" % line.strip())
             continue
         yield {
             "hostname": m.group(1),
@@ -255,6 +255,18 @@ def exec_router_commands_json(router, ssh_params, commands):
         _, stdout, _ = ssh.exec_command("set cli screen-length 0")
         assert stdout.channel.recv_exit_status() == 0
 
+        def _dups_to_list(pairs):
+            counter_map = {}
+            for k, v in pairs:
+                counter_map.setdefault(k, []).append(v)
+            result = {}
+            for k, v in counter_map.items():
+                if len(v) == 1:
+                    result[k] = v[0]
+                else:
+                    result[k] = v
+            return result
+
         for c in commands:
             juniper_logger.debug("command: '%s'" % (c + " | display json"))
             _, stdout, _ = ssh.exec_command(c + " | display json")
@@ -263,9 +275,9 @@ def exec_router_commands_json(router, ssh_params, commands):
             output = stdout.read()
             if output:
                 juniper_logger.debug("%r output: [%d] %r" % (router, len(output), output[:20]))
-                yield json.loads(output)
+                yield json.loads(output, object_pairs_hook=_dups_to_list)
             else:
-                juniper_logger.debug("%r output empty")
+                juniper_logger.debug("%r output empty" % router)
                 yield {}
 
 

test/test_bgp_peers.py

@@ -2,8 +2,9 @@ import json
 import re
 
 import jsonschema
+import pytest
 
-from inventory_provider import bgp
+from inventory_provider import juniper
 
 CACHE_SCHEMA = {
     "$schema": "http://json-schema.org/draft-07/schema#",
@@ -246,8 +247,10 @@ set routing-instances IAS protocols bgp group GEANT-IXv6-AMS-IX neighbor 2001:7f
 """
 
 
-def _validate_cached_data(f):
-    cache = json.loads(f.read())
+@pytest.fixture
+def cached_test_data():
+    with open("router-info.json") as f:
+        cache = json.loads(f.read())
     jsonschema.validate(cache, CACHE_SCHEMA)
     return cache
 
@@ -274,15 +277,13 @@ def _old_data(s):
         }
 
 
-def test_ipv4_neighbors():
+def test_ipv4_neighbors(cached_test_data):
     old_v4_data = dict((x["neighbor"], x) for x in _old_data(OLD_STYLE_V4_DATA))
-    with open("router-info.json") as f:
-        cache = _validate_cached_data(f)
 
     k = "mx1.ams.nl.geant.net"
-    v = cache[k]
+    v = cached_test_data[k]
 
-    neighbors = list(bgp.neighbors(v, group_expression=r'^GEANT-IX[\s-].*'))
+    neighbors = list(juniper.neighbors(v, group_expression=r'^GEANT-IX[\s-].*'))
     assert len(neighbors) == len(old_v4_data)
     for n in neighbors:
         address = n["name"]["data"]
@@ -290,17 +291,32 @@ def test_ipv4_neighbors():
         assert old_v4_data[address]["description"] == description
 
 
-def test_ipv6_neighbors():
+def test_ipv6_neighbors(cached_test_data):
     old_v6_data = dict((x["neighbor"], x) for x in _old_data(OLD_STYLE_V6_DATA))
-    with open("router-info.json") as f:
-        cache = _validate_cached_data(f)
 
     k = "mx1.ams.nl.geant.net"
-    v = cache[k]
+    v = cached_test_data[k]
 
-    neighbors = list(bgp.neighbors(v, group_expression=r'^GEANT-IXv6[\s-].*'))
+    neighbors = list(juniper.neighbors(v, group_expression=r'^GEANT-IXv6[\s-].*'))
     assert len(neighbors) == len(old_v6_data)
     for n in neighbors:
         address = n["name"]["data"]
         description = n["description"][0]["data"]
         assert old_v6_data[address]["description"] == description
+
+def test_interfaces_tmp(cached_test_data):
+
+    k = "mx1.ams.nl.geant.net"
+    v = cached_test_data[k]
+    # print(v["interfaces"])
+    # with open("/tmp/mx1.ams.nl.geant.net.interfaces.json", "w") as f:
+    #     f.write(json.dumps(v["interfaces"]))
+
+    for ifc in juniper.interfaces(v):
+        print(ifc)
+    #
+    # interfaces = {}
+    # for k, v in cached_test_data.items():
+    #     interfaces[k] = v["interfaces"]
+    # with open("/tmp/interfaces.json", "w") as f:
+    #     f.write(json.dumps(interfaces,sort_keys=True,indent=2, separators=(',', ': ')))