Added lldp py test

feature/b2b/lldp/test_lldp_neighbors.py

@@ -0,0 +1,78 @@
+import logging as log
+import pytest
+from helpers.otg import otg
+
+@pytest.mark.all
+@pytest.mark.cpdp
+
+def test_lldp_neighbors():
+    test_const = {
+        "txMac":       "00:00:01:01:01:01",
+		"rxMac":       "00:00:01:01:01:02",
+		"holdTime":    120,
+		"advInterval": 5,
+		"pduCount":    2,
+    }
+
+    api = otg.OtgApi()
+    c = lldp_neighbor_config(api, test_const)
+
+    api.set_config(c)
+
+    api.start_protocols()
+
+    api.wait_for(
+        fn=lambda: lldp_neigbhors_metrics_ok(api, test_const),
+        fn_name="wait_for_lldp_metrics",
+        timeout_seconds=30,
+    )
+
+    api.wait_for(
+        fn=lambda: lldp_neighbors_ok(api, test_const),
+        fn_name="wait_for_lldp_neighbors",
+        timeout_seconds=30,
+    )
+
+
+def lldp_neighbor_config(api, tc):
+    c = api.api.config()
+
+    ptx = c.ports.add(name="ptx", location=api.test_config.otg_ports[0])
+    prx = c.ports.add(name="prx", location=api.test_config.otg_ports[1])
+
+    ly = c.layer1.add(name="ly", port_names=[ptx.name, prx.name])
+    ly.speed = api.test_config.otg_speed
+
+    lldp_Tx = c.lldp.add(name="lldp_tx")
+    lldp_Rx = c.lldp.add(name="lldp_rx")
+
+    lldp_Tx.hold_time = tc["holdTime"]
+    lldp_Tx.advertisement_interval = tc["advInterval"]
+    lldp_Tx.connection.port_name = ptx.name
+    lldp_Tx.chassis_id.mac_address_subtype.value = tc["txMac"]
+
+    lldp_Rx.hold_time = tc["holdTime"]
+    lldp_Rx.advertisement_interval = tc["advInterval"]
+    lldp_Rx.connection.port_name = prx.name
+    lldp_Rx.chassis_id.mac_address_subtype.value = tc["rxMac"]
+
+    log.info("Config:\n%s", c)
+    return c
+
+def lldp_neigbhors_metrics_ok(api, tc):
+    for m in api.get_lldp_metrics():
+        if (
+            m.frames_tx < tc["pduCount"]
+            or m.frames_rx < tc["pduCount"]
+        ):
+            return False
+    return True
+
+def lldp_neighbors_ok(api, tc):
+    count = 0
+    for n in api.get_lldp_neighbors():
+        for i in ["txMac", "rxMac"]:
+            if n.chassis_id_type == "mac_address" and n.chassis_id == tc[i]:
+                count += 1
+
+    return count == 2

helpers/otg/otg.py

@@ -434,7 +434,73 @@ def get_capture(self, port_name):
             return CapturedPackets(b)
         finally:
             self.timer("get_capture", start)
+    def get_lldp_metrics(self):
+        start = datetime.datetime.now()
+        try:
+            log.info("Getting lldp metrics ...")
+            req = self.api.metrics_request()
+            req.lldp.lldp_names = []
+            metrics = self.api.get_metrics(req).lldp_metrics
+            print(metrics)
+
+            tb = table.Table(
+                "LLDP Metrics",
+                [
+                    "Name",
+                    "Frames Tx",
+                    "Frames Rx",
+                ],
+                15,
+            )
 
+            for m in metrics:
+                tb.append_row(
+                    [
+                        m.name,
+                        m.frames_tx,
+                        m.frames_rx,
+                    ]
+                )
+
+            log.info(tb)
+            return metrics
+        finally:
+            self.timer("get_lldp_metrics", start)
+
+    def get_lldp_neighbors(self):
+        start = datetime.datetime.now()
+        try:
+            log.info("Getting IPv4 Neighbors ...")
+            req = self.api.states_request()
+            req.lldp_neighbors.lldp_names = []
+            neighbors = self.api.get_states(req).lldp_neighbors
+
+            tb = table.Table(
+                "IPv4 Neighbors",
+                [
+                    "LLDP Name",
+			        "Chassis ID",
+			        "Chassis ID Type",
+			        "System Name",
+                ],
+                20,
+            )
+
+            for n in neighbors:
+                tb.append_row(
+                    [
+                        n.lldp_name,
+                        "" if n.chassis_id is None else n.chassis_id,
+                        "" if n.chassis_id_type is None else n.chassis_id_type,
+                        "" if n.system_name is None else n.system_name,
+                    ]
+                )
+
+            log.info(tb)
+            return neighbors
+
+        finally:
+            self.timer("get_lldp_neighbors", start)
 
 class CapturedPackets(object):
     def __init__(self, pcap_bytes):
@@ -511,3 +577,5 @@ class CapturedPacket(object):
     def __init__(self, sequence, data):
         self.sequence = sequence
         self.data = data
+
+