emulate old interfaces/<hostname> route

inventory_provider/pyez_test.py

+# import os
+from jnpr.junos import Device
+# from inventory_provider import config
+
+from lxml import etree
+
+
+def load_config(hostname, ssh_params):
+
+    dev = Device(
+        host=hostname,
+        user="Monit0r",
+        ssh_private_key_file=ssh_params['private-key'])
+    dev.open()
+    # data = dev.rpc.get_config(options={'format': 'json'})
+    return dev.rpc.get_config()
+
+
+def load_interfaces(hostname, ssh_params):
+    data = load_config(hostname, ssh_params)
+
+    # print(etree.tostring(data, encoding='unicode'))
+
+    def _ifc_info(e):
+        name = e.find('name')
+        assert name is not None, "expected interface 'name' child element"
+        description = e.find('description')
+        return {
+            'name': name.text,
+            'description': description.text if description is not None else None
+        }
+
+    # for i in data.xpath('//configuration/groups/interfaces/interface'):
+    for i in data.xpath('//configuration/interfaces/interface'):
+        info = _ifc_info(i)
+        yield info
+        # ifc_name = i.xpath('./name/text()')
+        # assert ifc_name
+        # print(ifc_name[0])
+        for u in i.xpath('./unit'):
+            unit_info = _ifc_info(u)
+            unit_info['name'] = "%s.%s" % (info['name'], unit_info['name'])
+            yield unit_info
+            # print("%s.%s" % (ifc_name[0], u))
+    #
+    # print("-----routing-instances-----")
+    # for i in data.xpath("//configuration/routing-instances/instance/interface"):
+    #     print(i.xpath('./name/text()'))
+
+
+# if __name__ == "__main__":
+#     with open("config.json") as f:
+#         params = config.load(f)
+#     for r in params['routers']:
+#         netconf = _load_netconf(r['hostname'], params['ssh'])
+#         with open(os.path.join(TEST_DATA_DIR, "%s-netconf.xml" % r['hostname']), 'w') as f:
+#             f.write(etree.tostring(netconf, encoding='unicode'))

test/test_pyez_parsing.py

+import os
+from inventory_provider import pyez_test
+from lxml import etree
+
+TEST_DATA_DIR = os.path.join(os.path.dirname(__file__), 'data')
+
+class MockedJunosRpc(object):
+
+    def __init__(self, hostname):
+        filename = os.path.join(TEST_DATA_DIR, "%s-netconf.xml" % hostname)
+        self.config = etree.parse(filename)
+
+    def get_config(self):
+        return self.config
+
+
+class MockedJunosDevice(object):
+
+    def __init__(self, **kwargs):
+        self.rpc = MockedJunosRpc(kwargs['host'])
+
+    def open(self):
+        pass
+
+
+def test_nr1(mocker, data_config):
+    mocker.patch(
+        'inventory_provider.pyez_test.Device',
+        # 'inventory_provider.pyez_test.jnpr.junos.Device',
+        MockedJunosDevice)
+
+    import json
+    for r in data_config['routers']:
+        print(r['hostname'])
+        x = pyez_test.load_interfaces(r['hostname'], data_config['ssh'])
+        print(json.dumps(list(x), indent=4))