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:
36 def tearDownClass(cls):
37 super(P2PEthernetAPI, cls).tearDownClass()
39 def create_p2p_ethernet(self, parent_if, sub_id, remote_mac):
40 p2p = VppP2PSubint(self, parent_if, sub_id, mac_pton(remote_mac))
41 self.p2p_sub_ifs.append(p2p)
43 def delete_p2p_ethernet(self, parent_if, remote_mac):
44 self.vapi.p2p_ethernet_del(parent_if.sw_if_index,
48 """delete/create p2p subif"""
49 self.logger.info("FFP_TEST_START_0000")
51 self.create_p2p_ethernet(self.pg0, 1, "de:ad:00:00:00:01")
52 self.create_p2p_ethernet(self.pg0, 2, "de:ad:00:00:00:02")
53 intfs = self.vapi.cli("show interface")
55 self.assertIn('pg0.1', intfs)
56 self.assertIn('pg0.2', intfs)
57 self.assertNotIn('pg0.5', intfs)
60 self.create_p2p_ethernet(self.pg0, 5, "de:ad:00:00:00:ff")
61 intfs = self.vapi.cli("show interface")
62 self.assertIn('pg0.5', intfs)
64 self.delete_p2p_ethernet(self.pg0, "de:ad:00:00:00:ff")
66 intfs = self.vapi.cli("show interface")
68 self.assertIn('pg0.1', intfs)
69 self.assertIn('pg0.2', intfs)
70 self.assertNotIn('pg0.5', intfs)
72 self.logger.info("FFP_TEST_FINISH_0000")
74 def test_p2p_subif_creation_1k(self):
75 """create 1k of p2p subifs"""
76 self.logger.info("FFP_TEST_START_0001")
80 mac = int("dead00000000", 16)
82 for i in range(1, clients+1):
84 macs.append(':'.join(re.findall('..', '{:02x}'.format(
86 self.vapi.p2p_ethernet_add(self.pg2.sw_if_index,
90 self.logger.info("Failed to create subif %d %s" % (
94 intfs = self.vapi.cli("show interface").split("\n")
97 if intf.startswith('pg2.'):
99 self.assertEqual(count, clients)
101 self.logger.info("FFP_TEST_FINISH_0001")
104 class P2PEthernetIPV6(VppTestCase):
105 """P2P Ethernet IPv6 tests"""
112 super(P2PEthernetIPV6, cls).setUpClass()
114 # Create pg interfaces
115 cls.create_pg_interfaces(range(3))
118 cls.pg_if_packet_sizes = [64, 512, 1518, 9018]
120 # Set up all interfaces
121 for i in cls.pg_interfaces:
124 cls.pg0.generate_remote_hosts(3)
125 cls.pg0.configure_ipv6_neighbors()
128 cls.pg1.generate_remote_hosts(3)
129 cls.pg1.configure_ipv6_neighbors()
130 cls.pg1.disable_ipv6_ra()
133 def tearDownClass(cls):
134 super(P2PEthernetIPV6, cls).tearDownClass()
137 super(P2PEthernetIPV6, self).setUp()
138 for p in self.packets:
139 self.packets.remove(p)
140 self.p2p_sub_ifs.append(
141 self.create_p2p_ethernet(self.pg0, 1,
142 self.pg0._remote_hosts[0].mac))
143 self.p2p_sub_ifs.append(
144 self.create_p2p_ethernet(self.pg0, 2,
145 self.pg0._remote_hosts[1].mac))
146 self.vapi.cli("trace add p2p-ethernet-input 50")
149 while len(self.p2p_sub_ifs):
150 p2p = self.p2p_sub_ifs.pop()
151 self.delete_p2p_ethernet(p2p)
153 super(P2PEthernetIPV6, self).tearDown()
155 def create_p2p_ethernet(self, parent_if, sub_id, remote_mac):
156 p2p = VppP2PSubint(self, parent_if, sub_id, mac_pton(remote_mac))
159 p2p.disable_ipv6_ra()
162 def delete_p2p_ethernet(self, p2p):
165 self.vapi.p2p_ethernet_del(p2p.parent.sw_if_index,
168 def create_stream(self, src_mac=None, dst_mac=None,
169 src_ip=None, dst_ip=None, size=None):
172 pkt_size = random.choice(self.pg_if_packet_sizes)
173 p = Ether(src=src_mac, dst=dst_mac)
174 p /= IPv6(src=src_ip, dst=dst_ip)
175 p /= (UDP(sport=1234, dport=4321) / Raw(b'\xa5' * 20))
176 self.extend_packet(p, pkt_size)
179 def send_packets(self, src_if=None, dst_if=None, packets=None, count=None):
180 self.pg_enable_capture([dst_if])
182 packets = self.packets
183 src_if.add_stream(packets)
187 return dst_if.get_capture(count)
189 def test_no_p2p_subif(self):
190 """standard routing without p2p subinterfaces"""
191 self.logger.info("FFP_TEST_START_0001")
193 self.pg0.config_ip6()
194 route_8000 = VppIpRoute(self, "8000::", 64,
195 [VppRoutePath(self.pg0.remote_ip6,
196 self.pg0.sw_if_index)])
197 route_8000.add_vpp_config()
199 self.packets = [(Ether(dst=self.pg1.local_mac,
200 src=self.pg1.remote_mac) /
201 IPv6(src="3001::1", dst="8000::100") /
202 UDP(sport=1234, dport=1234) /
204 self.send_packets(self.pg1, self.pg0)
206 self.pg0.unconfig_ip6()
207 self.logger.info("FFP_TEST_FINISH_0001")
209 def test_ip6_rx_p2p_subif(self):
210 """receive ipv6 packet via p2p subinterface"""
211 self.logger.info("FFP_TEST_START_0002")
213 route_9001 = VppIpRoute(self, "9001::", 64,
214 [VppRoutePath(self.pg1.remote_ip6,
215 self.pg1.sw_if_index)])
216 route_9001.add_vpp_config()
219 self.create_stream(src_mac=self.pg0._remote_hosts[0].mac,
220 dst_mac=self.pg0.local_mac,
221 src_ip=self.p2p_sub_ifs[0].remote_ip6,
224 self.send_packets(self.pg0, self.pg1, self.packets)
225 self.assert_packet_counter_equal('p2p-ethernet-input', 1)
227 route_9001.remove_vpp_config()
228 self.logger.info("FFP_TEST_FINISH_0002")
230 def test_ip6_rx_p2p_subif_route(self):
231 """route rx ip6 packet not matching p2p subinterface"""
232 self.logger.info("FFP_TEST_START_0003")
234 self.pg0.config_ip6()
236 route_3 = VppIpRoute(self, "9000::", 64,
237 [VppRoutePath(self.pg1._remote_hosts[0].ip6,
238 self.pg1.sw_if_index)])
239 route_3.add_vpp_config()
242 self.create_stream(src_mac="02:03:00:00:ff:ff",
243 dst_mac=self.pg0.local_mac,
247 self.send_packets(self.pg0, self.pg1)
249 self.pg0.unconfig_ip6()
251 route_3.remove_vpp_config()
253 self.logger.info("FFP_TEST_FINISH_0003")
255 def test_ip6_rx_p2p_subif_drop(self):
256 """drop rx packet not matching p2p subinterface"""
257 self.logger.info("FFP_TEST_START_0004")
259 route_9001 = VppIpRoute(self, "9000::", 64,
260 [VppRoutePath(self.pg1._remote_hosts[0].ip6,
261 self.pg1.sw_if_index)])
262 route_9001.add_vpp_config()
265 self.create_stream(src_mac="02:03:00:00:ff:ff",
266 dst_mac=self.pg0.local_mac,
271 self.send_packets(self.pg0, self.pg1, count=0)
272 self.logger.info("FFP_TEST_FINISH_0004")
274 def test_ip6_tx_p2p_subif(self):
275 """send packet via p2p subinterface"""
276 self.logger.info("FFP_TEST_START_0005")
278 self.pg0.config_ip6()
280 route_8000 = VppIpRoute(self, "8000::", 64,
281 [VppRoutePath(self.pg0.remote_hosts[0].ip6,
282 self.pg0.sw_if_index)])
283 route_8000.add_vpp_config()
284 route_8001 = VppIpRoute(self, "8001::", 64,
286 self.p2p_sub_ifs[0].remote_ip6,
287 self.p2p_sub_ifs[0].sw_if_index)])
288 route_8001.add_vpp_config()
289 route_8002 = VppIpRoute(self, "8002::", 64,
291 self.p2p_sub_ifs[1].remote_ip6,
292 self.p2p_sub_ifs[1].sw_if_index)])
293 route_8002.add_vpp_config()
295 for i in range(0, 3):
297 self.create_stream(src_mac=self.pg1.remote_mac,
298 dst_mac=self.pg1.local_mac,
299 src_ip=self.pg1.remote_ip6,
300 dst_ip="800%d::100" % i))
302 self.send_packets(self.pg1, self.pg0, count=3)
304 route_8000.remove_vpp_config()
305 route_8001.remove_vpp_config()
306 route_8002.remove_vpp_config()
308 self.pg0.unconfig_ip6()
309 self.logger.info("FFP_TEST_FINISH_0005")
311 def test_ip6_tx_p2p_subif_drop(self):
312 """drop tx ip6 packet not matching p2p subinterface"""
313 self.logger.info("FFP_TEST_START_0006")
316 self.create_stream(src_mac="02:03:00:00:ff:ff",
317 dst_mac=self.pg0.local_mac,
322 self.send_packets(self.pg0, self.pg1, count=0)
323 self.logger.info("FFP_TEST_FINISH_0006")
326 class P2PEthernetIPV4(VppTestCase):
327 """P2P Ethernet IPv4 tests"""
334 super(P2PEthernetIPV4, cls).setUpClass()
336 # Create pg interfaces
337 cls.create_pg_interfaces(range(3))
340 cls.pg_if_packet_sizes = [64, 512, 1518, 9018]
342 # Set up all interfaces
343 for i in cls.pg_interfaces:
347 cls.pg0.generate_remote_hosts(5)
348 cls.pg0.configure_ipv4_neighbors()
351 cls.pg1.generate_remote_hosts(5)
352 cls.pg1.configure_ipv4_neighbors()
355 def tearDownClass(cls):
356 super(P2PEthernetIPV4, cls).tearDownClass()
359 super(P2PEthernetIPV4, self).setUp()
360 for p in self.packets:
361 self.packets.remove(p)
362 self.p2p_sub_ifs.append(
363 self.create_p2p_ethernet(self.pg0, 1,
364 self.pg0._remote_hosts[0].mac))
365 self.p2p_sub_ifs.append(
366 self.create_p2p_ethernet(self.pg0, 2,
367 self.pg0._remote_hosts[1].mac))
368 self.vapi.cli("trace add p2p-ethernet-input 50")
371 while len(self.p2p_sub_ifs):
372 p2p = self.p2p_sub_ifs.pop()
373 self.delete_p2p_ethernet(p2p)
374 super(P2PEthernetIPV4, self).tearDown()
376 def create_stream(self, src_mac=None, dst_mac=None,
377 src_ip=None, dst_ip=None, size=None):
380 pkt_size = random.choice(self.pg_if_packet_sizes)
381 p = Ether(src=src_mac, dst=dst_mac)
382 p /= IP(src=src_ip, dst=dst_ip)
383 p /= (UDP(sport=1234, dport=4321) / Raw(b'\xa5' * 20))
384 self.extend_packet(p, pkt_size)
387 def send_packets(self, src_if=None, dst_if=None, packets=None, count=None):
388 self.pg_enable_capture([dst_if])
390 packets = self.packets
391 src_if.add_stream(packets)
395 return dst_if.get_capture(count)
397 def create_p2p_ethernet(self, parent_if, sub_id, remote_mac):
398 p2p = VppP2PSubint(self, parent_if, sub_id, mac_pton(remote_mac))
403 def delete_p2p_ethernet(self, p2p):
406 self.vapi.p2p_ethernet_del(p2p.parent.sw_if_index,
409 def test_ip4_rx_p2p_subif(self):
410 """receive ipv4 packet via p2p subinterface"""
411 self.logger.info("FFP_TEST_START_0002")
413 route_9000 = VppIpRoute(self, "9.0.0.0", 16,
414 [VppRoutePath(self.pg1.remote_ip4,
415 self.pg1.sw_if_index)])
416 route_9000.add_vpp_config()
419 self.create_stream(src_mac=self.pg0._remote_hosts[0].mac,
420 dst_mac=self.pg0.local_mac,
421 src_ip=self.p2p_sub_ifs[0].remote_ip4,
424 self.send_packets(self.pg0, self.pg1, self.packets)
426 self.assert_packet_counter_equal('p2p-ethernet-input', 1)
428 route_9000.remove_vpp_config()
429 self.logger.info("FFP_TEST_FINISH_0002")
431 def test_ip4_rx_p2p_subif_route(self):
432 """route rx packet not matching p2p subinterface"""
433 self.logger.info("FFP_TEST_START_0003")
435 route_9001 = VppIpRoute(self, "9.0.0.0", 24,
436 [VppRoutePath(self.pg1.remote_ip4,
437 self.pg1.sw_if_index)])
438 route_9001.add_vpp_config()
441 self.create_stream(src_mac="02:01:00:00:ff:ff",
442 dst_mac=self.pg0.local_mac,
446 self.send_packets(self.pg0, self.pg1)
448 route_9001.remove_vpp_config()
450 self.logger.info("FFP_TEST_FINISH_0003")
452 def test_ip4_tx_p2p_subif(self):
453 """send ip4 packet via p2p subinterface"""
454 self.logger.info("FFP_TEST_START_0005")
456 route_9100 = VppIpRoute(self, "9.1.0.100", 24,
457 [VppRoutePath(self.pg0.remote_ip4,
458 self.pg0.sw_if_index,
460 route_9100.add_vpp_config()
461 route_9200 = VppIpRoute(self, "9.2.0.100", 24,
462 [VppRoutePath(self.p2p_sub_ifs[0].remote_ip4,
463 self.p2p_sub_ifs[0].sw_if_index,
465 route_9200.add_vpp_config()
466 route_9300 = VppIpRoute(self, "9.3.0.100", 24,
467 [VppRoutePath(self.p2p_sub_ifs[1].remote_ip4,
468 self.p2p_sub_ifs[1].sw_if_index
470 route_9300.add_vpp_config()
472 for i in range(0, 3):
474 self.create_stream(src_mac=self.pg1.remote_mac,
475 dst_mac=self.pg1.local_mac,
476 src_ip=self.pg1.remote_ip4,
477 dst_ip="9.%d.0.100" % (i+1)))
479 self.send_packets(self.pg1, self.pg0)
481 # route_7000.remove_vpp_config()
482 route_9100.remove_vpp_config()
483 route_9200.remove_vpp_config()
484 route_9300.remove_vpp_config()
486 self.logger.info("FFP_TEST_FINISH_0005")
488 def test_ip4_tx_p2p_subif_drop(self):
489 """drop tx ip4 packet not matching p2p subinterface"""
490 self.logger.info("FFP_TEST_START_0006")
493 self.create_stream(src_mac="02:01:00:00:ff:ff",
494 dst_mac=self.pg0.local_mac,
499 self.send_packets(self.pg0, self.pg1, count=0)
500 self.logger.info("FFP_TEST_FINISH_0006")
503 if __name__ == '__main__':
504 unittest.main(testRunner=VppTestRunner)