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 util import mactobinary
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, mactobinary(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,
41 mactobinary(remote_mac))
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,
82 mactobinary(macs[i-1]),
85 print "Failed to create subif %d %s" % (i, macs[i-1])
88 intfs = self.vapi.cli("show interface").split("\n")
91 if intf.startswith('pg2.'):
93 self.assertEqual(count, clients)
95 self.logger.info("FFP_TEST_FINISH_0001")
98 class P2PEthernetIPV6(VppTestCase):
99 """P2P Ethernet IPv6 tests"""
106 super(P2PEthernetIPV6, cls).setUpClass()
108 # Create pg interfaces
109 cls.create_pg_interfaces(range(3))
112 cls.pg_if_packet_sizes = [64, 512, 1518, 9018]
114 # Set up all interfaces
115 for i in cls.pg_interfaces:
118 cls.pg0.generate_remote_hosts(3)
119 cls.pg0.configure_ipv6_neighbors()
122 cls.pg1.generate_remote_hosts(3)
123 cls.pg1.configure_ipv6_neighbors()
124 cls.pg1.disable_ipv6_ra()
127 super(P2PEthernetIPV6, self).setUp()
128 for p in self.packets:
129 self.packets.remove(p)
130 self.p2p_sub_ifs.append(
131 self.create_p2p_ethernet(self.pg0, 1,
132 self.pg0._remote_hosts[0].mac))
133 self.p2p_sub_ifs.append(
134 self.create_p2p_ethernet(self.pg0, 2,
135 self.pg0._remote_hosts[1].mac))
136 self.vapi.cli("trace add p2p-ethernet-input 50")
139 while len(self.p2p_sub_ifs):
140 p2p = self.p2p_sub_ifs.pop()
141 self.delete_p2p_ethernet(p2p)
143 super(P2PEthernetIPV6, self).tearDown()
145 def create_p2p_ethernet(self, parent_if, sub_id, remote_mac):
146 p2p = VppP2PSubint(self, parent_if, sub_id, mactobinary(remote_mac))
149 p2p.disable_ipv6_ra()
152 def delete_p2p_ethernet(self, p2p):
155 self.vapi.delete_p2pethernet_subif(p2p.parent.sw_if_index,
158 def create_stream(self, src_mac=None, dst_mac=None,
159 src_ip=None, dst_ip=None, size=None):
162 pkt_size = random.choice(self.pg_if_packet_sizes)
163 p = Ether(src=src_mac, dst=dst_mac)
164 p /= IPv6(src=src_ip, dst=dst_ip)
165 p /= (UDP(sport=1234, dport=4321) / Raw('\xa5' * 20))
166 self.extend_packet(p, pkt_size)
169 def send_packets(self, src_if=None, dst_if=None, packets=None, count=None):
170 self.pg_enable_capture([dst_if])
172 packets = self.packets
173 src_if.add_stream(packets)
177 return dst_if.get_capture(count)
179 def test_no_p2p_subif(self):
180 """standard routing without p2p subinterfaces"""
181 self.logger.info("FFP_TEST_START_0001")
183 route_8000 = VppIpRoute(self, "8000::", 64,
184 [VppRoutePath(self.pg0.remote_ip6,
185 self.pg0.sw_if_index,
186 proto=DpoProto.DPO_PROTO_IP6)],
188 route_8000.add_vpp_config()
190 self.packets = [(Ether(dst=self.pg1.local_mac,
191 src=self.pg1.remote_mac) /
192 IPv6(src="3001::1", dst="8000::100") /
193 UDP(sport=1234, dport=1234) /
195 self.send_packets(self.pg1, self.pg0)
197 self.logger.info("FFP_TEST_FINISH_0001")
199 def test_ip6_rx_p2p_subif(self):
200 """receive ipv6 packet via p2p subinterface"""
201 self.logger.info("FFP_TEST_START_0002")
203 route_9001 = VppIpRoute(self, "9001::", 64,
204 [VppRoutePath(self.pg1.remote_ip6,
205 self.pg1.sw_if_index,
206 proto=DpoProto.DPO_PROTO_IP6)],
208 route_9001.add_vpp_config()
211 self.create_stream(src_mac=self.pg0._remote_hosts[0].mac,
212 dst_mac=self.pg0.local_mac,
213 src_ip=self.p2p_sub_ifs[0].remote_ip6,
216 self.send_packets(self.pg0, self.pg1, self.packets)
217 self.assert_packet_counter_equal('p2p-ethernet-input', 1)
219 route_9001.remove_vpp_config()
220 self.logger.info("FFP_TEST_FINISH_0002")
222 def test_ip6_rx_p2p_subif_route(self):
223 """route rx ip6 packet not matching p2p subinterface"""
224 self.logger.info("FFP_TEST_START_0003")
226 self.pg0.config_ip6()
228 route_3 = VppIpRoute(self, "9000::", 64,
229 [VppRoutePath(self.pg1._remote_hosts[0].ip6,
230 self.pg1.sw_if_index,
231 proto=DpoProto.DPO_PROTO_IP6)],
233 route_3.add_vpp_config()
236 self.create_stream(src_mac="02:03:00:00:ff:ff",
237 dst_mac=self.pg0.local_mac,
241 self.send_packets(self.pg0, self.pg1)
243 self.pg0.unconfig_ip6()
245 route_3.remove_vpp_config()
247 self.logger.info("FFP_TEST_FINISH_0003")
249 def test_ip6_rx_p2p_subif_drop(self):
250 """drop rx packet not matching p2p subinterface"""
251 self.logger.info("FFP_TEST_START_0004")
253 route_9001 = VppIpRoute(self, "9000::", 64,
254 [VppRoutePath(self.pg1._remote_hosts[0].ip6,
255 self.pg1.sw_if_index,
256 proto=DpoProto.DPO_PROTO_IP6)],
258 route_9001.add_vpp_config()
261 self.create_stream(src_mac="02:03:00:00:ff:ff",
262 dst_mac=self.pg0.local_mac,
267 self.send_packets(self.pg0, self.pg1, count=0)
268 self.logger.info("FFP_TEST_FINISH_0004")
270 def test_ip6_tx_p2p_subif(self):
271 """send packet via p2p subinterface"""
272 self.logger.info("FFP_TEST_START_0005")
274 route_8000 = VppIpRoute(self, "8000::", 64,
275 [VppRoutePath(self.pg0.remote_ip6,
276 self.pg0.sw_if_index,
277 proto=DpoProto.DPO_PROTO_IP6)],
279 route_8000.add_vpp_config()
280 route_8001 = VppIpRoute(self, "8001::", 64,
281 [VppRoutePath(self.p2p_sub_ifs[0].remote_ip6,
282 self.p2p_sub_ifs[0].sw_if_index,
283 proto=DpoProto.DPO_PROTO_IP6)],
285 route_8001.add_vpp_config()
286 route_8002 = VppIpRoute(self, "8002::", 64,
287 [VppRoutePath(self.p2p_sub_ifs[1].remote_ip6,
288 self.p2p_sub_ifs[1].sw_if_index,
289 proto=DpoProto.DPO_PROTO_IP6)],
291 route_8002.add_vpp_config()
293 for i in range(0, 3):
295 self.create_stream(src_mac=self.pg1.remote_mac,
296 dst_mac=self.pg1.local_mac,
297 src_ip=self.pg1.remote_ip6,
298 dst_ip="800%d::100" % i))
300 self.send_packets(self.pg1, self.pg0, count=3)
302 route_8000.remove_vpp_config()
303 route_8001.remove_vpp_config()
304 route_8002.remove_vpp_config()
306 self.logger.info("FFP_TEST_FINISH_0005")
308 def test_ip6_tx_p2p_subif_drop(self):
309 """drop tx ip6 packet not matching p2p subinterface"""
310 self.logger.info("FFP_TEST_START_0006")
313 self.create_stream(src_mac="02:03:00:00:ff:ff",
314 dst_mac=self.pg0.local_mac,
319 self.send_packets(self.pg0, self.pg1, count=0)
320 self.logger.info("FFP_TEST_FINISH_0006")
323 class P2PEthernetIPV4(VppTestCase):
324 """P2P Ethernet IPv4 tests"""
331 super(P2PEthernetIPV4, cls).setUpClass()
333 # Create pg interfaces
334 cls.create_pg_interfaces(range(3))
337 cls.pg_if_packet_sizes = [64, 512, 1518, 9018]
339 # Set up all interfaces
340 for i in cls.pg_interfaces:
344 cls.pg0.generate_remote_hosts(5)
345 cls.pg0.configure_ipv4_neighbors()
348 cls.pg1.generate_remote_hosts(5)
349 cls.pg1.configure_ipv4_neighbors()
352 super(P2PEthernetIPV4, self).setUp()
353 for p in self.packets:
354 self.packets.remove(p)
355 self.p2p_sub_ifs.append(
356 self.create_p2p_ethernet(self.pg0, 1,
357 self.pg0._remote_hosts[0].mac))
358 self.p2p_sub_ifs.append(
359 self.create_p2p_ethernet(self.pg0, 2,
360 self.pg0._remote_hosts[1].mac))
361 self.vapi.cli("trace add p2p-ethernet-input 50")
364 while len(self.p2p_sub_ifs):
365 p2p = self.p2p_sub_ifs.pop()
366 self.delete_p2p_ethernet(p2p)
367 super(P2PEthernetIPV4, self).tearDown()
369 def create_stream(self, src_mac=None, dst_mac=None,
370 src_ip=None, dst_ip=None, size=None):
373 pkt_size = random.choice(self.pg_if_packet_sizes)
374 p = Ether(src=src_mac, dst=dst_mac)
375 p /= IP(src=src_ip, dst=dst_ip)
376 p /= (UDP(sport=1234, dport=4321) / Raw('\xa5' * 20))
377 self.extend_packet(p, pkt_size)
380 def send_packets(self, src_if=None, dst_if=None, packets=None, count=None):
381 self.pg_enable_capture([dst_if])
383 packets = self.packets
384 src_if.add_stream(packets)
388 return dst_if.get_capture(count)
390 def create_p2p_ethernet(self, parent_if, sub_id, remote_mac):
391 p2p = VppP2PSubint(self, parent_if, sub_id, mactobinary(remote_mac))
396 def delete_p2p_ethernet(self, p2p):
399 self.vapi.delete_p2pethernet_subif(p2p.parent.sw_if_index,
402 def test_ip4_rx_p2p_subif(self):
403 """receive ipv4 packet via p2p subinterface"""
404 self.logger.info("FFP_TEST_START_0002")
406 route_9000 = VppIpRoute(self, "9.0.0.0", 16,
407 [VppRoutePath(self.pg1.remote_ip4,
408 self.pg1.sw_if_index)])
409 route_9000.add_vpp_config()
412 self.create_stream(src_mac=self.pg0._remote_hosts[0].mac,
413 dst_mac=self.pg0.local_mac,
414 src_ip=self.p2p_sub_ifs[0].remote_ip4,
417 self.send_packets(self.pg0, self.pg1, self.packets)
419 self.assert_packet_counter_equal('p2p-ethernet-input', 1)
421 route_9000.remove_vpp_config()
422 self.logger.info("FFP_TEST_FINISH_0002")
424 def test_ip4_rx_p2p_subif_route(self):
425 """route rx packet not matching p2p subinterface"""
426 self.logger.info("FFP_TEST_START_0003")
428 route_9001 = VppIpRoute(self, "9.0.0.0", 24,
429 [VppRoutePath(self.pg1.remote_ip4,
430 self.pg1.sw_if_index)])
431 route_9001.add_vpp_config()
434 self.create_stream(src_mac="02:01:00:00:ff:ff",
435 dst_mac=self.pg0.local_mac,
439 self.send_packets(self.pg0, self.pg1)
441 route_9001.remove_vpp_config()
443 self.logger.info("FFP_TEST_FINISH_0003")
445 def test_ip4_tx_p2p_subif(self):
446 """send ip4 packet via p2p subinterface"""
447 self.logger.info("FFP_TEST_START_0005")
449 route_9100 = VppIpRoute(self, "9.1.0.100", 24,
450 [VppRoutePath(self.pg0.remote_ip4,
451 self.pg0.sw_if_index,
453 route_9100.add_vpp_config()
454 route_9200 = VppIpRoute(self, "9.2.0.100", 24,
455 [VppRoutePath(self.p2p_sub_ifs[0].remote_ip4,
456 self.p2p_sub_ifs[0].sw_if_index,
458 route_9200.add_vpp_config()
459 route_9300 = VppIpRoute(self, "9.3.0.100", 24,
460 [VppRoutePath(self.p2p_sub_ifs[1].remote_ip4,
461 self.p2p_sub_ifs[1].sw_if_index
463 route_9300.add_vpp_config()
465 for i in range(0, 3):
467 self.create_stream(src_mac=self.pg1.remote_mac,
468 dst_mac=self.pg1.local_mac,
469 src_ip=self.pg1.remote_ip4,
470 dst_ip="9.%d.0.100" % (i+1)))
472 self.send_packets(self.pg1, self.pg0)
474 # route_7000.remove_vpp_config()
475 route_9100.remove_vpp_config()
476 route_9200.remove_vpp_config()
477 route_9300.remove_vpp_config()
479 self.logger.info("FFP_TEST_FINISH_0005")
481 def test_ip4_tx_p2p_subif_drop(self):
482 """drop tx ip4 packet not matching p2p subinterface"""
483 self.logger.info("FFP_TEST_START_0006")
486 self.create_stream(src_mac="02:01:00:00:ff:ff",
487 dst_mac=self.pg0.local_mac,
492 self.send_packets(self.pg0, self.pg1, count=0)
493 self.logger.info("FFP_TEST_FINISH_0006")
496 if __name__ == '__main__':
497 unittest.main(testRunner=VppTestRunner)