6 from scapy.packet import Raw
7 from scapy.layers.l2 import Ether
8 from scapy.layers.inet import IP, UDP
9 from scapy.layers.inet6 import IPv6
11 from framework import VppTestCase
12 from asfframework import VppTestRunner
13 from vpp_sub_interface import VppP2PSubint
14 from vpp_ip_route import VppIpRoute, VppRoutePath
15 from vpp_papi import mac_pton
18 class P2PEthernetAPI(VppTestCase):
19 """P2P Ethernet tests"""
25 super(P2PEthernetAPI, cls).setUpClass()
27 # Create pg interfaces
28 cls.create_pg_interfaces(range(4))
30 # Set up all interfaces
31 for i in cls.pg_interfaces:
35 def tearDownClass(cls):
36 super(P2PEthernetAPI, cls).tearDownClass()
38 def create_p2p_ethernet(self, parent_if, sub_id, remote_mac):
39 p2p = VppP2PSubint(self, parent_if, sub_id, mac_pton(remote_mac))
40 self.p2p_sub_ifs.append(p2p)
42 def delete_p2p_ethernet(self, parent_if, remote_mac):
43 self.vapi.p2p_ethernet_del(parent_if.sw_if_index, mac_pton(remote_mac))
46 """delete/create p2p subif"""
47 self.logger.info("FFP_TEST_START_0000")
49 self.create_p2p_ethernet(self.pg0, 1, "de:ad:00:00:00:01")
50 self.create_p2p_ethernet(self.pg0, 2, "de:ad:00:00:00:02")
51 intfs = self.vapi.cli("show interface")
53 self.assertIn("pg0.1", intfs)
54 self.assertIn("pg0.2", intfs)
55 self.assertNotIn("pg0.5", intfs)
58 self.create_p2p_ethernet(self.pg0, 5, "de:ad:00:00:00:ff")
59 intfs = self.vapi.cli("show interface")
60 self.assertIn("pg0.5", intfs)
62 self.delete_p2p_ethernet(self.pg0, "de:ad:00:00:00:ff")
64 intfs = self.vapi.cli("show interface")
66 self.assertIn("pg0.1", intfs)
67 self.assertIn("pg0.2", intfs)
68 self.assertNotIn("pg0.5", intfs)
70 self.logger.info("FFP_TEST_FINISH_0000")
72 def test_p2p_subif_creation_1k(self):
73 """create 1k of p2p subifs"""
74 self.logger.info("FFP_TEST_START_0001")
78 mac = int("dead00000000", 16)
80 for i in range(1, clients + 1):
82 macs.append(":".join(re.findall("..", "{:02x}".format(mac + i))))
83 self.vapi.p2p_ethernet_add(
84 self.pg2.sw_if_index, mac_pton(macs[i - 1]), i
87 self.logger.info("Failed to create subif %d %s" % (i, macs[i - 1]))
90 intfs = self.vapi.cli("show interface").split("\n")
93 if intf.startswith("pg2."):
95 self.assertEqual(count, clients)
97 self.logger.info("FFP_TEST_FINISH_0001")
100 class P2PEthernetIPV6(VppTestCase):
101 """P2P Ethernet IPv6 tests"""
108 super(P2PEthernetIPV6, cls).setUpClass()
110 # Create pg interfaces
111 cls.create_pg_interfaces(range(3))
114 cls.pg_if_packet_sizes = [64, 512, 1518, 9018]
116 # Set up all interfaces
117 for i in cls.pg_interfaces:
120 cls.pg0.generate_remote_hosts(3)
121 cls.pg0.configure_ipv6_neighbors()
124 cls.pg1.generate_remote_hosts(3)
125 cls.pg1.configure_ipv6_neighbors()
126 cls.pg1.disable_ipv6_ra()
129 def tearDownClass(cls):
130 super(P2PEthernetIPV6, cls).tearDownClass()
133 super(P2PEthernetIPV6, self).setUp()
134 for p in self.packets:
135 self.packets.remove(p)
136 self.p2p_sub_ifs.append(
137 self.create_p2p_ethernet(self.pg0, 1, self.pg0._remote_hosts[0].mac)
139 self.p2p_sub_ifs.append(
140 self.create_p2p_ethernet(self.pg0, 2, self.pg0._remote_hosts[1].mac)
142 self.vapi.cli("trace add p2p-ethernet-input 50")
145 while len(self.p2p_sub_ifs):
146 p2p = self.p2p_sub_ifs.pop()
147 self.delete_p2p_ethernet(p2p)
149 super(P2PEthernetIPV6, self).tearDown()
151 def create_p2p_ethernet(self, parent_if, sub_id, remote_mac):
152 p2p = VppP2PSubint(self, parent_if, sub_id, mac_pton(remote_mac))
155 p2p.disable_ipv6_ra()
158 def delete_p2p_ethernet(self, p2p):
161 self.vapi.p2p_ethernet_del(p2p.parent.sw_if_index, p2p.p2p_remote_mac)
164 self, src_mac=None, dst_mac=None, src_ip=None, dst_ip=None, size=None
168 pkt_size = random.choice(self.pg_if_packet_sizes)
169 p = Ether(src=src_mac, dst=dst_mac)
170 p /= IPv6(src=src_ip, dst=dst_ip)
171 p /= UDP(sport=1234, dport=4321) / Raw(b"\xa5" * 20)
172 self.extend_packet(p, pkt_size)
175 def send_packets(self, src_if=None, dst_if=None, packets=None, count=None):
176 self.pg_enable_capture([dst_if])
178 packets = self.packets
179 src_if.add_stream(packets)
183 return dst_if.get_capture(count)
185 def test_no_p2p_subif(self):
186 """standard routing without p2p subinterfaces"""
187 self.logger.info("FFP_TEST_START_0001")
189 self.pg0.config_ip6()
190 route_8000 = VppIpRoute(
194 [VppRoutePath(self.pg0.remote_ip6, self.pg0.sw_if_index)],
196 route_8000.add_vpp_config()
200 Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac)
201 / IPv6(src="3001::1", dst="8000::100")
202 / UDP(sport=1234, dport=1234)
206 self.send_packets(self.pg1, self.pg0)
208 self.pg0.unconfig_ip6()
209 self.logger.info("FFP_TEST_FINISH_0001")
211 def test_ip6_rx_p2p_subif(self):
212 """receive ipv6 packet via p2p subinterface"""
213 self.logger.info("FFP_TEST_START_0002")
215 route_9001 = VppIpRoute(
219 [VppRoutePath(self.pg1.remote_ip6, self.pg1.sw_if_index)],
221 route_9001.add_vpp_config()
225 src_mac=self.pg0._remote_hosts[0].mac,
226 dst_mac=self.pg0.local_mac,
227 src_ip=self.p2p_sub_ifs[0].remote_ip6,
232 self.send_packets(self.pg0, self.pg1, self.packets)
233 self.assert_packet_counter_equal("p2p-ethernet-input", 1)
235 route_9001.remove_vpp_config()
236 self.logger.info("FFP_TEST_FINISH_0002")
238 def test_ip6_rx_p2p_subif_route(self):
239 """route rx ip6 packet not matching p2p subinterface"""
240 self.logger.info("FFP_TEST_START_0003")
242 self.pg0.config_ip6()
244 route_3 = VppIpRoute(
248 [VppRoutePath(self.pg1._remote_hosts[0].ip6, self.pg1.sw_if_index)],
250 route_3.add_vpp_config()
254 src_mac="02:03:00:00:ff:ff",
255 dst_mac=self.pg0.local_mac,
261 self.send_packets(self.pg0, self.pg1)
263 self.pg0.unconfig_ip6()
265 route_3.remove_vpp_config()
267 self.logger.info("FFP_TEST_FINISH_0003")
269 def test_ip6_rx_p2p_subif_drop(self):
270 """drop rx packet not matching p2p subinterface"""
271 self.logger.info("FFP_TEST_START_0004")
273 route_9001 = VppIpRoute(
277 [VppRoutePath(self.pg1._remote_hosts[0].ip6, self.pg1.sw_if_index)],
279 route_9001.add_vpp_config()
283 src_mac="02:03:00:00:ff:ff",
284 dst_mac=self.pg0.local_mac,
291 self.send_packets(self.pg0, self.pg1, count=0)
292 self.logger.info("FFP_TEST_FINISH_0004")
294 def test_ip6_tx_p2p_subif(self):
295 """send packet via p2p subinterface"""
296 self.logger.info("FFP_TEST_START_0005")
298 self.pg0.config_ip6()
300 route_8000 = VppIpRoute(
304 [VppRoutePath(self.pg0.remote_hosts[0].ip6, self.pg0.sw_if_index)],
306 route_8000.add_vpp_config()
307 route_8001 = VppIpRoute(
313 self.p2p_sub_ifs[0].remote_ip6, self.p2p_sub_ifs[0].sw_if_index
317 route_8001.add_vpp_config()
318 route_8002 = VppIpRoute(
324 self.p2p_sub_ifs[1].remote_ip6, self.p2p_sub_ifs[1].sw_if_index
328 route_8002.add_vpp_config()
330 for i in range(0, 3):
333 src_mac=self.pg1.remote_mac,
334 dst_mac=self.pg1.local_mac,
335 src_ip=self.pg1.remote_ip6,
336 dst_ip="800%d::100" % i,
340 self.send_packets(self.pg1, self.pg0, count=3)
342 route_8000.remove_vpp_config()
343 route_8001.remove_vpp_config()
344 route_8002.remove_vpp_config()
346 self.pg0.unconfig_ip6()
347 self.logger.info("FFP_TEST_FINISH_0005")
349 def test_ip6_tx_p2p_subif_drop(self):
350 """drop tx ip6 packet not matching p2p subinterface"""
351 self.logger.info("FFP_TEST_START_0006")
355 src_mac="02:03:00:00:ff:ff",
356 dst_mac=self.pg0.local_mac,
363 self.send_packets(self.pg0, self.pg1, count=0)
364 self.logger.info("FFP_TEST_FINISH_0006")
367 class P2PEthernetIPV4(VppTestCase):
368 """P2P Ethernet IPv4 tests"""
375 super(P2PEthernetIPV4, cls).setUpClass()
377 # Create pg interfaces
378 cls.create_pg_interfaces(range(3))
381 cls.pg_if_packet_sizes = [64, 512, 1518, 9018]
383 # Set up all interfaces
384 for i in cls.pg_interfaces:
388 cls.pg0.generate_remote_hosts(5)
389 cls.pg0.configure_ipv4_neighbors()
392 cls.pg1.generate_remote_hosts(5)
393 cls.pg1.configure_ipv4_neighbors()
396 def tearDownClass(cls):
397 super(P2PEthernetIPV4, cls).tearDownClass()
400 super(P2PEthernetIPV4, self).setUp()
401 for p in self.packets:
402 self.packets.remove(p)
403 self.p2p_sub_ifs.append(
404 self.create_p2p_ethernet(self.pg0, 1, self.pg0._remote_hosts[0].mac)
406 self.p2p_sub_ifs.append(
407 self.create_p2p_ethernet(self.pg0, 2, self.pg0._remote_hosts[1].mac)
409 self.vapi.cli("trace add p2p-ethernet-input 50")
412 while len(self.p2p_sub_ifs):
413 p2p = self.p2p_sub_ifs.pop()
414 self.delete_p2p_ethernet(p2p)
415 super(P2PEthernetIPV4, self).tearDown()
418 self, src_mac=None, dst_mac=None, src_ip=None, dst_ip=None, size=None
422 pkt_size = random.choice(self.pg_if_packet_sizes)
423 p = Ether(src=src_mac, dst=dst_mac)
424 p /= IP(src=src_ip, dst=dst_ip)
425 p /= UDP(sport=1234, dport=4321) / Raw(b"\xa5" * 20)
426 self.extend_packet(p, pkt_size)
429 def send_packets(self, src_if=None, dst_if=None, packets=None, count=None):
430 self.pg_enable_capture([dst_if])
432 packets = self.packets
433 src_if.add_stream(packets)
437 return dst_if.get_capture(count)
439 def create_p2p_ethernet(self, parent_if, sub_id, remote_mac):
440 p2p = VppP2PSubint(self, parent_if, sub_id, mac_pton(remote_mac))
445 def delete_p2p_ethernet(self, p2p):
448 self.vapi.p2p_ethernet_del(p2p.parent.sw_if_index, p2p.p2p_remote_mac)
450 def test_ip4_rx_p2p_subif(self):
451 """receive ipv4 packet via p2p subinterface"""
452 self.logger.info("FFP_TEST_START_0002")
454 route_9000 = VppIpRoute(
458 [VppRoutePath(self.pg1.remote_ip4, self.pg1.sw_if_index)],
460 route_9000.add_vpp_config()
464 src_mac=self.pg0._remote_hosts[0].mac,
465 dst_mac=self.pg0.local_mac,
466 src_ip=self.p2p_sub_ifs[0].remote_ip4,
471 self.send_packets(self.pg0, self.pg1, self.packets)
473 self.assert_packet_counter_equal("p2p-ethernet-input", 1)
475 route_9000.remove_vpp_config()
476 self.logger.info("FFP_TEST_FINISH_0002")
478 def test_ip4_rx_p2p_subif_route(self):
479 """route rx packet not matching p2p subinterface"""
480 self.logger.info("FFP_TEST_START_0003")
482 route_9001 = VppIpRoute(
486 [VppRoutePath(self.pg1.remote_ip4, self.pg1.sw_if_index)],
488 route_9001.add_vpp_config()
492 src_mac="02:01:00:00:ff:ff",
493 dst_mac=self.pg0.local_mac,
499 self.send_packets(self.pg0, self.pg1)
501 route_9001.remove_vpp_config()
503 self.logger.info("FFP_TEST_FINISH_0003")
505 def test_ip4_tx_p2p_subif(self):
506 """send ip4 packet via p2p subinterface"""
507 self.logger.info("FFP_TEST_START_0005")
509 route_9100 = VppIpRoute(
516 self.pg0.sw_if_index,
520 route_9100.add_vpp_config()
521 route_9200 = VppIpRoute(
527 self.p2p_sub_ifs[0].remote_ip4,
528 self.p2p_sub_ifs[0].sw_if_index,
532 route_9200.add_vpp_config()
533 route_9300 = VppIpRoute(
539 self.p2p_sub_ifs[1].remote_ip4, self.p2p_sub_ifs[1].sw_if_index
543 route_9300.add_vpp_config()
545 for i in range(0, 3):
548 src_mac=self.pg1.remote_mac,
549 dst_mac=self.pg1.local_mac,
550 src_ip=self.pg1.remote_ip4,
551 dst_ip="9.%d.0.100" % (i + 1),
555 self.send_packets(self.pg1, self.pg0)
557 # route_7000.remove_vpp_config()
558 route_9100.remove_vpp_config()
559 route_9200.remove_vpp_config()
560 route_9300.remove_vpp_config()
562 self.logger.info("FFP_TEST_FINISH_0005")
564 def test_ip4_tx_p2p_subif_drop(self):
565 """drop tx ip4 packet not matching p2p subinterface"""
566 self.logger.info("FFP_TEST_START_0006")
570 src_mac="02:01:00:00:ff:ff",
571 dst_mac=self.pg0.local_mac,
578 self.send_packets(self.pg0, self.pg1, count=0)
579 self.logger.info("FFP_TEST_FINISH_0006")
582 if __name__ == "__main__":
583 unittest.main(testRunner=VppTestRunner)