5 - L2 MAC learning enabled in l2bd
6 - 2 routed interfaces untagged, bvi (Bridge Virtual Interface)
7 - 2 bridged interfaces in l2bd with bvi
10 - sending ip4 eth pkts between routed interfaces
12 - 2 bridged interfaces
14 - 64B, 512B, 1518B, 9200B (ether_size)
16 - burst of pkts per interface
18 - routed pkts hitting different FIB entries
19 - bridged pkts hitting different MAC entries
22 - all packets received correctly
27 from random import choice
29 from scapy.packet import Raw
30 from scapy.layers.l2 import Ether
31 from scapy.layers.inet import IP, UDP
33 from framework import VppTestCase, VppTestRunner
34 from vpp_mac import mactobinary
35 from vpp_papi_provider import L2_PORT_TYPE
38 class TestIpIrb(VppTestCase):
44 #. Create BD with MAC learning enabled and put interfaces to this BD.
45 #. Configure IPv4 addresses on loopback interface and routed interface.
46 #. Configure MAC address binding to IPv4 neighbors on loop0.
47 #. Configure MAC address on pg2.
48 #. Loopback BVI interface has remote hosts, one half of hosts are
49 behind pg0 second behind pg1.
51 super(TestIpIrb, cls).setUpClass()
53 cls.pg_if_packet_sizes = [64, 512, 1518, 9018] # packet sizes
55 cls.remote_hosts_count = 250
57 # create 3 pg interfaces, 1 loopback interface
58 cls.create_pg_interfaces(range(3))
59 cls.create_loopback_interfaces(1)
61 cls.interfaces = list(cls.pg_interfaces)
62 cls.interfaces.extend(cls.lo_interfaces)
64 for i in cls.interfaces:
67 # Create BD with MAC learning enabled and put interfaces to this BD
68 cls.vapi.sw_interface_set_l2_bridge(
69 cls.loop0.sw_if_index, bd_id=cls.bd_id,
70 port_type=L2_PORT_TYPE.BVI)
71 cls.vapi.sw_interface_set_l2_bridge(
72 cls.pg0.sw_if_index, bd_id=cls.bd_id)
73 cls.vapi.sw_interface_set_l2_bridge(
74 cls.pg1.sw_if_index, bd_id=cls.bd_id)
76 # Configure IPv4 addresses on loopback interface and routed interface
77 cls.loop0.config_ip4()
80 # Configure MAC address binding to IPv4 neighbors on loop0
81 cls.loop0.generate_remote_hosts(cls.remote_hosts_count)
82 cls.loop0.configure_ipv4_neighbors()
83 # configure MAC address on pg2
86 # Loopback BVI interface has remote hosts, one half of hosts are behind
87 # pg0 second behind pg1
88 half = cls.remote_hosts_count // 2
89 cls.pg0.remote_hosts = cls.loop0.remote_hosts[:half]
90 cls.pg1.remote_hosts = cls.loop0.remote_hosts[half:]
93 """Run standard test teardown and log ``show l2patch``,
94 ``show l2fib verbose``,``show bridge-domain <bd_id> detail``,
97 super(TestIpIrb, self).tearDown()
99 self.logger.info(self.vapi.cli("show l2patch"))
100 self.logger.info(self.vapi.cli("show l2fib verbose"))
101 self.logger.info(self.vapi.cli("show bridge-domain %s detail" %
103 self.logger.info(self.vapi.cli("show ip arp"))
105 def create_stream(self, src_ip_if, dst_ip_if, packet_sizes):
107 for i in range(0, 257):
108 remote_dst_host = choice(dst_ip_if.remote_hosts)
109 info = self.create_packet_info(src_ip_if, dst_ip_if)
110 payload = self.info_to_payload(info)
111 p = (Ether(dst=src_ip_if.local_mac, src=src_ip_if.remote_mac) /
112 IP(src=src_ip_if.remote_ip4,
113 dst=remote_dst_host.ip4) /
114 UDP(sport=1234, dport=1234) /
117 size = packet_sizes[(i // 2) % len(packet_sizes)]
118 self.extend_packet(p, size)
122 def create_stream_l2_to_ip(self, src_l2_if, src_ip_if, dst_ip_if,
125 for i in range(0, 257):
126 info = self.create_packet_info(src_ip_if, dst_ip_if)
127 payload = self.info_to_payload(info)
129 host = choice(src_l2_if.remote_hosts)
131 p = (Ether(src=host.mac,
132 dst=src_ip_if.local_mac) /
134 dst=dst_ip_if.remote_ip4) /
135 UDP(sport=1234, dport=1234) /
139 size = packet_sizes[(i // 2) % len(packet_sizes)]
140 self.extend_packet(p, size)
145 def verify_capture_l2_to_ip(self, dst_ip_if, src_ip_if, capture):
147 for i in self.interfaces:
148 last_info[i.sw_if_index] = None
150 dst_ip_sw_if_index = dst_ip_if.sw_if_index
152 for packet in capture:
154 udp = packet[IP][UDP]
155 payload_info = self.payload_to_info(str(packet[IP][UDP][Raw]))
157 self.assertEqual(payload_info.dst, dst_ip_sw_if_index)
159 next_info = self.get_next_packet_info_for_interface2(
160 payload_info.src, dst_ip_sw_if_index,
161 last_info[payload_info.src])
162 last_info[payload_info.src] = next_info
163 self.assertTrue(next_info is not None)
164 saved_packet = next_info.data
165 self.assertTrue(next_info is not None)
168 self.assertEqual(packet.src, dst_ip_if.local_mac)
169 self.assertEqual(packet.dst, dst_ip_if.remote_mac)
172 host = src_ip_if.host_by_ip4(ip.src)
173 self.assertIsNotNone(host)
174 self.assertEqual(ip.dst, saved_packet[IP].dst)
175 self.assertEqual(ip.dst, dst_ip_if.remote_ip4)
178 self.assertEqual(udp.sport, saved_packet[UDP].sport)
179 self.assertEqual(udp.dport, saved_packet[UDP].dport)
181 def verify_capture(self, dst_ip_if, src_ip_if, capture):
183 for i in self.interfaces:
184 last_info[i.sw_if_index] = None
186 dst_ip_sw_if_index = dst_ip_if.sw_if_index
188 for packet in capture:
190 udp = packet[IP][UDP]
191 payload_info = self.payload_to_info(str(packet[IP][UDP][Raw]))
192 packet_index = payload_info.index
194 self.assertEqual(payload_info.dst, dst_ip_sw_if_index)
196 next_info = self.get_next_packet_info_for_interface2(
197 payload_info.src, dst_ip_sw_if_index,
198 last_info[payload_info.src])
199 last_info[payload_info.src] = next_info
200 self.assertTrue(next_info is not None)
201 self.assertEqual(packet_index, next_info.index)
202 saved_packet = next_info.data
203 self.assertTrue(next_info is not None)
206 self.assertEqual(packet.src, dst_ip_if.local_mac)
207 host = dst_ip_if.host_by_mac(packet.dst)
210 self.assertEqual(ip.src, src_ip_if.remote_ip4)
211 self.assertEqual(ip.dst, saved_packet[IP].dst)
212 self.assertEqual(ip.dst, host.ip4)
215 self.assertEqual(udp.sport, saved_packet[UDP].sport)
216 self.assertEqual(udp.dport, saved_packet[UDP].dport)
218 def test_ip4_irb_1(self):
222 - ip traffic from pg2 interface must ends in both pg0 and pg1
223 - arp entry present in loop0 interface for destination IP
224 - no l2 entree configured, pg0 and pg1 are same
227 stream = self.create_stream(
228 self.pg2, self.loop0, self.pg_if_packet_sizes)
229 self.pg2.add_stream(stream)
231 self.pg_enable_capture(self.pg_interfaces)
234 packet_count = self.get_packet_count_for_if_idx(self.loop0.sw_if_index)
236 rcvd1 = self.pg0.get_capture(packet_count)
237 rcvd2 = self.pg1.get_capture(packet_count)
239 self.verify_capture(self.loop0, self.pg2, rcvd1)
240 self.verify_capture(self.loop0, self.pg2, rcvd2)
242 self.assertListEqual(rcvd1.res, rcvd2.res)
244 def send_and_verify_l2_to_ip(self):
245 stream1 = self.create_stream_l2_to_ip(
246 self.pg0, self.loop0, self.pg2, self.pg_if_packet_sizes)
247 stream2 = self.create_stream_l2_to_ip(
248 self.pg1, self.loop0, self.pg2, self.pg_if_packet_sizes)
249 self.vapi.cli("clear trace")
250 self.pg0.add_stream(stream1)
251 self.pg1.add_stream(stream2)
253 self.pg_enable_capture(self.pg_interfaces)
256 rcvd = self.pg2.get_capture(514)
257 self.verify_capture_l2_to_ip(self.pg2, self.loop0, rcvd)
259 def test_ip4_irb_2(self):
263 - ip traffic from pg0 and pg1 ends on pg2
265 self.send_and_verify_l2_to_ip()
267 # change the BVI's mac and resed traffic
268 self.loop0.set_mac("00:00:00:11:11:33")
270 self.send_and_verify_l2_to_ip()
271 # check it wasn't flooded
272 self.pg1.assert_nothing_captured(remark="UU Flood")
275 if __name__ == '__main__':
276 unittest.main(testRunner=VppTestRunner)