| adrianvillin | 485f7bc | 2023-10-26 14:26:18 +0200 | [diff] [blame] | 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) |