1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
|
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)
@unittest.skipIf("lldp" in config.excluded_plugins, "Exclude lldp plugin tests")
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)
|
