netconf i/o is vendor-specific (merged conflicts)

brian_polling_manager/interface_stats/cli.py

@@ -9,8 +9,6 @@ import click
 import jsonschema
 from brian_polling_manager.interface_stats import vendors
 from brian_polling_manager.interface_stats.services import (
-    get_juniper_netconf,
-    get_netconf_from_source_dir,
     write_points_to_influx,
     write_points_to_stdout,
 )
@@ -85,14 +83,14 @@ def write_points(points: Iterable[dict], influx_params: dict, **kwargs):
 def get_netconf(router_name, vendor=Vendor.JUNIPER, **kwargs):
     source_dir = _APP_CONFIG_PARAMS.get("testing", {}).get("netconf-source-dir")
     if source_dir:
-        return get_netconf_from_source_dir(router_name, source_dir)
+        return juniper.get_netconf_interface_info_from_source_dir(router_name, source_dir)
 
     ssh_params = _APP_CONFIG_PARAMS[vendor.value]
-    return get_juniper_netconf(router_name, ssh_params, **kwargs)
+    return juniper.get_netconf_interface_info(router_name, ssh_params, **kwargs)
 
 
 def validate_router_hosts(
-    hostnames, vendor: Vendor, inprov_hosts=None, load_interfaces_=load_interfaces
+    hostnames: List[str], vendor: Vendor, inprov_hosts=None, load_interfaces_=load_interfaces
 ):
     if inprov_hosts is None:
         return True

brian_polling_manager/interface_stats/services.py

 import json
 import logging
-import pathlib
 import sys
 from typing import Callable, Iterable
 from urllib.parse import urlparse
 
-import ncclient.manager
 from influxdb import InfluxDBClient
-from lxml import etree
-from ncclient.devices.junos import JunosDeviceHandler
-from ncclient.xml_ import NCElement
 
 logger = logging.getLogger(__name__)
 
-DEFAULT_NETCONF_PORT = 830
-DEFAULT_NETCONF_CONNECT_TIMEOUT = 5  # seconds
 DEFAULT_INFLUX_BATCH_SIZE = 100
 DEFAULT_INFLUX_TIMEOUT = 5
 INFLUX_POINT = {
@@ -31,56 +24,6 @@ INFLUX_POINT = {
 }
 
 
-def get_juniper_netconf(router_name, ssh_params, timeout=10):
-    request = etree.Element("get-interface-information")
-    request.append(etree.Element("extensive"))
-    response = _rpc(
-        router_name,
-        ssh_params=ssh_params,
-        command=request,
-        manager_params={"timeout": timeout},
-    )
-    return _remove_ns(response)
-
-
-def _remove_ns(rpc_response):
-    junos_dev_handler = JunosDeviceHandler(
-        device_params={"name": "junos", "local": False}
-    )
-    return NCElement(
-        rpc_response, junos_dev_handler.transform_reply()
-    ).remove_namespaces(rpc_response.xml)
-
-
-def _rpc(
-    router_name: str,
-    ssh_params: dict,
-    manager_params: dict,
-    command: etree.Element,
-    port=DEFAULT_NETCONF_PORT,
-    timeout=DEFAULT_NETCONF_CONNECT_TIMEOUT,
-):
-    """
-    :param router_name: router fqdn
-    :param ssh_params: 'juniper' element from app config
-    :param command: an lxml etree element
-    :param port: the port to connect to
-    :param timeout: a timeout in seconds
-    :return: the router netconf rpc reponse as an lxml.Element
-    """
-    params = {
-        "host": router_name,
-        "port": port,
-        "timeout": timeout,
-        **ssh_params,
-        **manager_params,
-    }
-
-    with ncclient.manager.connect(**params) as router:
-        logger.debug(f"sending rpc command: {etree.tostring(command)}")
-        return router.rpc(command)
-
-
 def prepare_influx_params(influx_params, timeout=DEFAULT_INFLUX_TIMEOUT):
     url = urlparse(influx_params["hostname"])
     use_ssl = influx_params.get("ssl", url.scheme == "https")
@@ -96,6 +39,7 @@ def prepare_influx_params(influx_params, timeout=DEFAULT_INFLUX_TIMEOUT):
     }
 
 
