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_route import VppIpRoute, VppRoutePath, DpoProto
15 from util import mactobinary
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:
34 def create_p2p_ethernet(self, parent_if, sub_id, remote_mac):
35 p2p = VppP2PSubint(self, parent_if, sub_id, mactobinary(remote_mac))
36 self.p2p_sub_ifs.append(p2p)
38 def delete_p2p_ethernet(self, parent_if, remote_mac):
39 self.vapi.delete_p2pethernet_subif(parent_if.sw_if_index,
40 mactobinary(remote_mac))
43 """delete/create p2p subif"""
44 self.logger.info("FFP_TEST_START_0000")
46 self.create_p2p_ethernet(self.pg0, 1, "de:ad:00:00:00:01")
47 self.create_p2p_ethernet(self.pg0, 2, "de:ad:00:00:00:02")
48 intfs = self.vapi.cli("show interface")
50 self.assertNotEqual(intfs.find('pg0.1'), -1)
51 self.assertNotEqual(intfs.find('pg0.2'), -1)
52 self.assertEqual(intfs.find('pg0.5'), -1)
55 self.create_p2p_ethernet(self.pg0, 5, "de:ad:00:00:00:ff")
56 intfs = self.vapi.cli("show interface")
57 self.assertNotEqual(intfs.find('pg0.5'), -1)
59 self.delete_p2p_ethernet(self.pg0, "de:ad:00:00:00:ff")
61 intfs = self.vapi.cli("show interface")
63 self.assertNotEqual(intfs.find('pg0.1'), -1)
64 self.assertNotEqual(intfs.find('pg0.2'), -1)
65 self.assertEqual(intfs.find('pg0.5'), -1)
67 self.logger.info("FFP_TEST_FINISH_0000")
69 def test_p2p_subif_creation_1k(self):
70 """create 1k of p2p subifs"""
71 self.logger.info("FFP_TEST_START_0001")
75 mac = int("dead00000000", 16)
77 for i in range(1, clients+1):
79 macs.append(':'.join(re.findall('..', '{:02x}'.format(mac+i))))
80 self.vapi.create_p2pethernet_subif(self.pg2.sw_if_index,
81 mactobinary(macs[i-1]),
84 print "Failed to create subif %d %s" % (i, macs[i-1])
87 intfs = self.vapi.cli("show interface").split("\n")
90 if intf.startswith('pg2.'):
92 self.assertEqual(count, clients)
94 self.logger.info("FFP_TEST_FINISH_0001")
97 class P2PEthernetIPV6(VppTestCase):
98 """P2P Ethernet IPv6 tests"""
105 super(P2PEthernetIPV6, cls).setUpClass()
107 # Create pg interfaces
108 cls.create_pg_interfaces(range(3))
111 cls.pg_if_packet_sizes = [64, 512, 1518, 9018]
113 # Set up all interfaces
114 for i in cls.pg_interfaces:
117 cls.pg0.generate_remote_hosts(3)
118 cls.pg0.configure_ipv6_neighbors()
121 cls.pg1.generate_remote_hosts(3)
122 cls.pg1.configure_ipv6_neighbors()
123 cls.pg1.disable_ipv6_ra()
126 super(P2PEthernetIPV6, self).setUp()
127 for p in self.packets:
128 self.packets.remove(p)
129 self.create_p2p_ethernet(self.pg0, 1, self.pg0._remote_hosts[0].mac)
130 self.create_p2p_ethernet(self.pg0, 2, self.pg0._remote_hosts[1].mac)
131 self.p2p_sub_ifs[0].config_ip6()
132 self.p2p_sub_ifs[1].config_ip6()
133 self.vapi.cli("trace add p2p-ethernet-input 50")
136 self.delete_p2p_ethernet(self.pg0, self.pg0._remote_hosts[0].mac)
137 self.delete_p2p_ethernet(self.pg0, self.pg0._remote_hosts[1].mac)
138 super(P2PEthernetIPV6, self).tearDown()
140 def create_p2p_ethernet(self, parent_if, sub_id, remote_mac):
141 p2p = VppP2PSubint(self, parent_if, sub_id, mactobinary(remote_mac))
144 p2p.disable_ipv6_ra()
145 self.p2p_sub_ifs.append(p2p)
147 def delete_p2p_ethernet(self, parent_if, remote_mac):
148 self.vapi.delete_p2pethernet_subif(parent_if.sw_if_index,
149 mactobinary(remote_mac))
151 def create_stream(self, src_mac=None, dst_mac=None,
152 src_ip=None, dst_ip=None, size=None):
155 pkt_size = random.choice(self.pg_if_packet_sizes)
156 p = Ether(src=src_mac, dst=dst_mac)
157 p /= IPv6(src=src_ip, dst=dst_ip)
158 p /= (UDP(sport=1234, dport=4321) / Raw('\xa5' * 20))
159 self.extend_packet(p, pkt_size)
162 def send_packets(self, src_if=None, dst_if=None, packets=None, count=None):
163 self.pg_enable_capture([dst_if])
165 packets = self.packets
166 src_if.add_stream(packets)
170 return dst_if.get_capture(count)
172 def verify_counters(self, counter_id, expected_value):
173 counters = self.vapi.cli("sh errors").split('\n')
175 for i in range(1, len(counters)-1):
176 results = counters[i].split()
177 if results[1] == counter_id:
178 counter_value = int(results[0])
180 self.assertEqual(counter_value, expected_value)
182 def test_no_p2p_subif(self):
183 """standard routing without p2p subinterfaces"""
184 self.logger.info("FFP_TEST_START_0001")
186 route_8000 = VppIpRoute(self, "8000::", 64,
187 [VppRoutePath(self.pg0.remote_ip6,
188 self.pg0.sw_if_index,
189 proto=DpoProto.DPO_PROTO_IP6)],
191 route_8000.add_vpp_config()
193 self.packets = [(Ether(dst=self.pg1.local_mac,
194 src=self.pg1.remote_mac) /
195 IPv6(src="3001::1", dst="8000::100") /
196 UDP(sport=1234, dport=1234) /
198 self.send_packets(self.pg1, self.pg0)
200 self.logger.info("FFP_TEST_FINISH_0001")
202 def test_ip6_rx_p2p_subif(self):
203 """receive ipv6 packet via p2p subinterface"""
204 self.logger.info("FFP_TEST_START_0002")
206 route_9001 = VppIpRoute(self, "9001::", 64,
207 [VppRoutePath(self.pg1.remote_ip6,
208 self.pg1.sw_if_index,
209 proto=DpoProto.DPO_PROTO_IP6)],
211 route_9001.add_vpp_config()
214 self.create_stream(src_mac=self.pg0._remote_hosts[0].mac,
215 dst_mac=self.pg0.local_mac,
216 src_ip=self.p2p_sub_ifs[0].remote_ip6,
219 self.send_packets(self.pg0, self.pg1, self.packets)
220 self.verify_counters('p2p-ethernet-input', 1)
222 route_9001.remove_vpp_config()
223 self.logger.info("FFP_TEST_FINISH_0002")
225 def test_ip6_rx_p2p_subif_route(self):
226 """route rx ip6 packet not matching p2p subinterface"""
227 self.logger.info("FFP_TEST_START_0003")
229 self.pg0.config_ip6()
231 route_3 = VppIpRoute(self, "9000::", 64,
232 [VppRoutePath(self.pg1._remote_hosts[0].ip6,
233 self.pg1.sw_if_index,
234 proto=DpoProto.DPO_PROTO_IP6)],
236 route_3.add_vpp_config()
239 self.create_stream(src_mac="02:03:00:00:ff:ff",
240 dst_mac=self.pg0.local_mac,
244 self.send_packets(self.pg0, self.pg1)
246 self.pg0.unconfig_ip6()
248 route_3.remove_vpp_config()
250 self.logger.info("FFP_TEST_FINISH_0003")
252 def test_ip6_rx_p2p_subif_drop(self):
253 """drop rx packet not matching p2p subinterface"""
254 self.logger.info("FFP_TEST_START_0004")
256 route_9001 = VppIpRoute(self, "9000::", 64,
257 [VppRoutePath(self.pg1._remote_hosts[0].ip6,
258 self.pg1.sw_if_index,
259 proto=DpoProto.DPO_PROTO_IP6)],
261 route_9001.add_vpp_config()
264 self.create_stream(src_mac="02:03:00:00:ff:ff",
265 dst_mac=self.pg0.local_mac,
270 self.send_packets(self.pg0, self.pg1, count=0)
271 self.logger.info("FFP_TEST_FINISH_0004")
273 def test_ip6_tx_p2p_subif(self):
274 """send packet via p2p subinterface"""
275 self.logger.info("FFP_TEST_START_0005")
277 route_8000 = VppIpRoute(self, "8000::", 64,
278 [VppRoutePath(self.pg0.remote_ip6,
279 self.pg0.sw_if_index,
280 proto=DpoProto.DPO_PROTO_IP6)],
282 route_8000.add_vpp_config()
283 route_8001 = VppIpRoute(self, "8001::", 64,
284 [VppRoutePath(self.p2p_sub_ifs[0].remote_ip6,
285 self.p2p_sub_ifs[0].sw_if_index,
286 proto=DpoProto.DPO_PROTO_IP6)],
288 route_8001.add_vpp_config()
289 route_8002 = VppIpRoute(self, "8002::", 64,
290 [VppRoutePath(self.p2p_sub_ifs[1].remote_ip6,
291 self.p2p_sub_ifs[1].sw_if_index,
292 proto=DpoProto.DPO_PROTO_IP6)],
294 route_8002.add_vpp_config()
296 for i in range(0, 3):
298 self.create_stream(src_mac=self.pg1.remote_mac,
299 dst_mac=self.pg1.local_mac,
300 src_ip=self.pg1.remote_ip6,
301 dst_ip="800%d::100" % i))
303 self.send_packets(self.pg1, self.pg0, count=3)
305 route_8000.remove_vpp_config()
306 route_8001.remove_vpp_config()
307 route_8002.remove_vpp_config()
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 super(P2PEthernetIPV4, self).setUp()
356 for p in self.packets:
357 self.packets.remove(p)
358 self.create_p2p_ethernet(self.pg0, 1, self.pg0._remote_hosts[0].mac)
359 self.create_p2p_ethernet(self.pg0, 2, self.pg0._remote_hosts[1].mac)
360 self.p2p_sub_ifs[0].config_ip4()
361 self.p2p_sub_ifs[1].config_ip4()
362 self.vapi.cli("trace add p2p-ethernet-input 50")
365 self.delete_p2p_ethernet(self.pg0, self.pg0._remote_hosts[0].mac)
366 self.delete_p2p_ethernet(self.pg0, self.pg0._remote_hosts[1].mac)
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 verify_counters(self, counter_id, expected_value):
391 counters = self.vapi.cli("sh errors").split('\n')
393 for i in range(1, len(counters)-1):
394 results = counters[i].split()
395 if results[1] == counter_id:
396 counter_value = int(results[0])
398 self.assertEqual(counter_value, expected_value)
400 def create_p2p_ethernet(self, parent_if, sub_id, remote_mac):
401 p2p = VppP2PSubint(self, parent_if, sub_id, mactobinary(remote_mac))
404 self.p2p_sub_ifs.append(p2p)
406 def delete_p2p_ethernet(self, parent_if, remote_mac):
407 self.vapi.delete_p2pethernet_subif(parent_if.sw_if_index,
408 mactobinary(remote_mac))
410 def test_ip4_rx_p2p_subif(self):
411 """receive ipv4 packet via p2p subinterface"""
412 self.logger.info("FFP_TEST_START_0002")
414 route_9000 = VppIpRoute(self, "9.0.0.0", 16,
415 [VppRoutePath(self.pg1.remote_ip4,
416 self.pg1.sw_if_index)])
417 route_9000.add_vpp_config()
420 self.create_stream(src_mac=self.pg0._remote_hosts[0].mac,
421 dst_mac=self.pg0.local_mac,
422 src_ip=self.p2p_sub_ifs[0].remote_ip4,
425 self.send_packets(self.pg0, self.pg1, self.packets)
427 self.verify_counters('p2p-ethernet-input', 1)
429 route_9000.remove_vpp_config()
430 self.logger.info("FFP_TEST_FINISH_0002")
432 def test_ip4_rx_p2p_subif_route(self):
433 """route rx packet not matching p2p subinterface"""
434 self.logger.info("FFP_TEST_START_0003")
436 route_9001 = VppIpRoute(self, "9.0.0.0", 24,
437 [VppRoutePath(self.pg1.remote_ip4,
438 self.pg1.sw_if_index)])
439 route_9001.add_vpp_config()
442 self.create_stream(src_mac="02:01:00:00:ff:ff",
443 dst_mac=self.pg0.local_mac,
447 self.send_packets(self.pg0, self.pg1)
449 route_9001.remove_vpp_config()
451 self.logger.info("FFP_TEST_FINISH_0003")
453 def test_ip4_tx_p2p_subif(self):
454 """send ip4 packet via p2p subinterface"""
455 self.logger.info("FFP_TEST_START_0005")
457 route_9100 = VppIpRoute(self, "9.1.0.100", 24,
458 [VppRoutePath(self.pg0.remote_ip4,
459 self.pg0.sw_if_index,
461 route_9100.add_vpp_config()
462 route_9200 = VppIpRoute(self, "9.2.0.100", 24,
463 [VppRoutePath(self.p2p_sub_ifs[0].remote_ip4,
464 self.p2p_sub_ifs[0].sw_if_index,
466 route_9200.add_vpp_config()
467 route_9300 = VppIpRoute(self, "9.3.0.100", 24,
468 [VppRoutePath(self.p2p_sub_ifs[1].remote_ip4,
469 self.p2p_sub_ifs[1].sw_if_index
471 route_9300.add_vpp_config()
473 for i in range(0, 3):
475 self.create_stream(src_mac=self.pg1.remote_mac,
476 dst_mac=self.pg1.local_mac,
477 src_ip=self.pg1.remote_ip4,
478 dst_ip="9.%d.0.100" % (i+1)))
480 self.send_packets(self.pg1, self.pg0)
482 # route_7000.remove_vpp_config()
483 route_9100.remove_vpp_config()
484 route_9200.remove_vpp_config()
485 route_9300.remove_vpp_config()
487 self.logger.info("FFP_TEST_FINISH_0005")
489 def test_ip4_tx_p2p_subif_drop(self):
490 """drop tx ip4 packet not matching p2p subinterface"""
491 self.logger.info("FFP_TEST_START_0006")
494 self.create_stream(src_mac="02:01:00:00:ff:ff",
495 dst_mac=self.pg0.local_mac,
500 self.send_packets(self.pg0, self.pg1, count=0)
501 self.logger.info("FFP_TEST_FINISH_0006")
504 if __name__ == '__main__':
505 unittest.main(testRunner=VppTestRunner)