X-Git-Url: https://gerrit.fd.io/r/gitweb?a=blobdiff_plain;f=test%2Ftest_ip6.py;h=35061b0b53c5085dff398ef18b02cc262e45e39c;hb=097fa66b9;hp=f75fb4ee431b6ad49cf1932c905b02940de81ecd;hpb=e1ade684eb5a7e5743abfda47488165452cc5b64;p=vpp.git diff --git a/test/test_ip6.py b/test/test_ip6.py index f75fb4ee431..35061b0b53c 100644 --- a/test/test_ip6.py +++ b/test/test_ip6.py @@ -4,12 +4,13 @@ import socket import unittest from parameterized import parameterized +import scapy.compat import scapy.layers.inet6 as inet6 from scapy.contrib.mpls import MPLS from scapy.layers.inet6 import IPv6, ICMPv6ND_NS, ICMPv6ND_RS, \ ICMPv6ND_RA, ICMPv6NDOptMTU, ICMPv6NDOptSrcLLAddr, ICMPv6NDOptPrefixInfo, \ ICMPv6ND_NA, ICMPv6NDOptDstLLAddr, ICMPv6DestUnreach, icmp6types, \ - ICMPv6TimeExceeded, ICMPv6EchoRequest, ICMPv6EchoReply + ICMPv6TimeExceeded, ICMPv6EchoRequest, ICMPv6EchoReply, IPv6ExtHdrHopByHop from scapy.layers.l2 import Ether, Dot1Q from scapy.packet import Raw from scapy.utils import inet_pton, inet_ntop @@ -22,11 +23,11 @@ from util import ppp, ip6_normalize, mk_ll_addr from vpp_ip import DpoProto from vpp_ip_route import VppIpRoute, VppRoutePath, find_route, VppIpMRoute, \ VppMRoutePath, MRouteItfFlags, MRouteEntryFlags, VppMplsIpBind, \ - VppMplsRoute, VppMplsTable, VppIpTable + VppMplsRoute, VppMplsTable, VppIpTable, FibPathType from vpp_neighbor import find_nbr, VppNeighbor from vpp_pg_interface import is_ipv6_misc from vpp_sub_interface import VppSubInterface, VppDot1QSubint -from ipaddress import IPv6Network, IPv4Network +from ipaddress import IPv6Network, IPv4Network, IPv6Address AF_INET6 = socket.AF_INET6 @@ -35,6 +36,8 @@ try: except NameError: text_type = str +NUM_PKTS = 67 + class TestIPv6ND(VppTestCase): def validate_ra(self, intf, rx, dst_ip=None): @@ -160,6 +163,10 @@ class TestIPv6(TestIPv6ND): def setUpClass(cls): super(TestIPv6, cls).setUpClass() + @classmethod + def tearDownClass(cls): + super(TestIPv6, cls).tearDownClass() + def setUp(self): """ Perform test setup before test case. @@ -210,9 +217,6 @@ class TestIPv6(TestIPv6ND): i.config_ip6() i.resolve_ndp() - # config 2M FIB entries - self.config_fib_entries(200) - def tearDown(self): """Run standard test teardown and log ``show ip6 neighbors``.""" for i in self.interfaces: @@ -227,32 +231,6 @@ class TestIPv6(TestIPv6ND): self.logger.info(self.vapi.cli("show ip6 neighbors")) # info(self.vapi.cli("show ip6 fib")) # many entries - def config_fib_entries(self, count): - """For each interface add to the FIB table *count* routes to - "fd02::1/128" destination with interface's local address as next-hop - address. - - :param int count: Number of FIB entries. - - - *TODO:* check if the next-hop address shouldn't be remote address - instead of local address. - """ - n_int = len(self.interfaces) - percent = 0 - counter = 0.0 - dest_addr = inet_pton(AF_INET6, "fd02::1") - dest_addr_len = 128 - for i in self.interfaces: - next_hop_address = i.local_ip6n - for j in range(count / n_int): - self.vapi.ip_add_del_route( - dest_addr, dest_addr_len, next_hop_address, is_ipv6=1) - counter += 1 - if counter / count * 100 > percent: - self.logger.info("Configure %d FIB entries .. %d%% done" % - (count, percent)) - percent += 1 - def modify_packet(self, src_if, packet_size, pkt): """Add load, set destination IP and extend packet to required packet size for defined interface. @@ -318,7 +296,7 @@ class TestIPv6(TestIPv6ND): try: ip = packet[IPv6] udp = packet[inet6.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( @@ -350,6 +328,24 @@ class TestIPv6(TestIPv6ND): "Interface %s: Packet expected from interface %s " "didn't arrive" % (dst_if.name, i.name)) + def test_next_header_anomaly(self): + """ IPv6 next header anomaly test + + Test scenario: + - ipv6 next header field = Fragment Header (44) + - next header is ICMPv6 Echo Request + - wait for reassembly + """ + pkt = (Ether(src=self.pg0.local_mac, dst=self.pg0.remote_mac) / + IPv6(src=self.pg0.remote_ip6, dst=self.pg0.local_ip6, nh=44) / + ICMPv6EchoRequest()) + + self.pg0.add_stream(pkt) + self.pg_start() + + # wait for reassembly + self.sleep(10) + def test_fib(self): """ IPv6 FIB test @@ -456,8 +452,7 @@ class TestIPv6(TestIPv6ND): self.pg0._remote_hosts[2].ip6)) self.assertFalse(find_route(self, self.pg0._remote_hosts[2].ip6, - 128, - inet=AF_INET6)) + 128)) # # send an NS from a link local address to the interface's global @@ -484,8 +479,7 @@ class TestIPv6(TestIPv6ND): self.pg0._remote_hosts[2].ip6_ll)) self.assertFalse(find_route(self, self.pg0._remote_hosts[2].ip6_ll, - 128, - inet=AF_INET6)) + 128)) # # An NS to the router's own Link-local @@ -511,8 +505,7 @@ class TestIPv6(TestIPv6ND): self.pg0._remote_hosts[3].ip6_ll)) self.assertFalse(find_route(self, self.pg0._remote_hosts[3].ip6_ll, - 128, - inet=AF_INET6)) + 128)) def test_ns_duplicates(self): """ ND Duplicates""" @@ -560,7 +553,7 @@ class TestIPv6(TestIPv6ND): # # remove the duplicate on pg1 - # packet stream shoud generate NSs out of pg1 + # packet stream should generate NSs out of pg1 # ns_pg1.remove_vpp_config() @@ -623,7 +616,7 @@ class TestIPv6(TestIPv6ND): # decipher how to decode. this 1st layer of option always returns # nested classes, so a direct obj1=obj2 comparison always fails. # however, the getlayer(.., 2) does give one instnace. - # so we cheat here and construct a new opt instnace for comparison + # so we cheat here and construct a new opt instance for comparison rd = ICMPv6NDOptPrefixInfo( prefixlen=raos.prefixlen, prefix=raos.prefix, @@ -682,7 +675,8 @@ class TestIPv6(TestIPv6ND): self.send_and_assert_no_replies(self.pg0, pkts, "RA rate limited") # - # When we reconfiure the IPv6 RA config, we reset the RA rate limiting, + # When we reconfigure the IPv6 RA config, + # we reset the RA rate limiting, # so we need to do this before each test below so as not to drop # packets for rate limiting reasons. Test this works here. # @@ -853,7 +847,7 @@ class TestIPv6(TestIPv6ND): opt=opt) # - # Use the reset to defults option to revert to defaults + # Use the reset to defaults option to revert to defaults # L and A flag are clear in the advert # self.pg0.ip6_ra_prefix(self.pg0.local_ip6, @@ -902,7 +896,7 @@ class TestIPv6(TestIPv6ND): opt=opt) # - # Remove the first refix-info - expect the second is still in the + # Remove the first prefix-info - expect the second is still in the # advert # self.pg0.ip6_ra_prefix(self.pg0.local_ip6, @@ -922,7 +916,7 @@ class TestIPv6(TestIPv6ND): opt=opt) # - # Remove the second prefix-info - expect no prefix-info i nthe adverts + # Remove the second prefix-info - expect no prefix-info in the adverts # self.pg0.ip6_ra_prefix(self.pg1.local_ip6, self.pg1.local_ip6_prefix_len, @@ -942,6 +936,14 @@ class TestIPv6(TestIPv6ND): class TestICMPv6Echo(VppTestCase): """ ICMPv6 Echo Test Case """ + @classmethod + def setUpClass(cls): + super(TestICMPv6Echo, cls).setUpClass() + + @classmethod + def tearDownClass(cls): + super(TestICMPv6Echo, cls).tearDownClass() + def setUp(self): super(TestICMPv6Echo, self).setUp() @@ -971,7 +973,7 @@ class TestICMPv6Echo(VppTestCase): icmpv6_id = 0xb icmpv6_seq = 5 - icmpv6_data = '\x0a' * 18 + icmpv6_data = b'\x0a' * 18 p_echo_request = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) / IPv6(src=self.pg0.remote_ip6, @@ -1011,6 +1013,10 @@ class TestIPv6RD(TestIPv6ND): def setUpClass(cls): super(TestIPv6RD, cls).setUpClass() + @classmethod + def tearDownClass(cls): + super(TestIPv6RD, cls).tearDownClass() + def setUp(self): super(TestIPv6RD, self).setUp() @@ -1128,6 +1134,10 @@ class TestIPv6RDControlPlane(TestIPv6ND): def setUpClass(cls): super(TestIPv6RDControlPlane, cls).setUpClass() + @classmethod + def tearDownClass(cls): + super(TestIPv6RDControlPlane, cls).tearDownClass() + def setUp(self): super(TestIPv6RDControlPlane, self).setUp() @@ -1160,12 +1170,12 @@ class TestIPv6RDControlPlane(TestIPv6ND): def get_default_routes(fib): list = [] for entry in fib: - if entry.address_length == 0: - for path in entry.path: + if entry.route.prefix.prefixlen == 0: + for path in entry.route.paths: if path.sw_if_index != 0xFFFFFFFF: defaut_route = {} defaut_route['sw_if_index'] = path.sw_if_index - defaut_route['next_hop'] = path.next_hop + defaut_route['next_hop'] = path.nh.address.ip6 list.append(defaut_route) return list @@ -1173,20 +1183,20 @@ class TestIPv6RDControlPlane(TestIPv6ND): def get_interface_addresses(fib, pg): list = [] for entry in fib: - if entry.address_length == 128: - path = entry.path[0] + if entry.route.prefix.prefixlen == 128: + path = entry.route.paths[0] if path.sw_if_index == pg.sw_if_index: - list.append(entry.address) + list.append(str(entry.route.prefix.network_address)) return list def test_all(self): """ Test handling of SLAAC addresses and default routes """ - fib = self.vapi.ip6_fib_dump() + fib = self.vapi.ip_route_dump(0, True) default_routes = self.get_default_routes(fib) initial_addresses = set(self.get_interface_addresses(fib, self.pg0)) self.assertEqual(default_routes, []) - router_address = self.pg0.remote_ip6n_ll + router_address = IPv6Address(text_type(self.pg0.remote_ip6_ll)) self.vapi.ip6_nd_address_autoconfig(self.pg0.sw_if_index, 1, 1) @@ -1214,14 +1224,15 @@ class TestIPv6RDControlPlane(TestIPv6ND): self.sleep(0.1) - fib = self.vapi.ip6_fib_dump() + fib = self.vapi.ip_route_dump(0, True) # check FIB for new address addresses = set(self.get_interface_addresses(fib, self.pg0)) new_addresses = addresses.difference(initial_addresses) self.assertEqual(len(new_addresses), 1) - prefix = list(new_addresses)[0][:8] + '\0\0\0\0\0\0\0\0' - self.assertEqual(inet_ntop(AF_INET6, prefix), '1::') + prefix = IPv6Network(text_type("%s/%d" % (list(new_addresses)[0], 20)), + strict=False) + self.assertEqual(prefix, IPv6Network(text_type('1::/20'))) # check FIB for new default route default_routes = self.get_default_routes(fib) @@ -1238,7 +1249,7 @@ class TestIPv6RDControlPlane(TestIPv6ND): self.sleep(0.1) # check that default route is deleted - fib = self.vapi.ip6_fib_dump() + fib = self.vapi.ip_route_dump(0, True) default_routes = self.get_default_routes(fib) self.assertEqual(len(default_routes), 0) @@ -1252,7 +1263,7 @@ class TestIPv6RDControlPlane(TestIPv6ND): self.sleep(0.1) # check FIB for new default route - fib = self.vapi.ip6_fib_dump() + fib = self.vapi.ip_route_dump(0, True) default_routes = self.get_default_routes(fib) self.assertEqual(len(default_routes), 1) dr = default_routes[0] @@ -1267,7 +1278,7 @@ class TestIPv6RDControlPlane(TestIPv6ND): self.sleep(0.1) # check that default route still exists - fib = self.vapi.ip6_fib_dump() + fib = self.vapi.ip_route_dump(0, True) default_routes = self.get_default_routes(fib) self.assertEqual(len(default_routes), 1) dr = default_routes[0] @@ -1277,21 +1288,23 @@ class TestIPv6RDControlPlane(TestIPv6ND): self.sleep(1) # check that default route is deleted - fib = self.vapi.ip6_fib_dump() + fib = self.vapi.ip_route_dump(0, True) default_routes = self.get_default_routes(fib) self.assertEqual(len(default_routes), 0) # check FIB still contains the SLAAC address addresses = set(self.get_interface_addresses(fib, self.pg0)) new_addresses = addresses.difference(initial_addresses) + self.assertEqual(len(new_addresses), 1) - prefix = list(new_addresses)[0][:8] + '\0\0\0\0\0\0\0\0' - self.assertEqual(inet_ntop(AF_INET6, prefix), '1::') + prefix = IPv6Network(text_type("%s/%d" % (list(new_addresses)[0], 20)), + strict=False) + self.assertEqual(prefix, IPv6Network(text_type('1::/20'))) self.sleep(1) # check that SLAAC address is deleted - fib = self.vapi.ip6_fib_dump() + fib = self.vapi.ip_route_dump(0, True) addresses = set(self.get_interface_addresses(fib, self.pg0)) new_addresses = addresses.difference(initial_addresses) self.assertEqual(len(new_addresses), 0) @@ -1300,6 +1313,14 @@ class TestIPv6RDControlPlane(TestIPv6ND): class IPv6NDProxyTest(TestIPv6ND): """ IPv6 ND ProxyTest Case """ + @classmethod + def setUpClass(cls): + super(IPv6NDProxyTest, cls).setUpClass() + + @classmethod + def tearDownClass(cls): + super(IPv6NDProxyTest, cls).tearDownClass() + def setUp(self): super(IPv6NDProxyTest, self).setUp() @@ -1346,8 +1367,8 @@ class IPv6NDProxyTest(TestIPv6ND): # Add proxy support for the host # self.vapi.ip6nd_proxy_add_del( - inet_pton(AF_INET6, self.pg0._remote_hosts[2].ip6), - self.pg1.sw_if_index) + ip=inet_pton(AF_INET6, self.pg0._remote_hosts[2].ip6), + sw_if_index=self.pg1.sw_if_index) # # try that NS again. this time we expect an NA back @@ -1413,8 +1434,8 @@ class IPv6NDProxyTest(TestIPv6ND): lladdr=self.pg0._remote_hosts[2].mac)) self.vapi.ip6nd_proxy_add_del( - inet_pton(AF_INET6, self.pg0._remote_hosts[3].ip6), - self.pg2.sw_if_index) + ip=inet_pton(AF_INET6, self.pg0._remote_hosts[3].ip6), + sw_if_index=self.pg2.sw_if_index) self.send_and_expect_na(self.pg2, ns_pg2, "NS to proxy entry other interface", @@ -1453,13 +1474,11 @@ class IPv6NDProxyTest(TestIPv6ND): # remove the proxy configs # self.vapi.ip6nd_proxy_add_del( - inet_pton(AF_INET6, self.pg0._remote_hosts[2].ip6), - self.pg1.sw_if_index, - is_del=1) + ip=inet_pton(AF_INET6, self.pg0._remote_hosts[2].ip6), + sw_if_index=self.pg1.sw_if_index, is_del=1) self.vapi.ip6nd_proxy_add_del( - inet_pton(AF_INET6, self.pg0._remote_hosts[3].ip6), - self.pg2.sw_if_index, - is_del=1) + ip=inet_pton(AF_INET6, self.pg0._remote_hosts[3].ip6), + sw_if_index=self.pg2.sw_if_index, is_del=1) self.assertFalse(find_nbr(self, self.pg2.sw_if_index, @@ -1491,6 +1510,14 @@ class IPv6NDProxyTest(TestIPv6ND): class TestIPNull(VppTestCase): """ IPv6 routes via NULL """ + @classmethod + def setUpClass(cls): + super(TestIPNull, cls).setUpClass() + + @classmethod + def tearDownClass(cls): + super(TestIPNull, cls).tearDownClass() + def setUp(self): super(TestIPNull, self).setUp() @@ -1520,7 +1547,10 @@ class TestIPNull(VppTestCase): # # A route via IP NULL that will reply with ICMP unreachables # - ip_unreach = VppIpRoute(self, "2001::", 64, [], is_unreach=1, is_ip6=1) + ip_unreach = VppIpRoute( + self, "2001::", 64, + [VppRoutePath("::", 0xffffffff, + type=FibPathType.FIB_PATH_TYPE_ICMP_UNREACH)]) ip_unreach.add_vpp_config() self.pg0.add_stream(p) @@ -1540,8 +1570,10 @@ class TestIPNull(VppTestCase): # # A route via IP NULL that will reply with ICMP prohibited # - ip_prohibit = VppIpRoute(self, "2001::1", 128, [], - is_prohibit=1, is_ip6=1) + ip_prohibit = VppIpRoute( + self, "2001::1", 128, + [VppRoutePath("::", 0xffffffff, + type=FibPathType.FIB_PATH_TYPE_ICMP_PROHIBIT)]) ip_prohibit.add_vpp_config() self.pg0.add_stream(p) @@ -1559,13 +1591,21 @@ class TestIPNull(VppTestCase): class TestIPDisabled(VppTestCase): """ IPv6 disabled """ + @classmethod + def setUpClass(cls): + super(TestIPDisabled, cls).setUpClass() + + @classmethod + def tearDownClass(cls): + super(TestIPDisabled, cls).tearDownClass() + def setUp(self): super(TestIPDisabled, self).setUp() # create 2 pg interfaces self.create_pg_interfaces(range(2)) - # PG0 is IP enalbed + # PG0 is IP enabled self.pg0.admin_up() self.pg0.config_ip6() self.pg0.resolve_ndp() @@ -1594,8 +1634,7 @@ class TestIPDisabled(VppTestCase): [VppMRoutePath(self.pg1.sw_if_index, MRouteItfFlags.MFIB_ITF_FLAG_ACCEPT), VppMRoutePath(self.pg0.sw_if_index, - MRouteItfFlags.MFIB_ITF_FLAG_FORWARD)], - is_ip6=1) + MRouteItfFlags.MFIB_ITF_FLAG_FORWARD)]) route_ff_01.add_vpp_config() pu = (Ether(src=self.pg1.remote_mac, @@ -1648,6 +1687,14 @@ class TestIPDisabled(VppTestCase): class TestIP6LoadBalance(VppTestCase): """ IPv6 Load-Balancing """ + @classmethod + def setUpClass(cls): + super(TestIP6LoadBalance, cls).setUpClass() + + @classmethod + def tearDownClass(cls): + super(TestIP6LoadBalance, cls).tearDownClass() + def setUp(self): super(TestIP6LoadBalance, self).setUp() @@ -1706,7 +1753,7 @@ class TestIP6LoadBalance(VppTestCase): src_ip_pkts = [] src_mpls_pkts = [] - for ii in range(65): + for ii in range(NUM_PKTS): port_ip_hdr = ( IPv6(dst="3000::1", src="3000:1::1") / inet6.UDP(sport=1234, dport=1234 + ii) / @@ -1742,16 +1789,13 @@ class TestIP6LoadBalance(VppTestCase): src_ip_hdr)) # - # A route for the IP pacekts + # A route for the IP packets # route_3000_1 = VppIpRoute(self, "3000::1", 128, [VppRoutePath(self.pg1.remote_ip6, - self.pg1.sw_if_index, - proto=DpoProto.DPO_PROTO_IP6), + self.pg1.sw_if_index), VppRoutePath(self.pg2.remote_ip6, - self.pg2.sw_if_index, - proto=DpoProto.DPO_PROTO_IP6)], - is_ip6=1) + self.pg2.sw_if_index)]) route_3000_1.add_vpp_config() # @@ -1766,12 +1810,10 @@ class TestIP6LoadBalance(VppTestCase): route_67 = VppMplsRoute(self, 67, 0, [VppRoutePath(self.pg1.remote_ip6, self.pg1.sw_if_index, - labels=[67], - proto=DpoProto.DPO_PROTO_IP6), + labels=[67]), VppRoutePath(self.pg2.remote_ip6, self.pg2.sw_if_index, - labels=[67], - proto=DpoProto.DPO_PROTO_IP6)]) + labels=[67])]) route_67.add_vpp_config() # @@ -1780,7 +1822,7 @@ class TestIP6LoadBalance(VppTestCase): # src,dst # We are not going to ensure equal amounts of packets across each link, # since the hash algorithm is statistical and therefore this can never - # be guaranteed. But wuth 64 different packets we do expect some + # be guaranteed. But with 64 different packets we do expect some # balancing. So instead just ensure there is traffic on each link. # self.send_and_expect_load_balancing(self.pg0, port_ip_pkts, @@ -1796,7 +1838,7 @@ class TestIP6LoadBalance(VppTestCase): # # The packets with Entropy label in should not load-balance, - # since the Entorpy value is fixed. + # since the Entropy value is fixed. # self.send_and_expect_one_itf(self.pg0, port_ent_pkts, self.pg1) @@ -1805,7 +1847,8 @@ class TestIP6LoadBalance(VppTestCase): # - now only the stream with differing source address will # load-balance # - self.vapi.set_ip_flow_hash(0, is_ip6=1, src=1, dst=1, sport=0, dport=0) + self.vapi.set_ip_flow_hash(vrf_id=0, src=1, dst=1, sport=0, dport=0, + is_ipv6=1) self.send_and_expect_load_balancing(self.pg0, src_ip_pkts, [self.pg1, self.pg2]) @@ -1816,7 +1859,8 @@ class TestIP6LoadBalance(VppTestCase): # # change the flow hash config back to defaults # - self.vapi.set_ip_flow_hash(0, is_ip6=1, src=1, dst=1, sport=1, dport=1) + self.vapi.set_ip_flow_hash(vrf_id=0, src=1, dst=1, sport=1, dport=1, + is_ipv6=1) # # Recursive prefixes @@ -1843,22 +1887,16 @@ class TestIP6LoadBalance(VppTestCase): route_3000_2 = VppIpRoute(self, "3000::2", 128, [VppRoutePath(self.pg3.remote_ip6, - self.pg3.sw_if_index, - proto=DpoProto.DPO_PROTO_IP6), + self.pg3.sw_if_index), VppRoutePath(self.pg4.remote_ip6, - self.pg4.sw_if_index, - proto=DpoProto.DPO_PROTO_IP6)], - is_ip6=1) + self.pg4.sw_if_index)]) route_3000_2.add_vpp_config() route_4000_1 = VppIpRoute(self, "4000::1", 128, [VppRoutePath("3000::1", - 0xffffffff, - proto=DpoProto.DPO_PROTO_IP6), + 0xffffffff), VppRoutePath("3000::2", - 0xffffffff, - proto=DpoProto.DPO_PROTO_IP6)], - is_ip6=1) + 0xffffffff)]) route_4000_1.add_vpp_config() # @@ -1889,16 +1927,12 @@ class TestIP6LoadBalance(VppTestCase): route_5000_2 = VppIpRoute(self, "5000::2", 128, [VppRoutePath(self.pg3.remote_ip6, - self.pg3.sw_if_index, - proto=DpoProto.DPO_PROTO_IP6)], - is_ip6=1) + self.pg3.sw_if_index)]) route_5000_2.add_vpp_config() route_6000_1 = VppIpRoute(self, "6000::1", 128, [VppRoutePath("5000::2", - 0xffffffff, - proto=DpoProto.DPO_PROTO_IP6)], - is_ip6=1) + 0xffffffff)]) route_6000_1.add_vpp_config() # @@ -1911,6 +1945,14 @@ class TestIP6LoadBalance(VppTestCase): class TestIP6Punt(VppTestCase): """ IPv6 Punt Police/Redirect """ + @classmethod + def setUpClass(cls): + super(TestIP6Punt, cls).setUpClass() + + @classmethod + def tearDownClass(cls): + super(TestIP6Punt, cls).tearDownClass() + def setUp(self): super(TestIP6Punt, self).setUp() @@ -1952,7 +1994,7 @@ class TestIP6Punt(VppTestCase): # # add a policer # - policer = self.vapi.policer_add_del("ip6-punt", 400, 0, 10, 0, + policer = self.vapi.policer_add_del(b"ip6-punt", 400, 0, 10, 0, rate_type=1) self.vapi.ip_punt_police(policer.policer_index, is_ip6=1) @@ -1962,7 +2004,7 @@ class TestIP6Punt(VppTestCase): self.pg_start() # - # the number of packet recieved should be greater than 0, + # the number of packet received should be greater than 0, # but not equal to the number sent, since some were policed # rx = self.pg1._get_capture(1) @@ -1973,7 +2015,7 @@ class TestIP6Punt(VppTestCase): # remove the policer. back to full rx # self.vapi.ip_punt_police(policer.policer_index, is_add=0, is_ip6=1) - self.vapi.policer_add_del("ip6-punt", 400, 0, 10, 0, + self.vapi.policer_add_del(b"ip6-punt", 400, 0, 10, 0, rate_type=1, is_add=0) self.send_and_expect(self.pg0, pkts, self.pg1) @@ -2039,6 +2081,14 @@ class TestIP6Punt(VppTestCase): class TestIPDeag(VppTestCase): """ IPv6 Deaggregate Routes """ + @classmethod + def setUpClass(cls): + super(TestIPDeag, cls).setUpClass() + + @classmethod + def tearDownClass(cls): + super(TestIPDeag, cls).tearDownClass() + def setUp(self): super(TestIPDeag, self).setUp() @@ -2075,16 +2125,14 @@ class TestIPDeag(VppTestCase): route_to_dst = VppIpRoute(self, "1::1", 128, [VppRoutePath("::", 0xffffffff, - nh_table_id=1, - proto=DpoProto.DPO_PROTO_IP6)], - is_ip6=1) - route_to_src = VppIpRoute(self, "1::2", 128, - [VppRoutePath("::", - 0xffffffff, - nh_table_id=2, - is_source_lookup=1, - proto=DpoProto.DPO_PROTO_IP6)], - is_ip6=1) + nh_table_id=1)]) + route_to_src = VppIpRoute( + self, "1::2", 128, + [VppRoutePath("::", + 0xffffffff, + nh_table_id=2, + type=FibPathType.FIB_PATH_TYPE_SOURCE_LOOKUP)]) + route_to_dst.add_vpp_config() route_to_src.add_vpp_config() @@ -2115,9 +2163,7 @@ class TestIPDeag(VppTestCase): # route_in_dst = VppIpRoute(self, "1::1", 128, [VppRoutePath(self.pg1.remote_ip6, - self.pg1.sw_if_index, - proto=DpoProto.DPO_PROTO_IP6)], - is_ip6=1, + self.pg1.sw_if_index)], table_id=1) route_in_dst.add_vpp_config() @@ -2128,9 +2174,7 @@ class TestIPDeag(VppTestCase): # route_in_src = VppIpRoute(self, "2::2", 128, [VppRoutePath(self.pg2.remote_ip6, - self.pg2.sw_if_index, - proto=DpoProto.DPO_PROTO_IP6)], - is_ip6=1, + self.pg2.sw_if_index)], table_id=2) route_in_src.add_vpp_config() self.send_and_expect(self.pg0, pkts_src, self.pg2) @@ -2140,9 +2184,7 @@ class TestIPDeag(VppTestCase): # route_loop = VppIpRoute(self, "3::3", 128, [VppRoutePath("::", - 0xffffffff, - proto=DpoProto.DPO_PROTO_IP6)], - is_ip6=1) + 0xffffffff)]) route_loop.add_vpp_config() p_l = (Ether(src=self.pg0.remote_mac, @@ -2158,6 +2200,14 @@ class TestIPDeag(VppTestCase): class TestIP6Input(VppTestCase): """ IPv6 Input Exception Test Cases """ + @classmethod + def setUpClass(cls): + super(TestIP6Input, cls).setUpClass() + + @classmethod + def tearDownClass(cls): + super(TestIP6Input, cls).tearDownClass() + def setUp(self): super(TestIP6Input, self).setUp() @@ -2187,7 +2237,7 @@ class TestIP6Input(VppTestCase): inet6.UDP(sport=1234, dport=1234) / Raw('\xa5' * 100)) - rx = self.send_and_expect(self.pg0, p_version * 65, self.pg0) + rx = self.send_and_expect(self.pg0, p_version * NUM_PKTS, self.pg0) rx = rx[0] icmp = rx[ICMPv6TimeExceeded] @@ -2226,10 +2276,23 @@ class TestIP6Input(VppTestCase): l4 / Raw('\xa5' * 100)) - self.send_and_assert_no_replies(self.pg0, p_version * 65, + self.send_and_assert_no_replies(self.pg0, p_version * NUM_PKTS, remark=msg or "", timeout=timeout) + def test_hop_by_hop(self): + """ Hop-by-hop header test """ + + p = (Ether(src=self.pg0.remote_mac, + dst=self.pg0.local_mac) / + IPv6(src=self.pg0.remote_ip6, dst=self.pg0.local_ip6) / + IPv6ExtHdrHopByHop() / + inet6.UDP(sport=1234, dport=1234) / + Raw('\xa5' * 100)) + + self.pg0.add_stream(p) + self.pg_enable_capture(self.pg_interfaces) + self.pg_start() if __name__ == '__main__': unittest.main(testRunner=VppTestRunner)