X-Git-Url: https://gerrit.fd.io/r/gitweb?a=blobdiff_plain;f=test%2Ftest_ip4.py;h=ab733ac1bb662b26b07b9c11daed1bb202128bc5;hb=22ab6f7cbb0f6139302aa6ca9f0c96dba17a37a7;hp=c18ce4caa0a9c700529c0d85a758d6ca2e365478;hpb=a6fe463c1cb67a52f06e519a514a47a9b6af8057;p=vpp.git diff --git a/test/test_ip4.py b/test/test_ip4.py index c18ce4caa0a..ab733ac1bb6 100644 --- a/test/test_ip4.py +++ b/test/test_ip4.py @@ -4,6 +4,7 @@ import random import socket import unittest +import scapy.compat from scapy.contrib.mpls import MPLS from scapy.layers.inet import IP, UDP, TCP, ICMP, icmptypes, icmpcodes from scapy.layers.l2 import Ether, Dot1Q, ARP @@ -14,7 +15,7 @@ from framework import VppTestCase, VppTestRunner from util import ppp from vpp_ip_route import VppIpRoute, VppRoutePath, VppIpMRoute, \ VppMRoutePath, MRouteItfFlags, MRouteEntryFlags, VppMplsIpBind, \ - VppMplsTable, VppIpTable, VppIpAddress + VppMplsTable, VppIpTable from vpp_sub_interface import VppSubInterface, VppDot1QSubint, VppDot1ADSubint @@ -168,7 +169,7 @@ class TestIPv4(VppTestCase): try: ip = packet[IP] udp = packet[UDP] - payload_info = self.payload_to_info(str(packet[Raw])) + payload_info = self.payload_to_info(packet[Raw]) packet_index = payload_info.index self.assertEqual(payload_info.dst, dst_sw_if_index) self.logger.debug( @@ -310,7 +311,7 @@ class TestIPv4FibCrud(VppTestCase): dest_addr_len = 32 n_next_hop_addr = socket.inet_pton(socket.AF_INET, next_hop_addr) for _ in range(count): - n_dest_addr = '{:08x}'.format(dest_addr).decode('hex') + n_dest_addr = binascii.unhexlify('{:08x}'.format(dest_addr)) self.vapi.ip_add_del_route(n_dest_addr, dest_addr_len, n_next_hop_addr) added_ips.append(socket.inet_ntoa(n_dest_addr)) @@ -325,7 +326,7 @@ class TestIPv4FibCrud(VppTestCase): dest_addr_len = 32 n_next_hop_addr = socket.inet_pton(socket.AF_INET, next_hop_addr) for _ in range(count): - n_dest_addr = '{:08x}'.format(dest_addr).decode('hex') + n_dest_addr = binascii.unhexlify('{:08x}'.format(dest_addr)) self.vapi.ip_add_del_route(n_dest_addr, dest_addr_len, n_next_hop_addr, is_add=0) removed_ips.append(socket.inet_ntoa(n_dest_addr)) @@ -351,8 +352,8 @@ class TestIPv4FibCrud(VppTestCase): def _find_ip_match(self, find_in, pkt): for p in find_in: - if self.payload_to_info(str(p[Raw])) == \ - self.payload_to_info(str(pkt[Raw])): + if self.payload_to_info(p[Raw]) == \ + self.payload_to_info(pkt[Raw]): if p[IP].src != pkt[IP].src: break if p[IP].dst != pkt[IP].dst: @@ -395,7 +396,7 @@ class TestIPv4FibCrud(VppTestCase): for ip in ips: self.assertTrue(_ip_in_route_dump(ip, fib_dump), - 'IP {} is not in fib dump.'.format(ip)) + 'IP {!s} is not in fib dump.'.format(ip)) def verify_not_in_route_dump(self, fib_dump, ips): @@ -406,7 +407,7 @@ class TestIPv4FibCrud(VppTestCase): for ip in ips: self.assertFalse(_ip_in_route_dump(ip, fib_dump), - 'IP {} is in fib dump.'.format(ip)) + 'IP {!s} is in fib dump.'.format(ip)) @classmethod def setUpClass(cls): @@ -441,6 +442,9 @@ class TestIPv4FibCrud(VppTestCase): super(TestIPv4FibCrud, self).setUp() self.reset_packet_infos() + self.configured_routes = [] + self.deleted_routes = [] + def test_1_add_routes(self): """ Add 1k routes @@ -471,6 +475,9 @@ class TestIPv4FibCrud(VppTestCase): - delete 10 routes check with traffic script. """ + # config 1M FIB entries + self.configured_routes.extend(self.config_fib_many_to_one( + "10.0.0.0", self.pg0.remote_ip4, 100)) self.deleted_routes.extend(self.unconfig_fib_many_to_one( "10.0.0.10", self.pg0.remote_ip4, 10)) for x in self.deleted_routes: @@ -501,6 +508,14 @@ class TestIPv4FibCrud(VppTestCase): - re-add 5 routes check with traffic script. - add 100 routes check with traffic script. """ + # config 1M FIB entries + self.configured_routes.extend(self.config_fib_many_to_one( + "10.0.0.0", self.pg0.remote_ip4, 100)) + self.deleted_routes.extend(self.unconfig_fib_many_to_one( + "10.0.0.10", self.pg0.remote_ip4, 10)) + for x in self.deleted_routes: + self.configured_routes.remove(x) + tmp = self.config_fib_many_to_one( "10.0.0.10", self.pg0.remote_ip4, 5) self.configured_routes.extend(tmp) @@ -790,9 +805,9 @@ class TestIPSubNets(VppTestCase): # ip_addr_n = socket.inet_pton(socket.AF_INET, "10.10.10.10") - self.vapi.sw_interface_add_del_address(self.pg0.sw_if_index, - ip_addr_n, - 16) + self.vapi.sw_interface_add_del_address( + sw_if_index=self.pg0.sw_if_index, address=ip_addr_n, + address_length=16) pn = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) / @@ -809,10 +824,9 @@ class TestIPSubNets(VppTestCase): self.send_and_assert_no_replies(self.pg1, pb, "IP Broadcast address") # remove the sub-net and we are forwarding via the cover again - self.vapi.sw_interface_add_del_address(self.pg0.sw_if_index, - ip_addr_n, - 16, - is_add=0) + self.vapi.sw_interface_add_del_address( + sw_if_index=self.pg0.sw_if_index, address=ip_addr_n, + address_length=16, is_add=0) self.pg1.add_stream(pn) self.pg_enable_capture(self.pg_interfaces) self.pg_start() @@ -828,9 +842,9 @@ class TestIPSubNets(VppTestCase): # ip_addr_n = socket.inet_pton(socket.AF_INET, "10.10.10.10") - self.vapi.sw_interface_add_del_address(self.pg0.sw_if_index, - ip_addr_n, - 31) + self.vapi.sw_interface_add_del_address( + sw_if_index=self.pg0.sw_if_index, address=ip_addr_n, + address_length=31) pn = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) / @@ -845,10 +859,9 @@ class TestIPSubNets(VppTestCase): rx[ARP] # remove the sub-net and we are forwarding via the cover again - self.vapi.sw_interface_add_del_address(self.pg0.sw_if_index, - ip_addr_n, - 31, - is_add=0) + self.vapi.sw_interface_add_del_address( + sw_if_index=self.pg0.sw_if_index, address=ip_addr_n, + address_length=31, is_add=0) self.pg1.add_stream(pn) self.pg_enable_capture(self.pg_interfaces) self.pg_start() @@ -1123,7 +1136,7 @@ class TestIPPunt(VppTestCase): # # Configure a punt redirect via pg1. # - nh_addr = VppIpAddress(self.pg1.remote_ip4).encode() + nh_addr = self.pg1.remote_ip4 self.vapi.ip_punt_redirect(self.pg0.sw_if_index, self.pg1.sw_if_index, nh_addr) @@ -1133,7 +1146,7 @@ class TestIPPunt(VppTestCase): # # add a policer # - policer = self.vapi.policer_add_del("ip4-punt", 400, 0, 10, 0, + policer = self.vapi.policer_add_del(b"ip4-punt", 400, 0, 10, 0, rate_type=1) self.vapi.ip_punt_police(policer.policer_index) @@ -1154,7 +1167,7 @@ class TestIPPunt(VppTestCase): # remove the poilcer. back to full rx # self.vapi.ip_punt_police(policer.policer_index, is_add=0) - self.vapi.policer_add_del("ip4-punt", 400, 0, 10, 0, + self.vapi.policer_add_del(b"ip4-punt", 400, 0, 10, 0, rate_type=1, is_add=0) self.send_and_expect(self.pg0, pkts, self.pg1) @@ -1187,7 +1200,7 @@ class TestIPPunt(VppTestCase): # # Configure a punt redirects # - nh_address = VppIpAddress(self.pg3.remote_ip4).encode() + nh_address = self.pg3.remote_ip4 self.vapi.ip_punt_redirect(self.pg0.sw_if_index, self.pg3.sw_if_index, nh_address) @@ -1196,7 +1209,7 @@ class TestIPPunt(VppTestCase): nh_address) self.vapi.ip_punt_redirect(self.pg2.sw_if_index, self.pg3.sw_if_index, - VppIpAddress('0.0.0.0').encode()) + '0.0.0.0') # # Dump pg0 punt redirects @@ -1212,8 +1225,8 @@ class TestIPPunt(VppTestCase): self.assertEqual(len(punts), 3) for p in punts: self.assertEqual(p.punt.tx_sw_if_index, self.pg3.sw_if_index) - self.assertNotEqual(punts[1].punt.nh.un.ip4, self.pg3.remote_ip4) - self.assertEqual(punts[2].punt.nh.un.ip4.address, '\x00'*4) + self.assertNotEqual(punts[1].punt.nh, self.pg3.remote_ip4) + self.assertEqual(str(punts[2].punt.nh), '0.0.0.0') class TestIPDeag(VppTestCase): @@ -1470,6 +1483,25 @@ class TestIPInput(VppTestCase): # Reset MTU for subsequent tests self.vapi.sw_interface_set_mtu(self.pg1.sw_if_index, [9000, 0, 0, 0]) + # + # source address 0.0.0.0 and 25.255.255.255 and for-us + # + p_s0 = (Ether(src=self.pg0.remote_mac, + dst=self.pg0.local_mac) / + IP(src="0.0.0.0", + dst=self.pg0.local_ip4) / + ICMP(id=4, seq=4) / + Raw(load='\x0a' * 18)) + rx = self.send_and_assert_no_replies(self.pg0, p_s0 * 17) + + p_s0 = (Ether(src=self.pg0.remote_mac, + dst=self.pg0.local_mac) / + IP(src="255.255.255.255", + dst=self.pg0.local_ip4) / + ICMP(id=4, seq=4) / + Raw(load='\x0a' * 18)) + rx = self.send_and_assert_no_replies(self.pg0, p_s0 * 17) + class TestIPDirectedBroadcast(VppTestCase): """ IPv4 Directed Broadcast """