Add API to query full running config from a device, or running config…

… for a specific interface

docs/apiref/devices.rst

@@ -190,6 +190,32 @@ This will return both the generated configuration based on the template for
 this device type, and also a list of available vaiables that could be used
 in the template.
 
+Running config
+--------------
+
+To get the latest running config from the device you can use running_config:
+
+::
+
+  curl https://hostname/api/v1.0/device/<device_hostname>/running_config
+
+You can also specify to only get the running config for a specific interface:
+
+::
+
+  curl https://hostname/api/v1.0/device/<device_hostname>/running_config?interface=Ethernet1
+
+Example output:
+
+::
+
+   {
+       "status": "success",
+       "data": {
+           "config": "no switchport\nvrf MGMT\nip address 10.100.2.101/24\nno lldp transmit\nno lldp receive"
+       }
+   }
+
 View previous config
 --------------------
 

src/cnaas_nms/api/device.py

@@ -874,6 +874,35 @@ def get(self, hostname: str):
         return result
 
 
+class DeviceRunningConfigApi(Resource):
+    @jwt_required
+    @device_api.param("interface")
+    def get(self, hostname: str):
+        args = request.args
+        result = empty_result()
+        result["data"] = {"config": None}
+        if not Device.valid_hostname(hostname):
+            return empty_result(status="error", data="Invalid hostname specified"), 400
+
+        with sqla_session() as session:
+            dev: Device = session.query(Device).filter(Device.hostname == hostname).one_or_none()
+            if not dev:
+                return empty_result("error", "Device not found"), 404
+
+            try:
+                if "interface" in args:
+                    running_config = cnaas_nms.devicehandler.get.get_running_config_interface(
+                        session, hostname, args["interface"]
+                    )
+                else:
+                    running_config = cnaas_nms.devicehandler.get.get_running_config(hostname)
+            except Exception as e:
+                return empty_result("error", "Exception: {}".format(str(e))), 500
+
+        result["data"]["config"] = running_config
+        return result
+
+
 class DevicePreviousConfigApi(Resource):
     @jwt_required
     @device_api.param("job_id")
@@ -1166,6 +1195,7 @@ def post(self):
 device_api.add_resource(DeviceByIdApi, "/<int:device_id>")
 device_api.add_resource(DeviceByHostnameApi, "/<string:hostname>")
 device_api.add_resource(DeviceConfigApi, "/<string:hostname>/generate_config")
+device_api.add_resource(DeviceRunningConfigApi, "/<string:hostname>/running_config")
 device_api.add_resource(DevicePreviousConfigApi, "/<string:hostname>/previous_config")
 device_api.add_resource(DeviceApplyConfigApi, "/<string:hostname>/apply_config")
 device_api.add_resource(DeviceApi, "")

src/cnaas_nms/api/tests/test_device.py

@@ -263,6 +263,18 @@ def test_get_synchistory(self):
         self.assertEqual(result.status_code, 200, "Get synchistory for all devices failed")
         self.assertTrue("data" in result.json)
 
+    @pytest.mark.equipment
+    def test_get_running_config(self):
+        hostname = self.testdata["managed_dist"]
+        result = self.client.get(f"/api/v1.0/device/{hostname}/running_config")
+        self.assertEqual(result.status_code, 200, "Get running config failed")
+
+    @pytest.mark.equipment
+    def test_get_running_config_interface(self):
+        hostname = self.testdata["managed_dist"]
+        result = self.client.get(f"/api/v1.0/device/{hostname}/running_config", query_string={"interface": "Ethernet1"})
+        self.assertEqual(result.status_code, 200, "Get running config interface failed")
+
 
 if __name__ == "__main__":
     unittest.main()

src/cnaas_nms/devicehandler/get.py

@@ -2,6 +2,8 @@
 import re
 from typing import Dict, List, Optional
 
+from netutils.config import compliance
+from netutils.lib_mapper import NAPALM_LIB_MAPPER
 from nornir.core.filter import F
 from nornir.core.task import AggregatedResult
 from nornir_napalm.plugins.tasks import napalm_get
@@ -11,6 +13,7 @@
 from cnaas_nms.db.device import Device, DeviceType
 from cnaas_nms.db.device_vars import expand_interface_settings
 from cnaas_nms.db.interface import Interface, InterfaceConfigType, InterfaceError
+from cnaas_nms.db.session import sqla_session
 from cnaas_nms.tools.log import get_logger
 
 
@@ -19,14 +22,26 @@ def get_inventory():
     return nr.dict()["inventory"]
 
 
-def get_running_config(hostname):
+def get_running_config(hostname: str) -> Optional[str]:
     nr = cnaas_nms.devicehandler.nornir_helper.cnaas_init()
-    if hostname:
-        nr_filtered = nr.filter(name=hostname).filter(managed=True)
-    else:
-        nr_filtered = nr.filter(managed=True)
+    nr_filtered = nr.filter(name=hostname).filter(managed=True)
     nr_result = nr_filtered.run(task=napalm_get, getters=["config"])
-    return nr_result[hostname].result
+    if nr_result[hostname].failed:
+        raise nr_result[hostname][0].exception
+    else:
+        return nr_result[hostname].result["config"]["running"]
+
+
+def get_running_config_interface(session: sqla_session, hostname: str, interface: str) -> str:
+    running_config = get_running_config(hostname)
+    dev: Device = session.query(Device).filter(Device.hostname == hostname).one()
+    os_parser = compliance.parser_map[NAPALM_LIB_MAPPER.get(dev.platform)]
+    config_parsed = os_parser(running_config)
+    ret = []
+    for line in config_parsed.config_lines:
+        if f"interface {interface}" in line.parents:
+            ret.append(line.config_line.strip())
+    return "\n".join(ret)
 
 
 def calc_config_hash(hostname, config):

src/cnaas_nms/devicehandler/tests/test_get.py

@@ -89,6 +89,12 @@ def test_update_links(self):
             )
         pprint.pprint(new_links)
 
+    @pytest.mark.equipment
+    def test_get_running_config_interface(self):
+        with sqla_session() as session:
+            if_config: str = cnaas_nms.devicehandler.get.get_running_config_interface(session, "eosdist1", "Ethernet1")
+            assert if_config.strip(), "no config found"
+
 
 if __name__ == "__main__":
     unittest.main()