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.p2p_ethernet_del(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.assertIn('pg0.1', intfs)
52 self.assertIn('pg0.2', intfs)
53 self.assertNotIn('pg0.5', intfs)
56 self.create_p2p_ethernet(self.pg0, 5, "de:ad:00:00:00:ff")
57 intfs = self.vapi.cli("show interface")
58 self.assertIn('pg0.5', intfs)
60 self.delete_p2p_ethernet(self.pg0, "de:ad:00:00:00:ff")
62 intfs = self.vapi.cli("show interface")
64 self.assertIn('pg0.1', intfs)
65 self.assertIn('pg0.2', intfs)
66 self.assertNotIn('pg0.5', intfs)
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(
82 self.vapi.p2p_ethernet_add(self.pg2.sw_if_index,
86 self.logger.info("Failed to create subif %d %s" % (
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 super(P2PEthernetIPV6, self).setUp()
130 for p in self.packets:
131 self.packets.remove(p)
132 self.p2p_sub_ifs.append(
133 self.create_p2p_ethernet(self.pg0, 1,
134 self.pg0._remote_hosts[0].mac))
135 self.p2p_sub_ifs.append(
136 self.create_p2p_ethernet(self.pg0, 2,
137 self.pg0._remote_hosts[1].mac))
138 self.vapi.cli("trace add p2p-ethernet-input 50")
141 while len(self.p2p_sub_ifs):
142 p2p = self.p2p_sub_ifs.pop()
143 self.delete_p2p_ethernet(p2p)
145 super(P2PEthernetIPV6, self).tearDown()
147 def create_p2p_ethernet(self, parent_if, sub_id, remote_mac):
148 p2p = VppP2PSubint(self, parent_if, sub_id, mac_pton(remote_mac))
151 p2p.disable_ipv6_ra()
154 def delete_p2p_ethernet(self, p2p):
157 self.vapi.p2p_ethernet_del(p2p.parent.sw_if_index,
160 def create_stream(self, src_mac=None, dst_mac=None,
161 src_ip=None, dst_ip=None, size=None):
164 pkt_size = random.choice(self.pg_if_packet_sizes)
165 p = Ether(src=src_mac, dst=dst_mac)
166 p /= IPv6(src=src_ip, dst=dst_ip)
167 p /= (UDP(sport=1234, dport=4321) / Raw('\xa5' * 20))
168 self.extend_packet(p, pkt_size)
171 def send_packets(self, src_if=None, dst_if=None, packets=None, count=None):
172 self.pg_enable_capture([dst_if])
174 packets = self.packets
175 src_if.add_stream(packets)
179 return dst_if.get_capture(count)
181 def test_no_p2p_subif(self):
182 """standard routing without p2p subinterfaces"""
183 self.logger.info("FFP_TEST_START_0001")
185 route_8000 = VppIpRoute(self, "8000::", 64,
186 [VppRoutePath(self.pg0.remote_ip6,
187 self.pg0.sw_if_index,
188 proto=DpoProto.DPO_PROTO_IP6)],
190 route_8000.add_vpp_config()
192 self.packets = [(Ether(dst=self.pg1.local_mac,
193 src=self.pg1.remote_mac) /
194 IPv6(src="3001::1", dst="8000::100") /
195 UDP(sport=1234, dport=1234) /
197 self.send_packets(self.pg1, self.pg0)
199 self.logger.info("FFP_TEST_FINISH_0001")
201 def test_ip6_rx_p2p_subif(self):
202 """receive ipv6 packet via p2p subinterface"""
203 self.logger.info("FFP_TEST_START_0002")
205 route_9001 = VppIpRoute(self, "9001::", 64,
206 [VppRoutePath(self.pg1.remote_ip6,
207 self.pg1.sw_if_index,
208 proto=DpoProto.DPO_PROTO_IP6)],
210 route_9001.add_vpp_config()
213 self.create_stream(src_mac=self.pg0._remote_hosts[0].mac,
214 dst_mac=self.pg0.local_mac,
215 src_ip=self.p2p_sub_ifs[0].remote_ip6,
218 self.send_packets(self.pg0, self.pg1, self.packets)
219 self.assert_packet_counter_equal('p2p-ethernet-input', 1)
221 route_9001.remove_vpp_config()
222 self.logger.info("FFP_TEST_FINISH_0002")
224 def test_ip6_rx_p2p_subif_route(self):
225 """route rx ip6 packet not matching p2p subinterface"""
226 self.logger.info("FFP_TEST_START_0003")
228 self.pg0.config_ip6()
230 route_3 = VppIpRoute(self, "9000::", 64,
231 [VppRoutePath(self.pg1._remote_hosts[0].ip6,
232 self.pg1.sw_if_index,
233 proto=DpoProto.DPO_PROTO_IP6)],
235 route_3.add_vpp_config()
238 self.create_stream(src_mac="02:03:00:00:ff:ff",
239 dst_mac=self.pg0.local_mac,
243 self.send_packets(self.pg0, self.pg1)
245 self.pg0.unconfig_ip6()
247 route_3.remove_vpp_config()
249 self.logger.info("FFP_TEST_FINISH_0003")
251 def test_ip6_rx_p2p_subif_drop(self):
252 """drop rx packet not matching p2p subinterface"""
253 self.logger.info("FFP_TEST_START_0004")
255 route_9001 = VppIpRoute(self, "9000::", 64,
256 [VppRoutePath(self.pg1._remote_hosts[0].ip6,
257 self.pg1.sw_if_index,
258 proto=DpoProto.DPO_PROTO_IP6)],
260 route_9001.add_vpp_config()
263 self.create_stream(src_mac="02:03:00:00:ff:ff",
264 dst_mac=self.pg0.local_mac,
269 self.send_packets(self.pg0, self.pg1, count=0)
270 self.logger.info("FFP_TEST_FINISH_0004")
272 def test_ip6_tx_p2p_subif(self):
273 """send packet via p2p subinterface"""
274 self.logger.info("FFP_TEST_START_0005")
276 route_8000 = VppIpRoute(self, "8000::", 64,
277 [VppRoutePath(self.pg0.remote_ip6,
278 self.pg0.sw_if_index,
279 proto=DpoProto.DPO_PROTO_IP6)],
281 route_8000.add_vpp_config()
282 route_8001 = VppIpRoute(self, "8001::", 64,
283 [VppRoutePath(self.p2p_sub_ifs[0].remote_ip6,
284 self.p2p_sub_ifs[0].sw_if_index,
285 proto=DpoProto.DPO_PROTO_IP6)],
287 route_8001.add_vpp_config()
288 route_8002 = VppIpRoute(self, "8002::", 64,
289 [VppRoutePath(self.p2p_sub_ifs[1].remote_ip6,
290 self.p2p_sub_ifs[1].sw_if_index,
291 proto=DpoProto.DPO_PROTO_IP6)],
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.logger.info("FFP_TEST_FINISH_0005")
310 def test_ip6_tx_p2p_subif_drop(self):
311 """drop tx ip6 packet not matching p2p subinterface"""
312 self.logger.info("FFP_TEST_START_0006")
315 self.create_stream(src_mac="02:03:00:00:ff:ff",
316 dst_mac=self.pg0.local_mac,
321 self.send_packets(self.pg0, self.pg1, count=0)
322 self.logger.info("FFP_TEST_FINISH_0006")
325 class P2PEthernetIPV4(VppTestCase):
326 """P2P Ethernet IPv4 tests"""
333 super(P2PEthernetIPV4, cls).setUpClass()
335 # Create pg interfaces
336 cls.create_pg_interfaces(range(3))
339 cls.pg_if_packet_sizes = [64, 512, 1518, 9018]
341 # Set up all interfaces
342 for i in cls.pg_interfaces:
346 cls.pg0.generate_remote_hosts(5)
347 cls.pg0.configure_ipv4_neighbors()
350 cls.pg1.generate_remote_hosts(5)
351 cls.pg1.configure_ipv4_neighbors()
354 super(P2PEthernetIPV4, self).setUp()
355 for p in self.packets:
356 self.packets.remove(p)
357 self.p2p_sub_ifs.append(
358 self.create_p2p_ethernet(self.pg0, 1,
359 self.pg0._remote_hosts[0].mac))
360 self.p2p_sub_ifs.append(
361 self.create_p2p_ethernet(self.pg0, 2,
362 self.pg0._remote_hosts[1].mac))
363 self.vapi.cli("trace add p2p-ethernet-input 50")
366 while len(self.p2p_sub_ifs):
367 p2p = self.p2p_sub_ifs.pop()
368 self.delete_p2p_ethernet(p2p)
369 super(P2PEthernetIPV4, self).tearDown()
371 def create_stream(self, src_mac=None, dst_mac=None,
372 src_ip=None, dst_ip=None, size=None):
375 pkt_size = random.choice(self.pg_if_packet_sizes)
376 p = Ether(src=src_mac, dst=dst_mac)
377 p /= IP(src=src_ip, dst=dst_ip)
378 p /= (UDP(sport=1234, dport=4321) / Raw('\xa5' * 20))
379 self.extend_packet(p, pkt_size)
382 def send_packets(self, src_if=None, dst_if=None, packets=None, count=None):
383 self.pg_enable_capture([dst_if])
385 packets = self.packets
386 src_if.add_stream(packets)
390 return dst_if.get_capture(count)
392 def create_p2p_ethernet(self, parent_if, sub_id, remote_mac):
393 p2p = VppP2PSubint(self, parent_if, sub_id, mac_pton(remote_mac))
398 def delete_p2p_ethernet(self, p2p):
401 self.vapi.p2p_ethernet_del(p2p.parent.sw_if_index,
404 def test_ip4_rx_p2p_subif(self):
405 """receive ipv4 packet via p2p subinterface"""
406 self.logger.info("FFP_TEST_START_0002")
408 route_9000 = VppIpRoute(self, "9.0.0.0", 16,
409 [VppRoutePath(self.pg1.remote_ip4,
410 self.pg1.sw_if_index)])
411 route_9000.add_vpp_config()
414 self.create_stream(src_mac=self.pg0._remote_hosts[0].mac,
415 dst_mac=self.pg0.local_mac,
416 src_ip=self.p2p_sub_ifs[0].remote_ip4,
419 self.send_packets(self.pg0, self.pg1, self.packets)
421 self.assert_packet_counter_equal('p2p-ethernet-input', 1)
423 route_9000.remove_vpp_config()
424 self.logger.info("FFP_TEST_FINISH_0002")
426 def test_ip4_rx_p2p_subif_route(self):
427 """route rx packet not matching p2p subinterface"""
428 self.logger.info("FFP_TEST_START_0003")
430 route_9001 = VppIpRoute(self, "9.0.0.0", 24,
431 [VppRoutePath(self.pg1.remote_ip4,
432 self.pg1.sw_if_index)])
433 route_9001.add_vpp_config()
436 self.create_stream(src_mac="02:01:00:00:ff:ff",
437 dst_mac=self.pg0.local_mac,
441 self.send_packets(self.pg0, self.pg1)
443 route_9001.remove_vpp_config()
445 self.logger.info("FFP_TEST_FINISH_0003")
447 def test_ip4_tx_p2p_subif(self):
448 """send ip4 packet via p2p subinterface"""
449 self.logger.info("FFP_TEST_START_0005")
451 route_9100 = VppIpRoute(self, "9.1.0.100", 24,
452 [VppRoutePath(self.pg0.remote_ip4,
453 self.pg0.sw_if_index,
455 route_9100.add_vpp_config()
456 route_9200 = VppIpRoute(self, "9.2.0.100", 24,
457 [VppRoutePath(self.p2p_sub_ifs[0].remote_ip4,
458 self.p2p_sub_ifs[0].sw_if_index,
460 route_9200.add_vpp_config()
461 route_9300 = VppIpRoute(self, "9.3.0.100", 24,
462 [VppRoutePath(self.p2p_sub_ifs[1].remote_ip4,
463 self.p2p_sub_ifs[1].sw_if_index
465 route_9300.add_vpp_config()
467 for i in range(0, 3):
469 self.create_stream(src_mac=self.pg1.remote_mac,
470 dst_mac=self.pg1.local_mac,
471 src_ip=self.pg1.remote_ip4,
472 dst_ip="9.%d.0.100" % (i+1)))
474 self.send_packets(self.pg1, self.pg0)
476 # route_7000.remove_vpp_config()
477 route_9100.remove_vpp_config()
478 route_9200.remove_vpp_config()
479 route_9300.remove_vpp_config()
481 self.logger.info("FFP_TEST_FINISH_0005")
483 def test_ip4_tx_p2p_subif_drop(self):
484 """drop tx ip4 packet not matching p2p subinterface"""
485 self.logger.info("FFP_TEST_START_0006")
488 self.create_stream(src_mac="02:01:00:00:ff:ff",
489 dst_mac=self.pg0.local_mac,
494 self.send_packets(self.pg0, self.pg1, count=0)
495 self.logger.info("FFP_TEST_FINISH_0006")
498 if __name__ == '__main__':
499 unittest.main(testRunner=VppTestRunner)