skeleton bgp parsing & unit test

inventory_provider/bgp.py

+import re
+
+def neighbors(router, routing_instances=["IAS"], group_expression=None):
+
+    for config in router["bgp"]["configuration"]:
+        for ri in config["routing-instances"]:
+            for inst in ri["instance"]:
+                if inst["name"]["data"] not in routing_instances:
+                    continue
+                for prot in inst["protocols"]:
+                    for bgp in prot.get("bgp", []):
+                        for g in bgp.get("group", []):
+                            if group_expression and not \
+                                    re.match(group_expression, g["name"]["data"]):
+                                continue
+                            for n in g["neighbor"]:
+                                yield n

requirements.txt

@@ -3,3 +3,5 @@ mysql-connector
 pysnmp
 jsonschema
 paramiko
+
+pytest

test/test_bgp_peers.py

+import json
+import jsonschema
+
+from inventory_provider import bgp
+
+CACHE_SCHEMA = {
+    "$schema": "http://json-schema.org/draft-07/schema#",
+    "type": "object",
+    "patternProperties": {
+        "^.*\\.geant\\.net$": {
+            "type": "object",
+            "properties": {
+                "bgp": {"type": "object"},
+                "vrr": {"type": "object"},
+                "interfaces": {"type": "object"},
+                "snmp-interfaces": {
+                    "type": "array",
+                    "items": {
+                        "type": "object",
+                        "properties": {
+                            "v4Address": {"type": "string"},
+                            "v4Mask": {"type": "string"},
+                            "v4InterfaceName": {"type": "string"},
+                            "v6Address": {"type": "string"},
+                            "v6Mask": {"type": "string"},
+                            "v6InterfaceName": {"type": "string"},
+                        },
+                        "additionalProperties": False
+                    }
+                }
+            },
+            "required": ["bgp", "vrr", "interfaces", "snmp-interfaces"],
+            "additionalProperties": False
+        }
+    },
+    "additionalProperties": False
+}
+
+
+def _validate_cached_data(f):
+    cache = json.loads(f.read())
+    jsonschema.validate(cache, CACHE_SCHEMA)
+    return cache
+
+
+def test_ipv4_neighbors():
+    with open("router-info.json") as f:
+        cache = _validate_cached_data(f)
+
+    k = "mx1.ams.nl.geant.net"
+    v = cache[k]
+    for n in bgp.neighbors(v, group_expression=r'^GEANT-IX[\s-].*'):
+        print(n)