test/test_lldp.py

   1 from framework import VppTestCase, VppTestRunner
   2 import unittest
   3 from config import config
   4 from scapy.layers.l2 import Ether
   5 from scapy.contrib.lldp import (
   6     LLDPDUChassisID,
   7     LLDPDUPortID,
   8     LLDPDUTimeToLive,
   9     LLDPDUEndOfLLDPDU,
  10     LLDPDU,
  11 )
  12
  13
  14 @unittest.skipIf("lldp" in config.excluded_plugins, "Exclude lldp plugin tests")
  15 class TestLldpCli(VppTestCase):
  16     """LLDP plugin tests [CLI]"""
  17
  18     @classmethod
  19     def setUpClass(cls):
  20         super(TestLldpCli, cls).setUpClass()
  21         try:
  22             cls.create_pg_interfaces(range(2))
  23             for i in cls.pg_interfaces:
  24                 i.config_ip4()
  25                 i.config_ip6()
  26                 i.resolve_arp()
  27                 i.admin_up()
  28         except Exception:
  29             cls.tearDownClass()
  30             raise
  31
  32     @classmethod
  33     def tearDownClass(cls):
  34         for i in cls.pg_interfaces:
  35             i.unconfig_ip4()
  36             i.unconfig_ip6()
  37             i.admin_down()
  38         super(TestLldpCli, cls).tearDownClass()
  39
  40     def create_frame(self, src_if):
  41         if src_if == self.pg0:
  42             chassis_id = "01:02:03:04:05:06"
  43             port_id = "07:08:09:0a:0b:0c"
  44         else:
  45             chassis_id = "11:12:13:14:15:16"
  46             port_id = "17:18:19:1a:1b:1c"
  47
  48         lldp_frame = (
  49             Ether(src=src_if.remote_mac, dst="01:80:C2:00:00:03")
  50             / LLDPDU()
  51             / LLDPDUChassisID(subtype=4, id=chassis_id)
  52             / LLDPDUPortID(subtype=3, id=port_id)
  53             / LLDPDUTimeToLive(ttl=120)
  54             / LLDPDUEndOfLLDPDU()
  55         )
  56
  57         return lldp_frame
  58
  59     def test_lldp_cli(self):
  60         """Enable, send frames, show, disable, verify"""
  61
  62         packets = self.create_frame(self.pg0)
  63         self.pg0.add_stream(packets)
  64         packets = self.create_frame(self.pg1)
  65         self.pg1.add_stream(packets)
  66
  67         self.vapi.cli("set lldp system-name VPP tx-hold 4 tx-interval 10")
  68         # configure everything to increase coverage
  69         self.vapi.cli(
  70             f"set interface lldp pg0 port-desc vtf:pg0 mgmt-ip4"
  71             f" {self.pg0.local_ip4} mgmt-ip6 {self.pg0.local_ip6} mgmt-oid '1234'"
  72         )
  73         self.vapi.cli("set interface lldp pg1 port-desc vtf:pg1")
  74
  75         self.pg_start()
  76
  77         reply = self.vapi.cli("show lldp")
  78         expected = [
  79             "01:02:03:04:05:06",
  80             "07:08:09:0a:0b:0c",
  81             "11:12:13:14:15:16",
  82             "17:18:19:1a:1b:1c",
  83         ]
  84         for entry in expected:
  85             self.assertIn(entry, reply)
  86
  87         # only checking for an output
  88         reply = self.vapi.cli("show lldp detail")
  89         self.assertIn("Local Interface name: pg0", reply)
  90         self.assertIn("Local Interface name: pg1", reply)
  91
  92         # disable LLDP on an interface and verify
  93         self.vapi.cli("set interface lldp pg0 disable")
  94         reply = self.vapi.cli("show lldp")
  95         self.assertNotIn("pg0", reply)
  96
  97
  98 class TestLldpVapi(VppTestCase):
  99     """LLDP plugin test [VAPI]"""
 100
 101     @classmethod
 102     def setUpClass(cls):
 103         super(TestLldpVapi, cls).setUpClass()
 104         try:
 105             cls.create_pg_interfaces(range(2))
 106             for i in cls.pg_interfaces:
 107                 i.config_ip4()
 108                 i.config_ip6()
 109                 i.resolve_arp()
 110                 i.admin_up()
 111         except Exception:
 112             cls.tearDownClass()
 113             raise
 114
 115     @classmethod
 116     def tearDownClass(cls):
 117         for i in cls.pg_interfaces:
 118             i.unconfig_ip4()
 119             i.unconfig_ip6()
 120             i.admin_down()
 121         super(TestLldpVapi, cls).tearDownClass()
 122
 123     def test_lldp_vapi(self):
 124         """Enable, show, disable, verify"""
 125         self.vapi.lldp_config(tx_hold=4, tx_interval=1, system_name="VAPI")
 126         self.vapi.sw_interface_set_lldp(
 127             sw_if_index=1,
 128             mgmt_ip4=self.pg0.local_ip4,
 129             port_desc="vtf:pg0",
 130         )
 131         self.vapi.sw_interface_set_lldp(
 132             sw_if_index=2,
 133             mgmt_ip4=self.pg1.local_ip4,
 134             port_desc="vtf:pg1",
 135             mgmt_ip6=self.pg1.local_ip6,
 136             mgmt_oid=b"1",
 137         )
 138
 139         # only check if LLDP gets enabled, functionality is tested in CLI class
 140         reply = self.vapi.cli("show lldp")
 141         self.assertIn("pg1", reply)
 142
 143         self.vapi.sw_interface_set_lldp(sw_if_index=2, enable=False)
 144         reply = self.vapi.cli("show lldp")
 145         self.assertNotIn("pg1", reply)
 146
 147
 148 if __name__ == "__main__":
 149     unittest.main(testRunner=VppTestRunner)