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
36 class TestIpIrb(VppTestCase):
42 #. Create BD with MAC learning enabled and put interfaces to this BD.
43 #. Configure IPv4 addresses on loopback interface and routed interface.
44 #. Configure MAC address binding to IPv4 neighbors on loop0.
45 #. Configure MAC address on pg2.
46 #. Loopback BVI interface has remote hosts, one half of hosts are
47 behind pg0 second behind pg1.
49 super(TestIpIrb, cls).setUpClass()
51 cls.pg_if_packet_sizes = [64, 512, 1518, 9018] # packet sizes
53 cls.remote_hosts_count = 250
55 # create 3 pg interfaces, 1 loopback interface
56 cls.create_pg_interfaces(range(3))
57 cls.create_loopback_interfaces(1)
59 cls.interfaces = list(cls.pg_interfaces)
60 cls.interfaces.extend(cls.lo_interfaces)
62 for i in cls.interfaces:
65 # Create BD with MAC learning enabled and put interfaces to this BD
66 cls.vapi.sw_interface_set_l2_bridge(
67 cls.loop0.sw_if_index, bd_id=cls.bd_id, bvi=1)
68 cls.vapi.sw_interface_set_l2_bridge(
69 cls.pg0.sw_if_index, bd_id=cls.bd_id)
70 cls.vapi.sw_interface_set_l2_bridge(
71 cls.pg1.sw_if_index, bd_id=cls.bd_id)
73 # Configure IPv4 addresses on loopback interface and routed interface
74 cls.loop0.config_ip4()
77 # Configure MAC address binding to IPv4 neighbors on loop0
78 cls.loop0.generate_remote_hosts(cls.remote_hosts_count)
79 cls.loop0.configure_ipv4_neighbors()
80 # configure MAC address on pg2
83 # Loopback BVI interface has remote hosts, one half of hosts are behind
84 # pg0 second behind pg1
85 half = cls.remote_hosts_count // 2
86 cls.pg0.remote_hosts = cls.loop0.remote_hosts[:half]
87 cls.pg1.remote_hosts = cls.loop0.remote_hosts[half:]
90 """Run standard test teardown and log ``show l2patch``,
91 ``show l2fib verbose``,``show bridge-domain <bd_id> detail``,
94 super(TestIpIrb, self).tearDown()
96 self.logger.info(self.vapi.cli("show l2patch"))
97 self.logger.info(self.vapi.cli("show l2fib verbose"))
98 self.logger.info(self.vapi.cli("show bridge-domain %s detail" %
100 self.logger.info(self.vapi.cli("show ip arp"))
102 def create_stream(self, src_ip_if, dst_ip_if, packet_sizes):
104 for i in range(0, 257):
105 remote_dst_host = choice(dst_ip_if.remote_hosts)
106 info = self.create_packet_info(src_ip_if, dst_ip_if)
107 payload = self.info_to_payload(info)
108 p = (Ether(dst=src_ip_if.local_mac, src=src_ip_if.remote_mac) /
109 IP(src=src_ip_if.remote_ip4,
110 dst=remote_dst_host.ip4) /
111 UDP(sport=1234, dport=1234) /
114 size = packet_sizes[(i // 2) % len(packet_sizes)]
115 self.extend_packet(p, size)
119 def create_stream_l2_to_ip(self, src_l2_if, src_ip_if, dst_ip_if,
122 for i in range(0, 257):
123 info = self.create_packet_info(src_ip_if, dst_ip_if)
124 payload = self.info_to_payload(info)
126 host = choice(src_l2_if.remote_hosts)
128 p = (Ether(src=host.mac,
129 dst=src_ip_if.local_mac) /
131 dst=dst_ip_if.remote_ip4) /
132 UDP(sport=1234, dport=1234) /
136 size = packet_sizes[(i // 2) % len(packet_sizes)]
137 self.extend_packet(p, size)
142 def verify_capture_l2_to_ip(self, dst_ip_if, src_ip_if, capture):
144 for i in self.interfaces:
145 last_info[i.sw_if_index] = None
147 dst_ip_sw_if_index = dst_ip_if.sw_if_index
149 for packet in capture:
151 udp = packet[IP][UDP]
152 payload_info = self.payload_to_info(str(packet[IP][UDP][Raw]))
154 self.assertEqual(payload_info.dst, dst_ip_sw_if_index)
156 next_info = self.get_next_packet_info_for_interface2(
157 payload_info.src, dst_ip_sw_if_index,
158 last_info[payload_info.src])
159 last_info[payload_info.src] = next_info
160 self.assertTrue(next_info is not None)
161 saved_packet = next_info.data
162 self.assertTrue(next_info is not None)
165 self.assertEqual(packet.src, dst_ip_if.local_mac)
166 self.assertEqual(packet.dst, dst_ip_if.remote_mac)
169 host = src_ip_if.host_by_ip4(ip.src)
170 self.assertIsNotNone(host)
171 self.assertEqual(ip.dst, saved_packet[IP].dst)
172 self.assertEqual(ip.dst, dst_ip_if.remote_ip4)
175 self.assertEqual(udp.sport, saved_packet[UDP].sport)
176 self.assertEqual(udp.dport, saved_packet[UDP].dport)
178 def verify_capture(self, dst_ip_if, src_ip_if, capture):
180 for i in self.interfaces:
181 last_info[i.sw_if_index] = None
183 dst_ip_sw_if_index = dst_ip_if.sw_if_index
185 for packet in capture:
187 udp = packet[IP][UDP]
188 payload_info = self.payload_to_info(str(packet[IP][UDP][Raw]))
189 packet_index = payload_info.index
191 self.assertEqual(payload_info.dst, dst_ip_sw_if_index)
193 next_info = self.get_next_packet_info_for_interface2(
194 payload_info.src, dst_ip_sw_if_index,
195 last_info[payload_info.src])
196 last_info[payload_info.src] = next_info
197 self.assertTrue(next_info is not None)
198 self.assertEqual(packet_index, next_info.index)
199 saved_packet = next_info.data
200 self.assertTrue(next_info is not None)
203 self.assertEqual(packet.src, dst_ip_if.local_mac)
204 host = dst_ip_if.host_by_mac(packet.dst)
207 self.assertEqual(ip.src, src_ip_if.remote_ip4)
208 self.assertEqual(ip.dst, saved_packet[IP].dst)
209 self.assertEqual(ip.dst, host.ip4)
212 self.assertEqual(udp.sport, saved_packet[UDP].sport)
213 self.assertEqual(udp.dport, saved_packet[UDP].dport)
215 def test_ip4_irb_1(self):
219 - ip traffic from pg2 interface must ends in both pg0 and pg1
220 - arp entry present in loop0 interface for destination IP
221 - no l2 entree configured, pg0 and pg1 are same
224 stream = self.create_stream(
225 self.pg2, self.loop0, self.pg_if_packet_sizes)
226 self.pg2.add_stream(stream)
228 self.pg_enable_capture(self.pg_interfaces)
231 packet_count = self.get_packet_count_for_if_idx(self.loop0.sw_if_index)
233 rcvd1 = self.pg0.get_capture(packet_count)
234 rcvd2 = self.pg1.get_capture(packet_count)
236 self.verify_capture(self.loop0, self.pg2, rcvd1)
237 self.verify_capture(self.loop0, self.pg2, rcvd2)
239 self.assertListEqual(rcvd1.res, rcvd2.res)
241 def test_ip4_irb_2(self):
245 - ip traffic from pg0 and pg1 ends on pg2
248 stream1 = self.create_stream_l2_to_ip(
249 self.pg0, self.loop0, self.pg2, self.pg_if_packet_sizes)
250 stream2 = self.create_stream_l2_to_ip(
251 self.pg1, self.loop0, self.pg2, self.pg_if_packet_sizes)
252 self.pg0.add_stream(stream1)
253 self.pg1.add_stream(stream2)
255 self.pg_enable_capture(self.pg_interfaces)
258 rcvd = self.pg2.get_capture()
259 self.verify_capture_l2_to_ip(self.pg2, self.loop0, rcvd)
262 if __name__ == '__main__':
263 unittest.main(testRunner=VppTestRunner)