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 verify_counters(self, counter_id, expected_value):
180 counters = self.vapi.cli("sh errors").split('\n')
182 for i in range(1, len(counters)-1):
183 results = counters[i].split()
184 if results[1] == counter_id:
185 counter_value = int(results[0])
187 self.assertEqual(counter_value, expected_value)
189 def test_no_p2p_subif(self):
190 """standard routing without p2p subinterfaces"""
191 self.logger.info("FFP_TEST_START_0001")
193 route_8000 = VppIpRoute(self, "8000::", 64,
194 [VppRoutePath(self.pg0.remote_ip6,
195 self.pg0.sw_if_index,
196 proto=DpoProto.DPO_PROTO_IP6)],
198 route_8000.add_vpp_config()
200 self.packets = [(Ether(dst=self.pg1.local_mac,
201 src=self.pg1.remote_mac) /
202 IPv6(src="3001::1", dst="8000::100") /
203 UDP(sport=1234, dport=1234) /
205 self.send_packets(self.pg1, self.pg0)
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 proto=DpoProto.DPO_PROTO_IP6)],
218 route_9001.add_vpp_config()
221 self.create_stream(src_mac=self.pg0._remote_hosts[0].mac,
222 dst_mac=self.pg0.local_mac,
223 src_ip=self.p2p_sub_ifs[0].remote_ip6,
226 self.send_packets(self.pg0, self.pg1, self.packets)
227 self.verify_counters('p2p-ethernet-input', 1)
229 route_9001.remove_vpp_config()
230 self.logger.info("FFP_TEST_FINISH_0002")
232 def test_ip6_rx_p2p_subif_route(self):
233 """route rx ip6 packet not matching p2p subinterface"""
234 self.logger.info("FFP_TEST_START_0003")
236 self.pg0.config_ip6()
238 route_3 = VppIpRoute(self, "9000::", 64,
239 [VppRoutePath(self.pg1._remote_hosts[0].ip6,
240 self.pg1.sw_if_index,
241 proto=DpoProto.DPO_PROTO_IP6)],
243 route_3.add_vpp_config()
246 self.create_stream(src_mac="02:03:00:00:ff:ff",
247 dst_mac=self.pg0.local_mac,
251 self.send_packets(self.pg0, self.pg1)
253 self.pg0.unconfig_ip6()
255 route_3.remove_vpp_config()
257 self.logger.info("FFP_TEST_FINISH_0003")
259 def test_ip6_rx_p2p_subif_drop(self):
260 """drop rx packet not matching p2p subinterface"""
261 self.logger.info("FFP_TEST_START_0004")
263 route_9001 = VppIpRoute(self, "9000::", 64,
264 [VppRoutePath(self.pg1._remote_hosts[0].ip6,
265 self.pg1.sw_if_index,
266 proto=DpoProto.DPO_PROTO_IP6)],
268 route_9001.add_vpp_config()
271 self.create_stream(src_mac="02:03:00:00:ff:ff",
272 dst_mac=self.pg0.local_mac,
277 self.send_packets(self.pg0, self.pg1, count=0)
278 self.logger.info("FFP_TEST_FINISH_0004")
280 def test_ip6_tx_p2p_subif(self):
281 """send packet via p2p subinterface"""
282 self.logger.info("FFP_TEST_START_0005")
284 route_8000 = VppIpRoute(self, "8000::", 64,
285 [VppRoutePath(self.pg0.remote_ip6,
286 self.pg0.sw_if_index,
287 proto=DpoProto.DPO_PROTO_IP6)],
289 route_8000.add_vpp_config()
290 route_8001 = VppIpRoute(self, "8001::", 64,
291 [VppRoutePath(self.p2p_sub_ifs[0].remote_ip6,
292 self.p2p_sub_ifs[0].sw_if_index,
293 proto=DpoProto.DPO_PROTO_IP6)],
295 route_8001.add_vpp_config()
296 route_8002 = VppIpRoute(self, "8002::", 64,
297 [VppRoutePath(self.p2p_sub_ifs[1].remote_ip6,
298 self.p2p_sub_ifs[1].sw_if_index,
299 proto=DpoProto.DPO_PROTO_IP6)],
301 route_8002.add_vpp_config()
303 for i in range(0, 3):
305 self.create_stream(src_mac=self.pg1.remote_mac,
306 dst_mac=self.pg1.local_mac,
307 src_ip=self.pg1.remote_ip6,
308 dst_ip="800%d::100" % i))
310 self.send_packets(self.pg1, self.pg0, count=3)
312 route_8000.remove_vpp_config()
313 route_8001.remove_vpp_config()
314 route_8002.remove_vpp_config()
316 self.logger.info("FFP_TEST_FINISH_0005")
318 def test_ip6_tx_p2p_subif_drop(self):
319 """drop tx ip6 packet not matching p2p subinterface"""
320 self.logger.info("FFP_TEST_START_0006")
323 self.create_stream(src_mac="02:03:00:00:ff:ff",
324 dst_mac=self.pg0.local_mac,
329 self.send_packets(self.pg0, self.pg1, count=0)
330 self.logger.info("FFP_TEST_FINISH_0006")
333 class P2PEthernetIPV4(VppTestCase):
334 """P2P Ethernet IPv4 tests"""
341 super(P2PEthernetIPV4, cls).setUpClass()
343 # Create pg interfaces
344 cls.create_pg_interfaces(range(3))
347 cls.pg_if_packet_sizes = [64, 512, 1518, 9018]
349 # Set up all interfaces
350 for i in cls.pg_interfaces:
354 cls.pg0.generate_remote_hosts(5)
355 cls.pg0.configure_ipv4_neighbors()
358 cls.pg1.generate_remote_hosts(5)
359 cls.pg1.configure_ipv4_neighbors()
362 super(P2PEthernetIPV4, self).setUp()
363 for p in self.packets:
364 self.packets.remove(p)
365 self.p2p_sub_ifs.append(
366 self.create_p2p_ethernet(self.pg0, 1,
367 self.pg0._remote_hosts[0].mac))
368 self.p2p_sub_ifs.append(
369 self.create_p2p_ethernet(self.pg0, 2,
370 self.pg0._remote_hosts[1].mac))
371 self.vapi.cli("trace add p2p-ethernet-input 50")
374 while len(self.p2p_sub_ifs):
375 p2p = self.p2p_sub_ifs.pop()
376 self.delete_p2p_ethernet(p2p)
377 super(P2PEthernetIPV4, self).tearDown()
379 def create_stream(self, src_mac=None, dst_mac=None,
380 src_ip=None, dst_ip=None, size=None):
383 pkt_size = random.choice(self.pg_if_packet_sizes)
384 p = Ether(src=src_mac, dst=dst_mac)
385 p /= IP(src=src_ip, dst=dst_ip)
386 p /= (UDP(sport=1234, dport=4321) / Raw('\xa5' * 20))
387 self.extend_packet(p, pkt_size)
390 def send_packets(self, src_if=None, dst_if=None, packets=None, count=None):
391 self.pg_enable_capture([dst_if])
393 packets = self.packets
394 src_if.add_stream(packets)
398 return dst_if.get_capture(count)
400 def verify_counters(self, counter_id, expected_value):
401 counters = self.vapi.cli("sh errors").split('\n')
403 for i in range(1, len(counters)-1):
404 results = counters[i].split()
405 if results[1] == counter_id:
406 counter_value = int(results[0])
408 self.assertEqual(counter_value, expected_value)
410 def create_p2p_ethernet(self, parent_if, sub_id, remote_mac):
411 p2p = VppP2PSubint(self, parent_if, sub_id, mactobinary(remote_mac))
416 def delete_p2p_ethernet(self, p2p):
419 self.vapi.delete_p2pethernet_subif(p2p.parent.sw_if_index,
422 def test_ip4_rx_p2p_subif(self):
423 """receive ipv4 packet via p2p subinterface"""
424 self.logger.info("FFP_TEST_START_0002")
426 route_9000 = VppIpRoute(self, "9.0.0.0", 16,
427 [VppRoutePath(self.pg1.remote_ip4,
428 self.pg1.sw_if_index)])
429 route_9000.add_vpp_config()
432 self.create_stream(src_mac=self.pg0._remote_hosts[0].mac,
433 dst_mac=self.pg0.local_mac,
434 src_ip=self.p2p_sub_ifs[0].remote_ip4,
437 self.send_packets(self.pg0, self.pg1, self.packets)
439 self.verify_counters('p2p-ethernet-input', 1)
441 route_9000.remove_vpp_config()
442 self.logger.info("FFP_TEST_FINISH_0002")
444 def test_ip4_rx_p2p_subif_route(self):
445 """route rx packet not matching p2p subinterface"""
446 self.logger.info("FFP_TEST_START_0003")
448 route_9001 = VppIpRoute(self, "9.0.0.0", 24,
449 [VppRoutePath(self.pg1.remote_ip4,
450 self.pg1.sw_if_index)])
451 route_9001.add_vpp_config()
454 self.create_stream(src_mac="02:01:00:00:ff:ff",
455 dst_mac=self.pg0.local_mac,
459 self.send_packets(self.pg0, self.pg1)
461 route_9001.remove_vpp_config()
463 self.logger.info("FFP_TEST_FINISH_0003")
465 def test_ip4_tx_p2p_subif(self):
466 """send ip4 packet via p2p subinterface"""
467 self.logger.info("FFP_TEST_START_0005")
469 route_9100 = VppIpRoute(self, "9.1.0.100", 24,
470 [VppRoutePath(self.pg0.remote_ip4,
471 self.pg0.sw_if_index,
473 route_9100.add_vpp_config()
474 route_9200 = VppIpRoute(self, "9.2.0.100", 24,
475 [VppRoutePath(self.p2p_sub_ifs[0].remote_ip4,
476 self.p2p_sub_ifs[0].sw_if_index,
478 route_9200.add_vpp_config()
479 route_9300 = VppIpRoute(self, "9.3.0.100", 24,
480 [VppRoutePath(self.p2p_sub_ifs[1].remote_ip4,
481 self.p2p_sub_ifs[1].sw_if_index
483 route_9300.add_vpp_config()
485 for i in range(0, 3):
487 self.create_stream(src_mac=self.pg1.remote_mac,
488 dst_mac=self.pg1.local_mac,
489 src_ip=self.pg1.remote_ip4,
490 dst_ip="9.%d.0.100" % (i+1)))
492 self.send_packets(self.pg1, self.pg0)
494 # route_7000.remove_vpp_config()
495 route_9100.remove_vpp_config()
496 route_9200.remove_vpp_config()
497 route_9300.remove_vpp_config()
499 self.logger.info("FFP_TEST_FINISH_0005")
501 def test_ip4_tx_p2p_subif_drop(self):
502 """drop tx ip4 packet not matching p2p subinterface"""
503 self.logger.info("FFP_TEST_START_0006")
506 self.create_stream(src_mac="02:01:00:00:ff:ff",
507 dst_mac=self.pg0.local_mac,
512 self.send_packets(self.pg0, self.pg1, count=0)
513 self.logger.info("FFP_TEST_FINISH_0006")
516 if __name__ == '__main__':
517 unittest.main(testRunner=VppTestRunner)