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 util import mactobinary
37 class TestIpIrb(VppTestCase):
43 #. Create BD with MAC learning enabled and put interfaces to this BD.
44 #. Configure IPv4 addresses on loopback interface and routed interface.
45 #. Configure MAC address binding to IPv4 neighbors on loop0.
46 #. Configure MAC address on pg2.
47 #. Loopback BVI interface has remote hosts, one half of hosts are
48 behind pg0 second behind pg1.
50 super(TestIpIrb, cls).setUpClass()
52 cls.pg_if_packet_sizes = [64, 512, 1518, 9018] # packet sizes
54 cls.remote_hosts_count = 250
56 # create 3 pg interfaces, 1 loopback interface
57 cls.create_pg_interfaces(range(3))
58 cls.create_loopback_interfaces(1)
60 cls.interfaces = list(cls.pg_interfaces)
61 cls.interfaces.extend(cls.lo_interfaces)
63 for i in cls.interfaces:
66 # Create BD with MAC learning enabled and put interfaces to this BD
67 cls.vapi.sw_interface_set_l2_bridge(
68 cls.loop0.sw_if_index, bd_id=cls.bd_id, bvi=1)
69 cls.vapi.sw_interface_set_l2_bridge(
70 cls.pg0.sw_if_index, bd_id=cls.bd_id)
71 cls.vapi.sw_interface_set_l2_bridge(
72 cls.pg1.sw_if_index, bd_id=cls.bd_id)
74 # Configure IPv4 addresses on loopback interface and routed interface
75 cls.loop0.config_ip4()
78 # Configure MAC address binding to IPv4 neighbors on loop0
79 cls.loop0.generate_remote_hosts(cls.remote_hosts_count)
80 cls.loop0.configure_ipv4_neighbors()
81 # configure MAC address on pg2
84 # Loopback BVI interface has remote hosts, one half of hosts are behind
85 # pg0 second behind pg1
86 half = cls.remote_hosts_count // 2
87 cls.pg0.remote_hosts = cls.loop0.remote_hosts[:half]
88 cls.pg1.remote_hosts = cls.loop0.remote_hosts[half:]
91 """Run standard test teardown and log ``show l2patch``,
92 ``show l2fib verbose``,``show bridge-domain <bd_id> detail``,
95 super(TestIpIrb, self).tearDown()
97 self.logger.info(self.vapi.cli("show l2patch"))
98 self.logger.info(self.vapi.cli("show l2fib verbose"))
99 self.logger.info(self.vapi.cli("show bridge-domain %s detail" %
101 self.logger.info(self.vapi.cli("show ip arp"))
103 def create_stream(self, src_ip_if, dst_ip_if, packet_sizes):
105 for i in range(0, 257):
106 remote_dst_host = choice(dst_ip_if.remote_hosts)
107 info = self.create_packet_info(src_ip_if, dst_ip_if)
108 payload = self.info_to_payload(info)
109 p = (Ether(dst=src_ip_if.local_mac, src=src_ip_if.remote_mac) /
110 IP(src=src_ip_if.remote_ip4,
111 dst=remote_dst_host.ip4) /
112 UDP(sport=1234, dport=1234) /
115 size = packet_sizes[(i // 2) % len(packet_sizes)]
116 self.extend_packet(p, size)
120 def create_stream_l2_to_ip(self, src_l2_if, src_ip_if, dst_ip_if,
123 for i in range(0, 257):
124 info = self.create_packet_info(src_ip_if, dst_ip_if)
125 payload = self.info_to_payload(info)
127 host = choice(src_l2_if.remote_hosts)
129 p = (Ether(src=host.mac,
130 dst=src_ip_if.local_mac) /
132 dst=dst_ip_if.remote_ip4) /
133 UDP(sport=1234, dport=1234) /
137 size = packet_sizes[(i // 2) % len(packet_sizes)]
138 self.extend_packet(p, size)
143 def verify_capture_l2_to_ip(self, dst_ip_if, src_ip_if, capture):
145 for i in self.interfaces:
146 last_info[i.sw_if_index] = None
148 dst_ip_sw_if_index = dst_ip_if.sw_if_index
150 for packet in capture:
152 udp = packet[IP][UDP]
153 payload_info = self.payload_to_info(str(packet[IP][UDP][Raw]))
155 self.assertEqual(payload_info.dst, dst_ip_sw_if_index)
157 next_info = self.get_next_packet_info_for_interface2(
158 payload_info.src, dst_ip_sw_if_index,
159 last_info[payload_info.src])
160 last_info[payload_info.src] = next_info
161 self.assertTrue(next_info is not None)
162 saved_packet = next_info.data
163 self.assertTrue(next_info is not None)
166 self.assertEqual(packet.src, dst_ip_if.local_mac)
167 self.assertEqual(packet.dst, dst_ip_if.remote_mac)
170 host = src_ip_if.host_by_ip4(ip.src)
171 self.assertIsNotNone(host)
172 self.assertEqual(ip.dst, saved_packet[IP].dst)
173 self.assertEqual(ip.dst, dst_ip_if.remote_ip4)
176 self.assertEqual(udp.sport, saved_packet[UDP].sport)
177 self.assertEqual(udp.dport, saved_packet[UDP].dport)
179 def verify_capture(self, dst_ip_if, src_ip_if, capture):
181 for i in self.interfaces:
182 last_info[i.sw_if_index] = None
184 dst_ip_sw_if_index = dst_ip_if.sw_if_index
186 for packet in capture:
188 udp = packet[IP][UDP]
189 payload_info = self.payload_to_info(str(packet[IP][UDP][Raw]))
190 packet_index = payload_info.index
192 self.assertEqual(payload_info.dst, dst_ip_sw_if_index)
194 next_info = self.get_next_packet_info_for_interface2(
195 payload_info.src, dst_ip_sw_if_index,
196 last_info[payload_info.src])
197 last_info[payload_info.src] = next_info
198 self.assertTrue(next_info is not None)
199 self.assertEqual(packet_index, next_info.index)
200 saved_packet = next_info.data
201 self.assertTrue(next_info is not None)
204 self.assertEqual(packet.src, dst_ip_if.local_mac)
205 host = dst_ip_if.host_by_mac(packet.dst)
208 self.assertEqual(ip.src, src_ip_if.remote_ip4)
209 self.assertEqual(ip.dst, saved_packet[IP].dst)
210 self.assertEqual(ip.dst, host.ip4)
213 self.assertEqual(udp.sport, saved_packet[UDP].sport)
214 self.assertEqual(udp.dport, saved_packet[UDP].dport)
216 def test_ip4_irb_1(self):
220 - ip traffic from pg2 interface must ends in both pg0 and pg1
221 - arp entry present in loop0 interface for destination IP
222 - no l2 entree configured, pg0 and pg1 are same
225 stream = self.create_stream(
226 self.pg2, self.loop0, self.pg_if_packet_sizes)
227 self.pg2.add_stream(stream)
229 self.pg_enable_capture(self.pg_interfaces)
232 packet_count = self.get_packet_count_for_if_idx(self.loop0.sw_if_index)
234 rcvd1 = self.pg0.get_capture(packet_count)
235 rcvd2 = self.pg1.get_capture(packet_count)
237 self.verify_capture(self.loop0, self.pg2, rcvd1)
238 self.verify_capture(self.loop0, self.pg2, rcvd2)
240 self.assertListEqual(rcvd1.res, rcvd2.res)
242 def send_and_verify_l2_to_ip(self):
243 stream1 = self.create_stream_l2_to_ip(
244 self.pg0, self.loop0, self.pg2, self.pg_if_packet_sizes)
245 stream2 = self.create_stream_l2_to_ip(
246 self.pg1, self.loop0, self.pg2, self.pg_if_packet_sizes)
247 self.vapi.cli("clear trace")
248 self.pg0.add_stream(stream1)
249 self.pg1.add_stream(stream2)
251 self.pg_enable_capture(self.pg_interfaces)
254 rcvd = self.pg2.get_capture(514)
255 self.verify_capture_l2_to_ip(self.pg2, self.loop0, rcvd)
257 def test_ip4_irb_2(self):
261 - ip traffic from pg0 and pg1 ends on pg2
263 self.send_and_verify_l2_to_ip()
265 # change the BVI's mac and resed traffic
266 self.loop0.set_mac("00:00:00:11:11:33")
268 self.send_and_verify_l2_to_ip()
269 # check it wasn't flooded
270 self.pg1.assert_nothing_captured(remark="UU Flood")
273 if __name__ == '__main__':
274 unittest.main(testRunner=VppTestRunner)