6 from scapy.packet import Raw
7 from scapy.layers.l2 import Ether, Dot1Q
8 from scapy.layers.inet import IP, UDP
10 from framework import VppTestCase, VppTestRunner
11 from util import Host, ppp
12 from vpp_sub_interface import VppDot1QSubint, VppDot1ADSubint
15 class TestL2bd(VppTestCase):
16 """ L2BD Test Case """
21 Perform standard class setup (defined by class method setUpClass in
22 class VppTestCase) before running the test case, set test case related
23 variables and configure VPP.
25 :var int bd_id: Bridge domain ID.
26 :var int mac_entries_count: Number of MAC entries for bridge-domain to
28 :var int dot1q_tag: VLAN tag for dot1q sub-interface.
29 :var int dot1ad_sub_id: SubID of dot1ad sub-interface.
30 :var int dot1ad_outer_tag: VLAN S-tag for dot1ad sub-interface.
31 :var int dot1ad_inner_tag: VLAN C-tag for dot1ad sub-interface.
32 :var int sl_pkts_per_burst: Number of packets in burst for single-loop
34 :var int dl_pkts_per_burst: Number of packets in burst for dual-loop
37 super(TestL2bd, cls).setUpClass()
41 cls.mac_entries_count = 100
42 # cls.dot1q_sub_id = 100
44 cls.dot1ad_sub_id = 20
45 cls.dot1ad_outer_tag = 200
46 cls.dot1ad_inner_tag = 300
47 cls.sl_pkts_per_burst = 2
48 cls.dl_pkts_per_burst = 257
51 # create 3 pg interfaces
52 cls.create_pg_interfaces(range(3))
54 # create 2 sub-interfaces for pg1 and pg2
55 cls.sub_interfaces = [
56 VppDot1QSubint(cls, cls.pg1, cls.dot1q_tag),
57 VppDot1ADSubint(cls, cls.pg2, cls.dot1ad_sub_id,
58 cls.dot1ad_outer_tag, cls.dot1ad_inner_tag)]
60 # packet flows mapping pg0 -> pg1, pg2, etc.
62 cls.flows[cls.pg0] = [cls.pg1, cls.pg2]
63 cls.flows[cls.pg1] = [cls.pg0, cls.pg2]
64 cls.flows[cls.pg2] = [cls.pg0, cls.pg1]
67 cls.pg_if_packet_sizes = [64, 512, 1518, 9018]
68 cls.sub_if_packet_sizes = [64, 512, 1518 + 4, 9018 + 4]
70 cls.interfaces = list(cls.pg_interfaces)
71 cls.interfaces.extend(cls.sub_interfaces)
73 # Create BD with MAC learning enabled and put interfaces and
74 # sub-interfaces to this BD
75 for pg_if in cls.pg_interfaces:
76 sw_if_index = pg_if.sub_if.sw_if_index \
77 if hasattr(pg_if, 'sub_if') else pg_if.sw_if_index
78 cls.vapi.sw_interface_set_l2_bridge(rx_sw_if_index=sw_if_index,
81 # setup all interfaces
82 for i in cls.interfaces:
85 # mapping between packet-generator index and lists of test hosts
86 cls.hosts_by_pg_idx = dict()
88 # create test host entries and inject packets to learn MAC entries
89 # in the bridge-domain
90 cls.create_hosts_and_learn(cls.mac_entries_count)
91 cls.logger.info(cls.vapi.ppcli("show l2fib"))
94 super(TestL2bd, cls).tearDownClass()
99 Clear trace and packet infos before running each test.
101 super(TestL2bd, self).setUp()
102 self.reset_packet_infos()
106 Show various debug prints after each test.
108 super(TestL2bd, self).tearDown()
109 if not self.vpp_dead:
110 self.logger.info(self.vapi.ppcli("show l2fib verbose"))
111 self.logger.info(self.vapi.ppcli("show bridge-domain %s detail" %
115 def create_hosts_and_learn(cls, count):
117 Create required number of host MAC addresses and distribute them among
118 interfaces. Create host IPv4 address for every host MAC address. Create
119 L2 MAC packet stream with host MAC addresses per interface to let
120 the bridge domain learn these MAC addresses.
122 :param count: Integer number of hosts to create MAC/IPv4 addresses for.
124 n_int = len(cls.pg_interfaces)
125 macs_per_if = count / n_int
127 for pg_if in cls.pg_interfaces:
129 start_nr = macs_per_if * i
130 end_nr = count if i == (n_int - 1) else macs_per_if * (i + 1)
131 cls.hosts_by_pg_idx[pg_if.sw_if_index] = []
132 hosts = cls.hosts_by_pg_idx[pg_if.sw_if_index]
134 for j in range(start_nr, end_nr):
136 "00:00:00:ff:%02x:%02x" % (pg_if.sw_if_index, j),
137 "172.17.1%02x.%u" % (pg_if.sw_if_index, j))
138 packet = (Ether(dst="ff:ff:ff:ff:ff:ff", src=host.mac))
140 if hasattr(pg_if, 'sub_if'):
141 packet = pg_if.sub_if.add_dot1_layer(packet)
142 packets.append(packet)
143 pg_if.add_stream(packets)
144 cls.logger.info("Sending broadcast eth frames for MAC learning")
147 def create_stream(self, src_if, packet_sizes, packets_per_burst):
149 Create input packet stream for defined interface.
151 :param object src_if: Interface to create packet stream for.
152 :param list packet_sizes: List of required packet sizes.
153 :param int packets_per_burst: Number of packets in burst.
154 :return: Stream of packets.
157 for i in range(0, packets_per_burst):
158 dst_if = self.flows[src_if][i % 2]
159 dst_host = random.choice(self.hosts_by_pg_idx[dst_if.sw_if_index])
160 src_host = random.choice(self.hosts_by_pg_idx[src_if.sw_if_index])
161 pkt_info = self.create_packet_info(src_if, dst_if)
162 payload = self.info_to_payload(pkt_info)
163 p = (Ether(dst=dst_host.mac, src=src_host.mac) /
164 IP(src=src_host.ip4, dst=dst_host.ip4) /
165 UDP(sport=1234, dport=1234) /
167 pkt_info.data = p.copy()
168 if hasattr(src_if, 'sub_if'):
169 p = src_if.sub_if.add_dot1_layer(p)
170 size = random.choice(packet_sizes)
171 self.extend_packet(p, size)
175 def verify_capture(self, pg_if, capture):
177 Verify captured input packet stream for defined interface.
179 :param object pg_if: Interface to verify captured packet stream for.
180 :param list capture: Captured packet stream.
183 for i in self.pg_interfaces:
184 last_info[i.sw_if_index] = None
185 dst_sw_if_index = pg_if.sw_if_index
186 for packet in capture:
187 payload_info = self.payload_to_info(packet[Raw])
188 src_sw_if_index = payload_info.src
190 for ifc in self.pg_interfaces:
192 if ifc.sw_if_index == src_sw_if_index:
195 if hasattr(src_if, 'sub_if'):
196 # Check VLAN tags and Ethernet header
197 packet = src_if.sub_if.remove_dot1_layer(packet)
198 self.assertTrue(Dot1Q not in packet)
202 packet_index = payload_info.index
203 self.assertEqual(payload_info.dst, dst_sw_if_index)
204 self.logger.debug("Got packet on port %s: src=%u (id=%u)" %
205 (pg_if.name, payload_info.src, packet_index))
206 next_info = self.get_next_packet_info_for_interface2(
207 payload_info.src, dst_sw_if_index,
208 last_info[payload_info.src])
209 last_info[payload_info.src] = next_info
210 self.assertTrue(next_info is not None)
211 self.assertEqual(packet_index, next_info.index)
212 saved_packet = next_info.data
213 # Check standard fields
214 self.assertEqual(ip.src, saved_packet[IP].src)
215 self.assertEqual(ip.dst, saved_packet[IP].dst)
216 self.assertEqual(udp.sport, saved_packet[UDP].sport)
217 self.assertEqual(udp.dport, saved_packet[UDP].dport)
219 self.logger.error(ppp("Unexpected or invalid packet:", packet))
221 for i in self.pg_interfaces:
222 remaining_packet = self.get_next_packet_info_for_interface2(
223 i, dst_sw_if_index, last_info[i.sw_if_index])
225 remaining_packet is None,
226 "Port %u: Packet expected from source %u didn't arrive" %
227 (dst_sw_if_index, i.sw_if_index))
229 def run_l2bd_test(self, pkts_per_burst):
230 """ L2BD MAC learning test """
232 # Create incoming packet streams for packet-generator interfaces
233 for i in self.pg_interfaces:
234 packet_sizes = self.sub_if_packet_sizes if hasattr(i, 'sub_if') \
235 else self.pg_if_packet_sizes
236 pkts = self.create_stream(i, packet_sizes, pkts_per_burst)
239 # Enable packet capture and start packet sending
240 self.pg_enable_capture(self.pg_interfaces)
243 # Verify outgoing packet streams per packet-generator interface
244 for i in self.pg_interfaces:
245 capture = i.get_capture()
246 self.logger.info("Verifying capture on interface %s" % i.name)
247 self.verify_capture(i, capture)
249 def test_l2bd_sl(self):
250 """ L2BD MAC learning single-loop test
256 3 interfaces: untagged, dot1q, dot1ad (dot1q used instead of
257 dot1ad in the first version)
259 2.sending l2 eth pkts between 3 interface
260 64B, 512B, 1518B, 9200B (ether_size)
261 burst of 2 pkts per interface
264 self.run_l2bd_test(self.sl_pkts_per_burst)
266 def test_l2bd_dl(self):
267 """ L2BD MAC learning dual-loop test
273 3 interfaces: untagged, dot1q, dot1ad (dot1q used instead of
274 dot1ad in the first version)
276 2.sending l2 eth pkts between 3 interface
277 64B, 512B, 1518B, 9200B (ether_size)
278 burst of 257 pkts per interface
281 self.run_l2bd_test(self.dl_pkts_per_burst)
284 if __name__ == '__main__':
285 unittest.main(testRunner=VppTestRunner)