+#
 def influx_client(influx_params, timeout=DEFAULT_INFLUX_TIMEOUT):
     """
     build the influx connection
@@ -108,15 +52,6 @@ def influx_client(influx_params, timeout=DEFAULT_INFLUX_TIMEOUT):
     return InfluxDBClient(**prepare_influx_params(influx_params, timeout))
 
 
-def get_netconf_from_source_dir(
-    router_name: str, source_dir: str, suffix="-interface-info.xml"
-):
-    file = pathlib.Path(source_dir) / f"{router_name}{suffix}"
-    if not file.is_file():
-        raise ValueError(f"file {file} is not a valid file")
-    return etree.fromstring(file.read_text())
-
-
 def write_points_to_influx(
     points: Iterable[dict],
     influx_params: dict,

brian_polling_manager/interface_stats/vendors/__init__.py

@@ -16,8 +16,6 @@ class Vendor(enum.Enum):
 
 
 __all__ = [
-    "brian_points",
-    "error_counters",
     "BRIAN_POINT_FIELDS_SCHEMA",
     "ERROR_POINT_FIELDS_SCHEMA",
     "PHYSICAL_INTERFACE_COUNTER_SCHEMA",

brian_polling_manager/interface_stats/vendors/juniper.py

 import logging
+import pathlib
+
+from lxml import etree
+import ncclient.manager
+from ncclient.devices.junos import JunosDeviceHandler
+from ncclient.xml_ import NCElement
 
 logger = logging.getLogger(__name__)
 
+DEFAULT_NETCONF_PORT = 830
+DEFAULT_NETCONF_CONNECT_TIMEOUT = 5  # seconds
 
 PHYSICAL_INTERFACE_COUNTERS = {
     "__defaults__": {"transform": int, "required": False},
@@ -195,3 +203,60 @@ def interface_counters(ifc_doc):
     yield from _logical_interface_counters(ifc_doc)
 
 
+def get_netconf_interface_info(router_name, ssh_params, timeout=10):
+    request = etree.Element("get-interface-information")
+    request.append(etree.Element("extensive"))
+    response = _rpc(
+        router_name,
+        ssh_params=ssh_params,
+        command=request,
+        manager_params={"timeout": timeout},
+    )
+    return _remove_ns(response)
+
+
+def _remove_ns(rpc_response):
+    junos_dev_handler = JunosDeviceHandler(
+        device_params={"name": "junos", "local": False}
+    )
+    return NCElement(
+        rpc_response, junos_dev_handler.transform_reply()
+    ).remove_namespaces(rpc_response.xml)
+
+
+def _rpc(
+    router_name: str,
+    ssh_params: dict,
+    manager_params: dict,
+    command: etree.Element,
+    port=DEFAULT_NETCONF_PORT,
+    timeout=DEFAULT_NETCONF_CONNECT_TIMEOUT,
+):
+    """
+    :param router_name: router fqdn
+    :param ssh_params: 'juniper' element from app config
+    :param command: an lxml etree element
+    :param port: the port to connect to
+    :param timeout: a timeout in seconds
+    :return: the router netconf rpc reponse as an lxml.Element
+    """
+    params = {
+        "host": router_name,
+        "port": port,
+        "timeout": timeout,
+        **ssh_params,
+        **manager_params,
+    }
+
+    with ncclient.manager.connect(**params) as router:
+        logger.debug(f"sending rpc command: {etree.tostring(command)}")
+        return router.rpc(command)
+
+
+def get_netconf_interface_info_from_source_dir(
+    router_name: str, source_dir: str, suffix="-interface-info.xml"
+):
+    file = pathlib.Path(source_dir) / f"{router_name}{suffix}"
+    if not file.is_file():
+        raise ValueError(f"file {file} is not a valid file")
+    return etree.fromstring(file.read_text())

brian_polling_manager/interface_stats/vendors/nokia.py

-def physical_interface_counters(ifc_doc): ...
-def logical_interface_counters(ifc_doc): ...
+
+
+def interface_counters(ifc_doc):
+    assert False, "TODO"

test/interface_stats/test_interface_stats.py

@@ -381,7 +381,7 @@ class TestGetJuniperNetConnf:
     def mocked_rpc(self, data_dir):
         raw_response = data_dir.joinpath(self.RAW_RESPONSE_FILE).read_text()
         with patch.object(
-            services, "_rpc", return_value=RPCReply(raw_response)
+            juniper, "_rpc", return_value=RPCReply(raw_response)
         ) as mock:
             yield mock