7 from scapy.packet import Raw
8 from scapy.layers.l2 import Ether
9 from scapy.layers.inet import IP, UDP
10 from scapy.layers.inet6 import IPv6
12 from framework import VppTestCase, VppTestRunner
13 from vpp_sub_interface import VppP2PSubint
14 from vpp_ip import DpoProto
15 from vpp_ip_route import VppIpRoute, VppRoutePath
16 from vpp_papi import mac_pton
19 class P2PEthernetAPI(VppTestCase):
20 """P2P Ethernet tests"""
26 super(P2PEthernetAPI, cls).setUpClass()
28 # Create pg interfaces
29 cls.create_pg_interfaces(range(4))
31 # Set up all interfaces
32 for i in cls.pg_interfaces:
35 def create_p2p_ethernet(self, parent_if, sub_id, remote_mac):
36 p2p = VppP2PSubint(self, parent_if, sub_id, mac_pton(remote_mac))
37 self.p2p_sub_ifs.append(p2p)
39 def delete_p2p_ethernet(self, parent_if, remote_mac):
40 self.vapi.delete_p2pethernet_subif(parent_if.sw_if_index,
44 """delete/create p2p subif"""
45 self.logger.info("FFP_TEST_START_0000")
47 self.create_p2p_ethernet(self.pg0, 1, "de:ad:00:00:00:01")
48 self.create_p2p_ethernet(self.pg0, 2, "de:ad:00:00:00:02")
49 intfs = self.vapi.cli("show interface")
51 self.assertNotEqual(intfs.find('pg0.1'), -1)
52 self.assertNotEqual(intfs.find('pg0.2'), -1)
53 self.assertEqual(intfs.find('pg0.5'), -1)
56 self.create_p2p_ethernet(self.pg0, 5, "de:ad:00:00:00:ff")
57 intfs = self.vapi.cli("show interface")
58 self.assertNotEqual(intfs.find('pg0.5'), -1)
60 self.delete_p2p_ethernet(self.pg0, "de:ad:00:00:00:ff")
62 intfs = self.vapi.cli("show interface")
64 self.assertNotEqual(intfs.find('pg0.1'), -1)
65 self.assertNotEqual(intfs.find('pg0.2'), -1)
66 self.assertEqual(intfs.find('pg0.5'), -1)
68 self.logger.info("FFP_TEST_FINISH_0000")
70 def test_p2p_subif_creation_1k(self):
71 """create 1k of p2p subifs"""
72 self.logger.info("FFP_TEST_START_0001")
76 mac = int("dead00000000", 16)
78 for i in range(1, clients+1):
80 macs.append(':'.join(re.findall('..', '{:02x}'.format(mac+i))))
81 self.vapi.create_p2pethernet_subif(self.pg2.sw_if_index,
85 self.logger.info("Failed to create subif %d %s" % (
89 intfs = self.vapi.cli("show interface").split("\n")
92 if intf.startswith('pg2.'):
94 self.assertEqual(count, clients)
96 self.logger.info("FFP_TEST_FINISH_0001")
99 class P2PEthernetIPV6(VppTestCase):
100 """P2P Ethernet IPv6 tests"""
107 super(P2PEthernetIPV6, cls).setUpClass()
109 # Create pg interfaces
110 cls.create_pg_interfaces(range(3))
113 cls.pg_if_packet_sizes = [64, 512, 1518, 9018]
115 # Set up all interfaces
116 for i in cls.pg_interfaces:
119 cls.pg0.generate_remote_hosts(3)
120 cls.pg0.configure_ipv6_neighbors()
123 cls.pg1.generate_remote_hosts(3)
124 cls.pg1.configure_ipv6_neighbors()
125 cls.pg1.disable_ipv6_ra()
128 super(P2PEthernetIPV6, self).setUp()
129 for p in self.packets:
130 self.packets.remove(p)
131 self.p2p_sub_ifs.append(
132 self.create_p2p_ethernet(self.pg0, 1,
133 self.pg0._remote_hosts[0].mac))
134 self.p2p_sub_ifs.append(
135 self.create_p2p_ethernet(self.pg0, 2,
136 self.pg0._remote_hosts[1].mac))
137 self.vapi.cli("trace add p2p-ethernet-input 50")
140 while len(self.p2p_sub_ifs):
141 p2p = self.p2p_sub_ifs.pop()
142 self.delete_p2p_ethernet(p2p)
144 super(P2PEthernetIPV6, self).tearDown()
146 def create_p2p_ethernet(self, parent_if, sub_id, remote_mac):
147 p2p = VppP2PSubint(self, parent_if, sub_id, mac_pton(remote_mac))
150 p2p.disable_ipv6_ra()
153 def delete_p2p_ethernet(self, p2p):
156 self.vapi.delete_p2pethernet_subif(p2p.parent.sw_if_index,
159 def create_stream(self, src_mac=None, dst_mac=None,
160 src_ip=None, dst_ip=None, size=None):
163 pkt_size = random.choice(self.pg_if_packet_sizes)
164 p = Ether(src=src_mac, dst=dst_mac)
165 p /= IPv6(src=src_ip, dst=dst_ip)
166 p /= (UDP(sport=1234, dport=4321) / Raw('\xa5' * 20))
167 self.extend_packet(p, pkt_size)
170 def send_packets(self, src_if=None, dst_if=None, packets=None, count=None):
171 self.pg_enable_capture([dst_if])
173 packets = self.packets
174 src_if.add_stream(packets)
178 return dst_if.get_capture(count)
180 def test_no_p2p_subif(self):
181 """standard routing without p2p subinterfaces"""
182 self.logger.info("FFP_TEST_START_0001")
184 route_8000 = VppIpRoute(self, "8000::", 64,
185 [VppRoutePath(self.pg0.remote_ip6,
186 self.pg0.sw_if_index,
187 proto=DpoProto.DPO_PROTO_IP6)],
189 route_8000.add_vpp_config()
191 self.packets = [(Ether(dst=self.pg1.local_mac,
192 src=self.pg1.remote_mac) /
193 IPv6(src="3001::1", dst="8000::100") /
194 UDP(sport=1234, dport=1234) /
196 self.send_packets(self.pg1, self.pg0)
198 self.logger.info("FFP_TEST_FINISH_0001")
200 def test_ip6_rx_p2p_subif(self):
201 """receive ipv6 packet via p2p subinterface"""
202 self.logger.info("FFP_TEST_START_0002")
204 route_9001 = VppIpRoute(self, "9001::", 64,
205 [VppRoutePath(self.pg1.remote_ip6,
206 self.pg1.sw_if_index,
207 proto=DpoProto.DPO_PROTO_IP6)],
209 route_9001.add_vpp_config()
212 self.create_stream(src_mac=self.pg0._remote_hosts[0].mac,
213 dst_mac=self.pg0.local_mac,
214 src_ip=self.p2p_sub_ifs[0].remote_ip6,
217 self.send_packets(self.pg0, self.pg1, self.packets)
218 self.assert_packet_counter_equal('p2p-ethernet-input', 1)
220 route_9001.remove_vpp_config()
221 self.logger.info("FFP_TEST_FINISH_0002")
223 def test_ip6_rx_p2p_subif_route(self):
224 """route rx ip6 packet not matching p2p subinterface"""
225 self.logger.info("FFP_TEST_START_0003")
227 self.pg0.config_ip6()
229 route_3 = VppIpRoute(self, "9000::", 64,
230 [VppRoutePath(self.pg1._remote_hosts[0].ip6,
231 self.pg1.sw_if_index,
232 proto=DpoProto.DPO_PROTO_IP6)],
234 route_3.add_vpp_config()
237 self.create_stream(src_mac="02:03:00:00:ff:ff",
238 dst_mac=self.pg0.local_mac,
242 self.send_packets(self.pg0, self.pg1)
244 self.pg0.unconfig_ip6()
246 route_3.remove_vpp_config()
248 self.logger.info("FFP_TEST_FINISH_0003")
250 def test_ip6_rx_p2p_subif_drop(self):
251 """drop rx packet not matching p2p subinterface"""
252 self.logger.info("FFP_TEST_START_0004")
254 route_9001 = VppIpRoute(self, "9000::", 64,
255 [VppRoutePath(self.pg1._remote_hosts[0].ip6,
256 self.pg1.sw_if_index,
257 proto=DpoProto.DPO_PROTO_IP6)],
259 route_9001.add_vpp_config()
262 self.create_stream(src_mac="02:03:00:00:ff:ff",
263 dst_mac=self.pg0.local_mac,
268 self.send_packets(self.pg0, self.pg1, count=0)
269 self.logger.info("FFP_TEST_FINISH_0004")
271 def test_ip6_tx_p2p_subif(self):
272 """send packet via p2p subinterface"""
273 self.logger.info("FFP_TEST_START_0005")
275 route_8000 = VppIpRoute(self, "8000::", 64,
276 [VppRoutePath(self.pg0.remote_ip6,
277 self.pg0.sw_if_index,
278 proto=DpoProto.DPO_PROTO_IP6)],
280 route_8000.add_vpp_config()
281 route_8001 = VppIpRoute(self, "8001::", 64,
282 [VppRoutePath(self.p2p_sub_ifs[0].remote_ip6,
283 self.p2p_sub_ifs[0].sw_if_index,
284 proto=DpoProto.DPO_PROTO_IP6)],
286 route_8001.add_vpp_config()
287 route_8002 = VppIpRoute(self, "8002::", 64,
288 [VppRoutePath(self.p2p_sub_ifs[1].remote_ip6,
289 self.p2p_sub_ifs[1].sw_if_index,
290 proto=DpoProto.DPO_PROTO_IP6)],
292 route_8002.add_vpp_config()
294 for i in range(0, 3):
296 self.create_stream(src_mac=self.pg1.remote_mac,
297 dst_mac=self.pg1.local_mac,
298 src_ip=self.pg1.remote_ip6,
299 dst_ip="800%d::100" % i))
301 self.send_packets(self.pg1, self.pg0, count=3)
303 route_8000.remove_vpp_config()
304 route_8001.remove_vpp_config()
305 route_8002.remove_vpp_config()
307 self.logger.info("FFP_TEST_FINISH_0005")
309 def test_ip6_tx_p2p_subif_drop(self):
310 """drop tx ip6 packet not matching p2p subinterface"""
311 self.logger.info("FFP_TEST_START_0006")
314 self.create_stream(src_mac="02:03:00:00:ff:ff",
315 dst_mac=self.pg0.local_mac,
320 self.send_packets(self.pg0, self.pg1, count=0)
321 self.logger.info("FFP_TEST_FINISH_0006")
324 class P2PEthernetIPV4(VppTestCase):
325 """P2P Ethernet IPv4 tests"""
332 super(P2PEthernetIPV4, cls).setUpClass()
334 # Create pg interfaces
335 cls.create_pg_interfaces(range(3))
338 cls.pg_if_packet_sizes = [64, 512, 1518, 9018]
340 # Set up all interfaces
341 for i in cls.pg_interfaces:
345 cls.pg0.generate_remote_hosts(5)
346 cls.pg0.configure_ipv4_neighbors()
349 cls.pg1.generate_remote_hosts(5)
350 cls.pg1.configure_ipv4_neighbors()
353 super(P2PEthernetIPV4, self).setUp()
354 for p in self.packets:
355 self.packets.remove(p)
356 self.p2p_sub_ifs.append(
357 self.create_p2p_ethernet(self.pg0, 1,
358 self.pg0._remote_hosts[0].mac))
359 self.p2p_sub_ifs.append(
360 self.create_p2p_ethernet(self.pg0, 2,
361 self.pg0._remote_hosts[1].mac))
362 self.vapi.cli("trace add p2p-ethernet-input 50")
365 while len(self.p2p_sub_ifs):
366 p2p = self.p2p_sub_ifs.pop()
367 self.delete_p2p_ethernet(p2p)
368 super(P2PEthernetIPV4, self).tearDown()
370 def create_stream(self, src_mac=None, dst_mac=None,
371 src_ip=None, dst_ip=None, size=None):
374 pkt_size = random.choice(self.pg_if_packet_sizes)
375 p = Ether(src=src_mac, dst=dst_mac)
376 p /= IP(src=src_ip, dst=dst_ip)
377 p /= (UDP(sport=1234, dport=4321) / Raw('\xa5' * 20))
378 self.extend_packet(p, pkt_size)
381 def send_packets(self, src_if=None, dst_if=None, packets=None, count=None):
382 self.pg_enable_capture([dst_if])
384 packets = self.packets
385 src_if.add_stream(packets)
389 return dst_if.get_capture(count)
391 def create_p2p_ethernet(self, parent_if, sub_id, remote_mac):
392 p2p = VppP2PSubint(self, parent_if, sub_id, mac_pton(remote_mac))
397 def delete_p2p_ethernet(self, p2p):
400 self.vapi.delete_p2pethernet_subif(p2p.parent.sw_if_index,
403 def test_ip4_rx_p2p_subif(self):
404 """receive ipv4 packet via p2p subinterface"""
405 self.logger.info("FFP_TEST_START_0002")
407 route_9000 = VppIpRoute(self, "9.0.0.0", 16,
408 [VppRoutePath(self.pg1.remote_ip4,
409 self.pg1.sw_if_index)])
410 route_9000.add_vpp_config()
413 self.create_stream(src_mac=self.pg0._remote_hosts[0].mac,
414 dst_mac=self.pg0.local_mac,
415 src_ip=self.p2p_sub_ifs[0].remote_ip4,
418 self.send_packets(self.pg0, self.pg1, self.packets)
420 self.assert_packet_counter_equal('p2p-ethernet-input', 1)
422 route_9000.remove_vpp_config()
423 self.logger.info("FFP_TEST_FINISH_0002")
425 def test_ip4_rx_p2p_subif_route(self):
426 """route rx packet not matching p2p subinterface"""
427 self.logger.info("FFP_TEST_START_0003")
429 route_9001 = VppIpRoute(self, "9.0.0.0", 24,
430 [VppRoutePath(self.pg1.remote_ip4,
431 self.pg1.sw_if_index)])
432 route_9001.add_vpp_config()
435 self.create_stream(src_mac="02:01:00:00:ff:ff",
436 dst_mac=self.pg0.local_mac,
440 self.send_packets(self.pg0, self.pg1)
442 route_9001.remove_vpp_config()
444 self.logger.info("FFP_TEST_FINISH_0003")
446 def test_ip4_tx_p2p_subif(self):
447 """send ip4 packet via p2p subinterface"""
448 self.logger.info("FFP_TEST_START_0005")
450 route_9100 = VppIpRoute(self, "9.1.0.100", 24,
451 [VppRoutePath(self.pg0.remote_ip4,
452 self.pg0.sw_if_index,
454 route_9100.add_vpp_config()
455 route_9200 = VppIpRoute(self, "9.2.0.100", 24,
456 [VppRoutePath(self.p2p_sub_ifs[0].remote_ip4,
457 self.p2p_sub_ifs[0].sw_if_index,
459 route_9200.add_vpp_config()
460 route_9300 = VppIpRoute(self, "9.3.0.100", 24,
461 [VppRoutePath(self.p2p_sub_ifs[1].remote_ip4,
462 self.p2p_sub_ifs[1].sw_if_index
464 route_9300.add_vpp_config()
466 for i in range(0, 3):
468 self.create_stream(src_mac=self.pg1.remote_mac,
469 dst_mac=self.pg1.local_mac,
470 src_ip=self.pg1.remote_ip4,
471 dst_ip="9.%d.0.100" % (i+1)))
473 self.send_packets(self.pg1, self.pg0)
475 # route_7000.remove_vpp_config()
476 route_9100.remove_vpp_config()
477 route_9200.remove_vpp_config()
478 route_9300.remove_vpp_config()
480 self.logger.info("FFP_TEST_FINISH_0005")
482 def test_ip4_tx_p2p_subif_drop(self):
483 """drop tx ip4 packet not matching p2p subinterface"""
484 self.logger.info("FFP_TEST_START_0006")
487 self.create_stream(src_mac="02:01:00:00:ff:ff",
488 dst_mac=self.pg0.local_mac,
493 self.send_packets(self.pg0, self.pg1, count=0)
494 self.logger.info("FFP_TEST_FINISH_0006")
497 if __name__ == '__main__':
498 unittest.main(testRunner=VppTestRunner)