1 files changed, 150 insertions, 0 deletions
diff --git a/test/test_lldp.py b/test/test_lldp.py
new file mode 100644
index 00000000000..0a69be7c7b1
--- /dev/null
+++ b/test/test_lldp.py
@@ -0,0 +1,150 @@
+from asfframework import VppTestRunner
+from framework import VppTestCase
+import unittest
+from config import config
+from scapy.layers.l2 import Ether
+from scapy.contrib.lldp import (
+    LLDPDUChassisID,
+    LLDPDUPortID,
+    LLDPDUTimeToLive,
+    LLDPDUEndOfLLDPDU,
+    LLDPDU,
+)
+
+
+@unittest.skipIf("lldp" in config.excluded_plugins, "Exclude lldp plugin tests")
+class TestLldpCli(VppTestCase):
+    """LLDP plugin tests [CLI]"""
+
+    @classmethod
+    def setUpClass(cls):
+        super(TestLldpCli, cls).setUpClass()
+        try:
+            cls.create_pg_interfaces(range(2))
+            for i in cls.pg_interfaces:
+                i.config_ip4()
+                i.config_ip6()
+                i.resolve_arp()
+                i.admin_up()
+        except Exception:
+            cls.tearDownClass()
+            raise
+
+    @classmethod
+    def tearDownClass(cls):
+        for i in cls.pg_interfaces:
+            i.unconfig_ip4()
+            i.unconfig_ip6()
+            i.admin_down()
+        super(TestLldpCli, cls).tearDownClass()
+
+    def create_frame(self, src_if):
+        if src_if == self.pg0:
+            chassis_id = "01:02:03:04:05:06"
+            port_id = "07:08:09:0a:0b:0c"
+        else:
+            chassis_id = "11:12:13:14:15:16"
+            port_id = "17:18:19:1a:1b:1c"
+
+        lldp_frame = (
+            Ether(src=src_if.remote_mac, dst="01:80:C2:00:00:03")
+            / LLDPDU()
+            / LLDPDUChassisID(subtype=4, id=chassis_id)
+            / LLDPDUPortID(subtype=3, id=port_id)
+            / LLDPDUTimeToLive(ttl=120)
+            / LLDPDUEndOfLLDPDU()
+        )
+
+        return lldp_frame
+
+    def test_lldp_cli(self):
+        """Enable, send frames, show, disable, verify"""
+
+        packets = self.create_frame(self.pg0)
+        self.pg0.add_stream(packets)
+        packets = self.create_frame(self.pg1)
+        self.pg1.add_stream(packets)
+
+        self.vapi.cli("set lldp system-name VPP tx-hold 4 tx-interval 10")
+        # configure everything to increase coverage
+        self.vapi.cli(
+            f"set interface lldp pg0 port-desc vtf:pg0 mgmt-ip4"
+            f" {self.pg0.local_ip4} mgmt-ip6 {self.pg0.local_ip6} mgmt-oid '1234'"
+        )
+        self.vapi.cli("set interface lldp pg1 port-desc vtf:pg1")
+
+        self.pg_start()
+
+        reply = self.vapi.cli("show lldp")
+        expected = [
+            "01:02:03:04:05:06",
+            "07:08:09:0a:0b:0c",
+            "11:12:13:14:15:16",
+            "17:18:19:1a:1b:1c",
+        ]
+        for entry in expected:
+            self.assertIn(entry, reply)
+
+        # only checking for an output
+        reply = self.vapi.cli("show lldp detail")
+        self.assertIn("Local Interface name: pg0", reply)
+        self.assertIn("Local Interface name: pg1", reply)
+
+        # disable LLDP on an interface and verify
+        self.vapi.cli("set interface lldp pg0 disable")
+        reply = self.vapi.cli("show lldp")
+        self.assertNotIn("pg0", reply)
+
+
+class TestLldpVapi(VppTestCase):
+    """LLDP plugin test [VAPI]"""
+
+    @classmethod
+    def setUpClass(cls):
+        super(TestLldpVapi, cls).setUpClass()
+        try:
+            cls.create_pg_interfaces(range(2))
+            for i in cls.pg_interfaces:
+                i.config_ip4()
+                i.config_ip6()
+                i.resolve_arp()
+                i.admin_up()
+        except Exception:
+            cls.tearDownClass()
+            raise
+
+    @classmethod
+    def tearDownClass(cls):
+        for i in cls.pg_interfaces:
+            i.unconfig_ip4()
+            i.unconfig_ip6()
+            i.admin_down()
+        super(TestLldpVapi, cls).tearDownClass()
+
+    def test_lldp_vapi(self):
+        """Enable, show, disable, verify"""
+        self.vapi.lldp_config(tx_hold=4, tx_interval=1, system_name="VAPI")
+        self.vapi.sw_interface_set_lldp(
+            sw_if_index=1,
+            mgmt_ip4=self.pg0.local_ip4,
+            port_desc="vtf:pg0",
+        )
+        self.vapi.sw_interface_set_lldp(
+            sw_if_index=2,
+            mgmt_ip4=self.pg1.local_ip4,
+            port_desc="vtf:pg1",
+            mgmt_ip6=self.pg1.local_ip6,
+            mgmt_oid=b"1",
+        )
+
+        # only check if LLDP gets enabled, functionality is tested in CLI class
+        reply = self.vapi.cli("show lldp")
+        self.assertIn("pg1", reply)
+
+        self.vapi.sw_interface_set_lldp(sw_if_index=2, enable=False)
+        reply = self.vapi.cli("show lldp")
+        self.assertNotIn("pg1", reply)
+
+
+if __name__ == "__main__":
+    unittest.main(testRunner=VppTestRunner)