test/test_lldp.py

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