6 from framework import VppTestCase, VppTestRunner
7 from vpp_sub_interface import VppSubInterface, VppDot1QSubint, VppDot1ADSubint
8 from vpp_ip_route import VppIpRoute, VppRoutePath
10 from scapy.packet import Raw
11 from scapy.layers.l2 import Ether, Dot1Q
12 from scapy.layers.inet import IP, UDP, ICMP, icmptypes, icmpcodes
16 class TestIPv4(VppTestCase):
17 """ IPv4 Test Case """
21 Perform test setup before test case.
24 - create 3 pg interfaces
25 - untagged pg0 interface
26 - Dot1Q subinterface on pg1
27 - Dot1AD subinterface on pg2
29 - put it into UP state
31 - resolve neighbor address using ARP
32 - configure 200 fib entries
34 :ivar list interfaces: pg interfaces and subinterfaces.
35 :ivar dict flows: IPv4 packet flows in test.
36 :ivar list pg_if_packet_sizes: packet sizes in test.
38 super(TestIPv4, self).setUp()
40 # create 3 pg interfaces
41 self.create_pg_interfaces(range(3))
43 # create 2 subinterfaces for pg1 and pg2
44 self.sub_interfaces = [
45 VppDot1QSubint(self, self.pg1, 100),
46 VppDot1ADSubint(self, self.pg2, 200, 300, 400)]
48 # packet flows mapping pg0 -> pg1.sub, pg2.sub, etc.
50 self.flows[self.pg0] = [self.pg1.sub_if, self.pg2.sub_if]
51 self.flows[self.pg1.sub_if] = [self.pg0, self.pg2.sub_if]
52 self.flows[self.pg2.sub_if] = [self.pg0, self.pg1.sub_if]
55 self.pg_if_packet_sizes = [64, 512, 1518, 9018]
56 self.sub_if_packet_sizes = [64, 512, 1518 + 4, 9018 + 4]
58 self.interfaces = list(self.pg_interfaces)
59 self.interfaces.extend(self.sub_interfaces)
61 # setup all interfaces
62 for i in self.interfaces:
67 # config 2M FIB entries
68 self.config_fib_entries(200)
71 """Run standard test teardown and log ``show ip arp``."""
72 super(TestIPv4, self).tearDown()
74 self.logger.info(self.vapi.cli("show ip arp"))
75 # info(self.vapi.cli("show ip fib")) # many entries
77 def config_fib_entries(self, count):
78 """For each interface add to the FIB table *count* routes to
79 "10.0.0.1/32" destination with interface's local address as next-hop
82 :param int count: Number of FIB entries.
84 - *TODO:* check if the next-hop address shouldn't be remote address
85 instead of local address.
87 n_int = len(self.interfaces)
90 dest_addr = socket.inet_pton(socket.AF_INET, "10.0.0.1")
92 for i in self.interfaces:
93 next_hop_address = i.local_ip4n
94 for j in range(count / n_int):
95 self.vapi.ip_add_del_route(
96 dest_addr, dest_addr_len, next_hop_address)
98 if counter / count * 100 > percent:
99 self.logger.info("Configure %d FIB entries .. %d%% done" %
103 def create_stream(self, src_if, packet_sizes):
104 """Create input packet stream for defined interface.
106 :param VppInterface src_if: Interface to create packet stream for.
107 :param list packet_sizes: Required packet sizes.
110 for i in range(0, 257):
111 dst_if = self.flows[src_if][i % 2]
112 info = self.create_packet_info(src_if, dst_if)
113 payload = self.info_to_payload(info)
114 p = (Ether(dst=src_if.local_mac, src=src_if.remote_mac) /
115 IP(src=src_if.remote_ip4, dst=dst_if.remote_ip4) /
116 UDP(sport=1234, dport=1234) /
119 if isinstance(src_if, VppSubInterface):
120 p = src_if.add_dot1_layer(p)
121 size = packet_sizes[(i // 2) % len(packet_sizes)]
122 self.extend_packet(p, size)
126 def verify_capture(self, dst_if, capture):
127 """Verify captured input packet stream for defined interface.
129 :param VppInterface dst_if: Interface to verify captured packet stream
131 :param list capture: Captured packet stream.
133 self.logger.info("Verifying capture on interface %s" % dst_if.name)
135 for i in self.interfaces:
136 last_info[i.sw_if_index] = None
138 dst_sw_if_index = dst_if.sw_if_index
139 if hasattr(dst_if, 'parent'):
141 for packet in capture:
143 # Check VLAN tags and Ethernet header
144 packet = dst_if.remove_dot1_layer(packet)
145 self.assertTrue(Dot1Q not in packet)
149 payload_info = self.payload_to_info(str(packet[Raw]))
150 packet_index = payload_info.index
151 self.assertEqual(payload_info.dst, dst_sw_if_index)
153 "Got packet on port %s: src=%u (id=%u)" %
154 (dst_if.name, payload_info.src, packet_index))
155 next_info = self.get_next_packet_info_for_interface2(
156 payload_info.src, dst_sw_if_index,
157 last_info[payload_info.src])
158 last_info[payload_info.src] = next_info
159 self.assertTrue(next_info is not None)
160 self.assertEqual(packet_index, next_info.index)
161 saved_packet = next_info.data
162 # Check standard fields
163 self.assertEqual(ip.src, saved_packet[IP].src)
164 self.assertEqual(ip.dst, saved_packet[IP].dst)
165 self.assertEqual(udp.sport, saved_packet[UDP].sport)
166 self.assertEqual(udp.dport, saved_packet[UDP].dport)
168 self.logger.error(ppp("Unexpected or invalid packet:", packet))
170 for i in self.interfaces:
171 remaining_packet = self.get_next_packet_info_for_interface2(
172 i.sw_if_index, dst_sw_if_index, last_info[i.sw_if_index])
173 self.assertTrue(remaining_packet is None,
174 "Interface %s: Packet expected from interface %s "
175 "didn't arrive" % (dst_if.name, i.name))
182 - Create IPv4 stream for pg0 interface
183 - Create IPv4 tagged streams for pg1's and pg2's subinterface.
184 - Send and verify received packets on each interface.
187 pkts = self.create_stream(self.pg0, self.pg_if_packet_sizes)
188 self.pg0.add_stream(pkts)
190 for i in self.sub_interfaces:
191 pkts = self.create_stream(i, self.sub_if_packet_sizes)
192 i.parent.add_stream(pkts)
194 self.pg_enable_capture(self.pg_interfaces)
197 pkts = self.pg0.get_capture()
198 self.verify_capture(self.pg0, pkts)
200 for i in self.sub_interfaces:
201 pkts = i.parent.get_capture()
202 self.verify_capture(i, pkts)
205 class TestIPv4FibCrud(VppTestCase):
206 """ FIB - add/update/delete - ip4 routes
214 ..note:: Python API is too slow to add many routes, needs replacement.
217 def config_fib_many_to_one(self, start_dest_addr, next_hop_addr, count):
220 :param start_dest_addr:
221 :param next_hop_addr:
223 :return list: added ips with 32 prefix
226 dest_addr = int(socket.inet_pton(socket.AF_INET,
227 start_dest_addr).encode('hex'),
230 n_next_hop_addr = socket.inet_pton(socket.AF_INET, next_hop_addr)
231 for _ in range(count):
232 n_dest_addr = '{:08x}'.format(dest_addr).decode('hex')
233 self.vapi.ip_add_del_route(n_dest_addr, dest_addr_len,
235 added_ips.append(socket.inet_ntoa(n_dest_addr))
239 def unconfig_fib_many_to_one(self, start_dest_addr, next_hop_addr, count):
242 dest_addr = int(socket.inet_pton(socket.AF_INET,
243 start_dest_addr).encode('hex'),
246 n_next_hop_addr = socket.inet_pton(socket.AF_INET, next_hop_addr)
247 for _ in range(count):
248 n_dest_addr = '{:08x}'.format(dest_addr).decode('hex')
249 self.vapi.ip_add_del_route(n_dest_addr, dest_addr_len,
250 n_next_hop_addr, is_add=0)
251 removed_ips.append(socket.inet_ntoa(n_dest_addr))
255 def create_stream(self, src_if, dst_if, dst_ips, count):
258 for _ in range(count):
259 dst_addr = random.choice(dst_ips)
260 info = self.create_packet_info(src_if, dst_if)
261 payload = self.info_to_payload(info)
262 p = (Ether(dst=src_if.local_mac, src=src_if.remote_mac) /
263 IP(src=src_if.remote_ip4, dst=dst_addr) /
264 UDP(sport=1234, dport=1234) /
267 self.extend_packet(p, random.choice(self.pg_if_packet_sizes))
272 def _find_ip_match(self, find_in, pkt):
274 if self.payload_to_info(str(p[Raw])) == \
275 self.payload_to_info(str(pkt[Raw])):
276 if p[IP].src != pkt[IP].src:
278 if p[IP].dst != pkt[IP].dst:
280 if p[UDP].sport != pkt[UDP].sport:
282 if p[UDP].dport != pkt[UDP].dport:
288 def _match_route_detail(route_detail, ip, address_length=32, table_id=0):
289 if route_detail.address == socket.inet_pton(socket.AF_INET, ip):
290 if route_detail.table_id != table_id:
292 elif route_detail.address_length != address_length:
299 def verify_capture(self, dst_interface, received_pkts, expected_pkts):
300 self.assertEqual(len(received_pkts), len(expected_pkts))
301 to_verify = list(expected_pkts)
302 for p in received_pkts:
303 self.assertEqual(p.src, dst_interface.local_mac)
304 self.assertEqual(p.dst, dst_interface.remote_mac)
305 x = self._find_ip_match(to_verify, p)
307 self.assertListEqual(to_verify, [])
309 def verify_route_dump(self, fib_dump, ips):
311 def _ip_in_route_dump(ip, fib_dump):
312 return next((route for route in fib_dump
313 if self._match_route_detail(route, ip)),
317 self.assertTrue(_ip_in_route_dump(ip, fib_dump),
318 'IP {} is not in fib dump.'.format(ip))
320 def verify_not_in_route_dump(self, fib_dump, ips):
322 def _ip_in_route_dump(ip, fib_dump):
323 return next((route for route in fib_dump
324 if self._match_route_detail(route, ip)),
328 self.assertFalse(_ip_in_route_dump(ip, fib_dump),
329 'IP {} is in fib dump.'.format(ip))
334 #. Create and initialize 3 pg interfaces.
335 #. initialize class attributes configured_routes and deleted_routes
336 to store information between tests.
338 super(TestIPv4FibCrud, cls).setUpClass()
341 # create 3 pg interfaces
342 cls.create_pg_interfaces(range(3))
344 cls.interfaces = list(cls.pg_interfaces)
346 # setup all interfaces
347 for i in cls.interfaces:
352 cls.configured_routes = []
353 cls.deleted_routes = []
354 cls.pg_if_packet_sizes = [64, 512, 1518, 9018]
357 super(TestIPv4FibCrud, cls).tearDownClass()
361 super(TestIPv4FibCrud, self).setUp()
362 self.reset_packet_infos()
364 def test_1_add_routes(self):
367 - add 100 routes check with traffic script.
369 # config 1M FIB entries
370 self.configured_routes.extend(self.config_fib_many_to_one(
371 "10.0.0.0", self.pg0.remote_ip4, 100))
373 fib_dump = self.vapi.ip_fib_dump()
374 self.verify_route_dump(fib_dump, self.configured_routes)
376 self.stream_1 = self.create_stream(
377 self.pg1, self.pg0, self.configured_routes, 100)
378 self.stream_2 = self.create_stream(
379 self.pg2, self.pg0, self.configured_routes, 100)
380 self.pg1.add_stream(self.stream_1)
381 self.pg2.add_stream(self.stream_2)
383 self.pg_enable_capture(self.pg_interfaces)
386 pkts = self.pg0.get_capture(len(self.stream_1) + len(self.stream_2))
387 self.verify_capture(self.pg0, pkts, self.stream_1 + self.stream_2)
389 def test_2_del_routes(self):
390 """ Delete 100 routes
392 - delete 10 routes check with traffic script.
394 self.deleted_routes.extend(self.unconfig_fib_many_to_one(
395 "10.0.0.10", self.pg0.remote_ip4, 10))
396 for x in self.deleted_routes:
397 self.configured_routes.remove(x)
399 fib_dump = self.vapi.ip_fib_dump()
400 self.verify_route_dump(fib_dump, self.configured_routes)
402 self.stream_1 = self.create_stream(
403 self.pg1, self.pg0, self.configured_routes, 100)
404 self.stream_2 = self.create_stream(
405 self.pg2, self.pg0, self.configured_routes, 100)
406 self.stream_3 = self.create_stream(
407 self.pg1, self.pg0, self.deleted_routes, 100)
408 self.stream_4 = self.create_stream(
409 self.pg2, self.pg0, self.deleted_routes, 100)
410 self.pg1.add_stream(self.stream_1 + self.stream_3)
411 self.pg2.add_stream(self.stream_2 + self.stream_4)
412 self.pg_enable_capture(self.pg_interfaces)
415 pkts = self.pg0.get_capture(len(self.stream_1) + len(self.stream_2))
416 self.verify_capture(self.pg0, pkts, self.stream_1 + self.stream_2)
418 def test_3_add_new_routes(self):
421 - re-add 5 routes check with traffic script.
422 - add 100 routes check with traffic script.
424 tmp = self.config_fib_many_to_one(
425 "10.0.0.10", self.pg0.remote_ip4, 5)
426 self.configured_routes.extend(tmp)
428 self.deleted_routes.remove(x)
430 self.configured_routes.extend(self.config_fib_many_to_one(
431 "10.0.1.0", self.pg0.remote_ip4, 100))
433 fib_dump = self.vapi.ip_fib_dump()
434 self.verify_route_dump(fib_dump, self.configured_routes)
436 self.stream_1 = self.create_stream(
437 self.pg1, self.pg0, self.configured_routes, 300)
438 self.stream_2 = self.create_stream(
439 self.pg2, self.pg0, self.configured_routes, 300)
440 self.stream_3 = self.create_stream(
441 self.pg1, self.pg0, self.deleted_routes, 100)
442 self.stream_4 = self.create_stream(
443 self.pg2, self.pg0, self.deleted_routes, 100)
445 self.pg1.add_stream(self.stream_1 + self.stream_3)
446 self.pg2.add_stream(self.stream_2 + self.stream_4)
447 self.pg_enable_capture(self.pg_interfaces)
450 pkts = self.pg0.get_capture(len(self.stream_1) + len(self.stream_2))
451 self.verify_capture(self.pg0, pkts, self.stream_1 + self.stream_2)
453 def test_4_del_routes(self):
454 """ Delete 1.5k routes
456 - delete 5 routes check with traffic script.
457 - add 100 routes check with traffic script.
459 self.deleted_routes.extend(self.unconfig_fib_many_to_one(
460 "10.0.0.0", self.pg0.remote_ip4, 15))
461 self.deleted_routes.extend(self.unconfig_fib_many_to_one(
462 "10.0.0.20", self.pg0.remote_ip4, 85))
463 self.deleted_routes.extend(self.unconfig_fib_many_to_one(
464 "10.0.1.0", self.pg0.remote_ip4, 100))
465 fib_dump = self.vapi.ip_fib_dump()
466 self.verify_not_in_route_dump(fib_dump, self.deleted_routes)
469 class TestIPNull(VppTestCase):
470 """ IPv4 routes via NULL """
473 super(TestIPNull, self).setUp()
475 # create 2 pg interfaces
476 self.create_pg_interfaces(range(1))
478 for i in self.pg_interfaces:
484 super(TestIPNull, self).tearDown()
485 for i in self.pg_interfaces:
489 def test_ip_null(self):
490 """ IP NULL route """
493 # A route via IP NULL that will reply with ICMP unreachables
495 ip_unreach = VppIpRoute(self, "10.0.0.1", 32, [], is_unreach=1)
496 ip_unreach.add_vpp_config()
498 p_unreach = (Ether(src=self.pg0.remote_mac,
499 dst=self.pg0.local_mac) /
500 IP(src=self.pg0.remote_ip4, dst="10.0.0.1") /
501 UDP(sport=1234, dport=1234) /
504 self.pg0.add_stream(p_unreach)
505 self.pg_enable_capture(self.pg_interfaces)
508 rx = self.pg0.get_capture(1)
512 self.assertEqual(icmptypes[icmp.type], "dest-unreach")
513 self.assertEqual(icmpcodes[icmp.type][icmp.code], "host-unreachable")
514 self.assertEqual(icmp.src, self.pg0.remote_ip4)
515 self.assertEqual(icmp.dst, "10.0.0.1")
518 # ICMP replies are rate limited. so sit and spin.
523 # A route via IP NULL that will reply with ICMP prohibited
525 ip_prohibit = VppIpRoute(self, "10.0.0.2", 32, [], is_prohibit=1)
526 ip_prohibit.add_vpp_config()
528 p_prohibit = (Ether(src=self.pg0.remote_mac,
529 dst=self.pg0.local_mac) /
530 IP(src=self.pg0.remote_ip4, dst="10.0.0.2") /
531 UDP(sport=1234, dport=1234) /
534 self.pg0.add_stream(p_prohibit)
535 self.pg_enable_capture(self.pg_interfaces)
538 rx = self.pg0.get_capture(1)
543 self.assertEqual(icmptypes[icmp.type], "dest-unreach")
544 self.assertEqual(icmpcodes[icmp.type][icmp.code], "host-prohibited")
545 self.assertEqual(icmp.src, self.pg0.remote_ip4)
546 self.assertEqual(icmp.dst, "10.0.0.2")
549 if __name__ == '__main__':
550 unittest.main(testRunner=VppTestRunner)