X-Git-Url: https://gerrit.fd.io/r/gitweb?a=blobdiff_plain;f=test%2Ftest_ip6.py;h=1b3b9d0a1e611d55b80ddbd2b6f1cc179518d0e3;hb=3d5f08a82;hp=36532cedf128ecde78ae0a5911dd2c4445782558;hpb=efd7bc2b1c8db160933ed3e9ab3cde0d07aaf863;p=vpp.git diff --git a/test/test_ip6.py b/test/test_ip6.py index 36532cedf12..1b3b9d0a1e6 100644 --- a/test/test_ip6.py +++ b/test/test_ip6.py @@ -1,6 +1,7 @@ #!/usr/bin/env python3 import socket +from socket import inet_pton, inet_ntop import unittest from parameterized import parameterized @@ -10,24 +11,27 @@ 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, IPv6ExtHdrHopByHop + ICMPv6TimeExceeded, ICMPv6EchoRequest, ICMPv6EchoReply, \ + IPv6ExtHdrHopByHop, ICMPv6MLReport2, ICMPv6MLDMultAddrRec from scapy.layers.l2 import Ether, Dot1Q from scapy.packet import Raw -from scapy.utils import inet_pton, inet_ntop from scapy.utils6 import in6_getnsma, in6_getnsmac, in6_ptop, in6_islladdr, \ in6_mactoifaceid from six import moves -from framework import VppTestCase, VppTestRunner +from framework import VppTestCase, VppTestRunner, tag_run_solo from util import ppp, ip6_normalize, mk_ll_addr -from vpp_ip import DpoProto +from vpp_papi import VppEnum +from vpp_ip import DpoProto, VppIpPuntPolicer, VppIpPuntRedirect from vpp_ip_route import VppIpRoute, VppRoutePath, find_route, VppIpMRoute, \ - VppMRoutePath, MRouteItfFlags, MRouteEntryFlags, VppMplsIpBind, \ - VppMplsRoute, VppMplsTable, VppIpTable, FibPathType, \ - VppIpInterfaceAddress + VppMRoutePath, VppMplsIpBind, \ + VppMplsRoute, VppMplsTable, VppIpTable, FibPathType, FibPathProto, \ + VppIpInterfaceAddress, find_route_in_dump, find_mroute_in_dump, \ + VppIp6LinkLocalAddress from vpp_neighbor import find_nbr, VppNeighbor from vpp_pg_interface import is_ipv6_misc from vpp_sub_interface import VppSubInterface, VppDot1QSubint +from vpp_policer import VppPolicer from ipaddress import IPv6Network, IPv6Address AF_INET6 = socket.AF_INET6 @@ -138,6 +142,7 @@ class TestIPv6ND(VppTestCase): def send_and_expect_ns(self, tx_intf, rx_intf, pkts, tgt_ip, filter_out_fn=is_ipv6_misc): + self.vapi.cli("clear trace") tx_intf.add_stream(pkts) self.pg_enable_capture(self.pg_interfaces) self.pg_start() @@ -157,6 +162,7 @@ class TestIPv6ND(VppTestCase): self.assertEqual(ip.dst, dip) +@tag_run_solo class TestIPv6(TestIPv6ND): """ IPv6 Test Case """ @@ -222,7 +228,6 @@ class TestIPv6(TestIPv6ND): """Run standard test teardown and log ``show ip6 neighbors``.""" for i in self.interfaces: i.unconfig_ip6() - i.ip6_disable() i.admin_down() for i in self.sub_interfaces: i.remove_vpp_config() @@ -578,9 +583,12 @@ class TestIPv6(TestIPv6ND): self.pg0.remote_ip6, self.pg1.remote_hosts[1].ip6) - def validate_ra(self, intf, rx, dst_ip=None, mtu=9000, pi_opt=None): + def validate_ra(self, intf, rx, dst_ip=None, src_ip=None, + mtu=9000, pi_opt=None): if not dst_ip: dst_ip = intf.remote_ip6 + if not src_ip: + src_ip = mk_ll_addr(intf.local_mac) # unicasted packets must come to the unicast mac self.assertEqual(rx[Ether].dst, intf.remote_mac) @@ -595,8 +603,7 @@ class TestIPv6(TestIPv6ND): # and come from the router's link local self.assertTrue(in6_islladdr(rx[IPv6].src)) - self.assertEqual(in6_ptop(rx[IPv6].src), - in6_ptop(mk_ll_addr(intf.local_mac))) + self.assertEqual(in6_ptop(rx[IPv6].src), in6_ptop(src_ip)) # it should contain the links MTU ra = rx[ICMPv6ND_RA] @@ -635,7 +642,9 @@ class TestIPv6(TestIPv6ND): def send_and_expect_ra(self, intf, pkts, remark, dst_ip=None, filter_out_fn=is_ipv6_misc, - opt=None): + opt=None, + src_ip=None): + self.vapi.cli("clear trace") intf.add_stream(pkts) self.pg_enable_capture(self.pg_interfaces) self.pg_start() @@ -643,7 +652,7 @@ class TestIPv6(TestIPv6ND): self.assertEqual(len(rx), 1) rx = rx[0] - self.validate_ra(intf, rx, dst_ip, pi_opt=opt) + self.validate_ra(intf, rx, dst_ip, src_ip=src_ip, pi_opt=opt) def test_rs(self): """ IPv6 Router Solicitation Exceptions @@ -666,8 +675,7 @@ class TestIPv6(TestIPv6ND): # - expect an RA in return # p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) / - IPv6( - dst=self.pg0.local_ip6, src=self.pg0.remote_ip6) / + IPv6(dst=self.pg0.local_ip6, src=self.pg0.remote_ip6) / ICMPv6ND_RS()) pkts = [p] self.send_and_expect_ra(self.pg0, pkts, "Genuine RS") @@ -925,16 +933,127 @@ class TestIPv6(TestIPv6ND): self.pg1.local_ip6_prefix_len), is_no=1) + # + # change the link's link local, so we know that works too. + # + self.vapi.sw_interface_ip6_set_link_local_address( + sw_if_index=self.pg0.sw_if_index, + ip="fe80::88") + self.pg0.ip6_ra_config(send_unicast=1) self.send_and_expect_ra(self.pg0, p, "RA with Prefix reverted to defaults", - dst_ip=ll) + dst_ip=ll, + src_ip="fe80::88") # # Reset the periodic advertisements back to default values # self.pg0.ip6_ra_config(no=1, suppress=1, send_unicast=0) + def test_mld(self): + """ MLD Report """ + # + # test one MLD is sent after applying an IPv6 Address on an interface + # + self.pg_enable_capture(self.pg_interfaces) + self.pg_start() + + subitf = VppDot1QSubint(self, self.pg1, 99) + + subitf.admin_up() + subitf.config_ip6() + + rxs = self.pg1._get_capture(timeout=4, filter_out_fn=None) + + # + # hunt for the MLD on vlan 99 + # + for rx in rxs: + # make sure ipv6 packets with hop by hop options have + # correct checksums + self.assert_packet_checksums_valid(rx) + if rx.haslayer(IPv6ExtHdrHopByHop) and \ + rx.haslayer(Dot1Q) and \ + rx[Dot1Q].vlan == 99: + mld = rx[ICMPv6MLReport2] + + self.assertEqual(mld.records_number, 4) + + +class TestIPv6RouteLookup(VppTestCase): + """ IPv6 Route Lookup Test Case """ + routes = [] + + def route_lookup(self, prefix, exact): + return self.vapi.api(self.vapi.papi.ip_route_lookup, + { + 'table_id': 0, + 'exact': exact, + 'prefix': prefix, + }) + + @classmethod + def setUpClass(cls): + super(TestIPv6RouteLookup, cls).setUpClass() + + @classmethod + def tearDownClass(cls): + super(TestIPv6RouteLookup, cls).tearDownClass() + + def setUp(self): + super(TestIPv6RouteLookup, self).setUp() + + drop_nh = VppRoutePath("::1", 0xffffffff, + type=FibPathType.FIB_PATH_TYPE_DROP) + + # Add 3 routes + r = VppIpRoute(self, "2001:1111::", 32, [drop_nh]) + r.add_vpp_config() + self.routes.append(r) + + r = VppIpRoute(self, "2001:1111:2222::", 48, [drop_nh]) + r.add_vpp_config() + self.routes.append(r) + + r = VppIpRoute(self, "2001:1111:2222::1", 128, [drop_nh]) + r.add_vpp_config() + self.routes.append(r) + + def tearDown(self): + # Remove the routes we added + for r in self.routes: + r.remove_vpp_config() + + super(TestIPv6RouteLookup, self).tearDown() + + def test_exact_match(self): + # Verify we find the host route + prefix = "2001:1111:2222::1/128" + result = self.route_lookup(prefix, True) + assert (prefix == str(result.route.prefix)) + + # Verify we find a middle prefix route + prefix = "2001:1111:2222::/48" + result = self.route_lookup(prefix, True) + assert (prefix == str(result.route.prefix)) + + # Verify we do not find an available LPM. + with self.vapi.assert_negative_api_retval(): + self.route_lookup("2001::2/128", True) + + def test_longest_prefix_match(self): + # verify we find lpm + lpm_prefix = "2001:1111:2222::/48" + result = self.route_lookup("2001:1111:2222::2/128", False) + assert (lpm_prefix == str(result.route.prefix)) + + # Verify we find the exact when not requested + result = self.route_lookup(lpm_prefix, False) + assert (lpm_prefix == str(result.route.prefix)) + + # Can't seem to delete the default route so no negative LPM test. + class TestIPv6IfAddrRoute(VppTestCase): """ IPv6 Interface Addr Route Test Case """ @@ -1008,6 +1127,24 @@ class TestIPv6IfAddrRoute(VppTestCase): self.assertFalse(find_route(self, addr1, 128)) self.assertFalse(find_route(self, addr2, 128)) + def test_ipv6_ifaddr_del(self): + """ Delete an interface address that does not exist """ + + loopbacks = self.create_loopback_interfaces(1) + lo = self.lo_interfaces[0] + + lo.config_ip6() + lo.admin_up() + + # + # try and remove pg0's subnet from lo + # + with self.vapi.assert_negative_api_retval(): + self.vapi.sw_interface_add_del_address( + sw_if_index=lo.sw_if_index, + prefix=self.pg0.local_ip6_prefix, + is_add=0) + class TestICMPv6Echo(VppTestCase): """ ICMPv6 Echo Test Case """ @@ -1029,13 +1166,13 @@ class TestICMPv6Echo(VppTestCase): for i in self.pg_interfaces: i.admin_up() i.config_ip6() + i.resolve_ndp(link_layer=True) i.resolve_ndp() def tearDown(self): super(TestICMPv6Echo, self).tearDown() for i in self.pg_interfaces: i.unconfig_ip6() - i.ip6_disable() i.admin_down() def test_icmpv6_echo(self): @@ -1047,39 +1184,34 @@ class TestICMPv6Echo(VppTestCase): - Check outgoing ICMPv6 Echo Reply message on pg0 interface. """ - icmpv6_id = 0xb - icmpv6_seq = 5 - 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, - dst=self.pg0.local_ip6) / - ICMPv6EchoRequest( - id=icmpv6_id, - seq=icmpv6_seq, - data=icmpv6_data)) - - self.pg0.add_stream(p_echo_request) + # test both with global and local ipv6 addresses + dsts = (self.pg0.local_ip6, self.pg0.local_ip6_ll) + id = 0xb + seq = 5 + data = b'\x0a' * 18 + p = list() + for dst in dsts: + p.append((Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) / + IPv6(src=self.pg0.remote_ip6, dst=dst) / + ICMPv6EchoRequest(id=id, seq=seq, data=data))) + + self.pg0.add_stream(p) self.pg_enable_capture(self.pg_interfaces) self.pg_start() - - rx = self.pg0.get_capture(1) - rx = rx[0] - ether = rx[Ether] - ipv6 = rx[IPv6] - icmpv6 = rx[ICMPv6EchoReply] - - self.assertEqual(ether.src, self.pg0.local_mac) - self.assertEqual(ether.dst, self.pg0.remote_mac) - - self.assertEqual(ipv6.src, self.pg0.local_ip6) - self.assertEqual(ipv6.dst, self.pg0.remote_ip6) - - self.assertEqual( - icmp6types[icmpv6.type], "Echo Reply") - self.assertEqual(icmpv6.id, icmpv6_id) - self.assertEqual(icmpv6.seq, icmpv6_seq) - self.assertEqual(icmpv6.data, icmpv6_data) + rxs = self.pg0.get_capture(len(dsts)) + + for rx, dst in zip(rxs, dsts): + ether = rx[Ether] + ipv6 = rx[IPv6] + icmpv6 = rx[ICMPv6EchoReply] + self.assertEqual(ether.src, self.pg0.local_mac) + self.assertEqual(ether.dst, self.pg0.remote_mac) + self.assertEqual(ipv6.src, dst) + self.assertEqual(ipv6.dst, self.pg0.remote_ip6) + self.assertEqual(icmp6types[icmpv6.type], "Echo Reply") + self.assertEqual(icmpv6.id, id) + self.assertEqual(icmpv6.seq, seq) + self.assertEqual(icmpv6.data, data) class TestIPv6RD(TestIPv6ND): @@ -1159,7 +1291,7 @@ class TestIPv6RD(TestIPv6ND): def test_rd_receive_router_advertisement(self): """ Verify events triggered by received RA packets """ - self.vapi.want_ip6_ra_events() + self.vapi.want_ip6_ra_events(enable=1) prefix_info_1 = ICMPv6NDOptPrefixInfo( prefix="1::2", @@ -1265,6 +1397,17 @@ class TestIPv6RDControlPlane(TestIPv6ND): list.append(str(entry.route.prefix.network_address)) return list + def wait_for_no_default_route(self, n_tries=50, s_time=1): + while (n_tries): + fib = self.vapi.ip_route_dump(0, True) + default_routes = self.get_default_routes(fib) + if 0 == len(default_routes): + return True + n_tries = n_tries - 1 + self.sleep(s_time) + + return False + def test_all(self): """ Test handling of SLAAC addresses and default routes """ @@ -1364,9 +1507,7 @@ class TestIPv6RDControlPlane(TestIPv6ND): self.sleep_on_vpp_time(1) # check that default route is deleted - fib = self.vapi.ip_route_dump(0, True) - default_routes = self.get_default_routes(fib) - self.assertEqual(len(default_routes), 0) + self.assertTrue(self.wait_for_no_default_route()) # check FIB still contains the SLAAC address addresses = set(self.get_interface_addresses(fib, self.pg0)) @@ -1443,7 +1584,7 @@ class IPv6NDProxyTest(TestIPv6ND): # Add proxy support for the host # self.vapi.ip6nd_proxy_add_del( - ip=inet_pton(AF_INET6, self.pg0._remote_hosts[2].ip6), + is_add=1, ip=inet_pton(AF_INET6, self.pg0._remote_hosts[2].ip6), sw_if_index=self.pg1.sw_if_index) # @@ -1510,7 +1651,7 @@ class IPv6NDProxyTest(TestIPv6ND): lladdr=self.pg0._remote_hosts[2].mac)) self.vapi.ip6nd_proxy_add_del( - ip=inet_pton(AF_INET6, self.pg0._remote_hosts[3].ip6), + is_add=1, 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, @@ -1551,10 +1692,10 @@ class IPv6NDProxyTest(TestIPv6ND): # self.vapi.ip6nd_proxy_add_del( ip=inet_pton(AF_INET6, self.pg0._remote_hosts[2].ip6), - sw_if_index=self.pg1.sw_if_index, is_del=1) + sw_if_index=self.pg1.sw_if_index, is_add=0) self.vapi.ip6nd_proxy_add_del( ip=inet_pton(AF_INET6, self.pg0._remote_hosts[3].ip6), - sw_if_index=self.pg2.sw_if_index, is_del=1) + sw_if_index=self.pg2.sw_if_index, is_add=0) self.assertFalse(find_nbr(self, self.pg2.sw_if_index, @@ -1698,6 +1839,8 @@ class TestIPDisabled(VppTestCase): def test_ip_disabled(self): """ IP Disabled """ + MRouteItfFlags = VppEnum.vl_api_mfib_itf_flags_t + MRouteEntryFlags = VppEnum.vl_api_mfib_entry_flags_t # # An (S,G). # one accepting interface, pg0, 2 forwarding interfaces @@ -1706,11 +1849,11 @@ class TestIPDisabled(VppTestCase): self, "::", "ffef::1", 128, - MRouteEntryFlags.MFIB_ENTRY_FLAG_NONE, + MRouteEntryFlags.MFIB_API_ENTRY_FLAG_NONE, [VppMRoutePath(self.pg1.sw_if_index, - MRouteItfFlags.MFIB_ITF_FLAG_ACCEPT), + MRouteItfFlags.MFIB_API_ITF_FLAG_ACCEPT), VppMRoutePath(self.pg0.sw_if_index, - MRouteItfFlags.MFIB_ITF_FLAG_FORWARD)]) + MRouteItfFlags.MFIB_API_ITF_FLAG_FORWARD)]) route_ff_01.add_vpp_config() pu = (Ether(src=self.pg1.remote_mac, @@ -1800,9 +1943,12 @@ class TestIP6LoadBalance(VppTestCase): def send_and_expect_load_balancing(self, input, pkts, outputs): self.pg_send(input, pkts) + rxs = [] for oo in outputs: rx = oo._get_capture(1) self.assertNotEqual(0, len(rx)) + rxs.append(rx) + return rxs def send_and_expect_one_itf(self, input, pkts, itf): self.pg_send(input, pkts) @@ -1901,16 +2047,31 @@ class TestIP6LoadBalance(VppTestCase): # 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, - [self.pg1, self.pg2]) + rx = self.send_and_expect_load_balancing(self.pg0, port_ip_pkts, + [self.pg1, self.pg2]) + n_ip_pg0 = len(rx[0]) self.send_and_expect_load_balancing(self.pg0, src_ip_pkts, [self.pg1, self.pg2]) self.send_and_expect_load_balancing(self.pg0, port_mpls_pkts, [self.pg1, self.pg2]) self.send_and_expect_load_balancing(self.pg0, src_mpls_pkts, [self.pg1, self.pg2]) - self.send_and_expect_load_balancing(self.pg0, port_mpls_neos_pkts, - [self.pg1, self.pg2]) + rx = self.send_and_expect_load_balancing(self.pg0, port_mpls_neos_pkts, + [self.pg1, self.pg2]) + n_mpls_pg0 = len(rx[0]) + + # + # change the router ID and expect the distribution changes + # + self.vapi.set_ip_flow_hash_router_id(router_id=0x11111111) + + rx = self.send_and_expect_load_balancing(self.pg0, port_ip_pkts, + [self.pg1, self.pg2]) + self.assertNotEqual(n_ip_pg0, len(rx[0])) + + rx = self.send_and_expect_load_balancing(self.pg0, src_mpls_pkts, + [self.pg1, self.pg2]) + self.assertNotEqual(n_mpls_pg0, len(rx[0])) # # The packets with Entropy label in should not load-balance, @@ -1923,8 +2084,8 @@ class TestIP6LoadBalance(VppTestCase): # - now only the stream with differing source address will # load-balance # - self.vapi.set_ip_flow_hash(vrf_id=0, src=1, dst=1, sport=0, dport=0, - is_ipv6=1) + self.vapi.set_ip_flow_hash(vrf_id=0, src=1, dst=1, proto=1, + sport=0, dport=0, is_ipv6=1) self.send_and_expect_load_balancing(self.pg0, src_ip_pkts, [self.pg1, self.pg2]) @@ -1936,7 +2097,7 @@ class TestIP6LoadBalance(VppTestCase): # change the flow hash config back to defaults # self.vapi.set_ip_flow_hash(vrf_id=0, src=1, dst=1, sport=1, dport=1, - is_ipv6=1) + proto=1, is_ipv6=1) # # Recursive prefixes @@ -2061,18 +2222,20 @@ class TestIP6Punt(VppTestCase): # Configure a punt redirect via pg1. # nh_addr = self.pg1.remote_ip6 - self.vapi.ip_punt_redirect(self.pg0.sw_if_index, - self.pg1.sw_if_index, - nh_addr) + ip_punt_redirect = VppIpPuntRedirect(self, self.pg0.sw_if_index, + self.pg1.sw_if_index, nh_addr) + ip_punt_redirect.add_vpp_config() self.send_and_expect(self.pg0, pkts, self.pg1) # # add a policer # - 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) + policer = VppPolicer(self, "ip6-punt", 400, 0, 10, 0, rate_type=1) + policer.add_vpp_config() + ip_punt_policer = VppIpPuntPolicer(self, policer.policer_index, + is_ip6=True) + ip_punt_policer.add_vpp_config() self.vapi.cli("clear trace") self.pg0.add_stream(pkts) @@ -2090,33 +2253,25 @@ 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(b"ip6-punt", 400, 0, 10, 0, - rate_type=1, is_add=0) + ip_punt_policer.remove_vpp_config() + policer.remove_vpp_config() self.send_and_expect(self.pg0, pkts, self.pg1) # # remove the redirect. expect full drop. # - self.vapi.ip_punt_redirect(self.pg0.sw_if_index, - self.pg1.sw_if_index, - nh_addr, - is_add=0) + ip_punt_redirect.remove_vpp_config() self.send_and_assert_no_replies(self.pg0, pkts, "IP no punt config") # # Add a redirect that is not input port selective # - self.vapi.ip_punt_redirect(0xffffffff, - self.pg1.sw_if_index, - nh_addr) + ip_punt_redirect = VppIpPuntRedirect(self, 0xffffffff, + self.pg1.sw_if_index, nh_addr) + ip_punt_redirect.add_vpp_config() self.send_and_expect(self.pg0, pkts, self.pg1) - - self.vapi.ip_punt_redirect(0xffffffff, - self.pg1.sw_if_index, - nh_addr, - is_add=0) + ip_punt_redirect.remove_vpp_config() def test_ip_punt_dump(self): """ IP6 punt redirect dump""" @@ -2124,24 +2279,23 @@ class TestIP6Punt(VppTestCase): # # Configure a punt redirects # - nh_addr = self.pg3.remote_ip6 - self.vapi.ip_punt_redirect(self.pg0.sw_if_index, - self.pg3.sw_if_index, - nh_addr) - self.vapi.ip_punt_redirect(self.pg1.sw_if_index, - self.pg3.sw_if_index, - nh_addr) - self.vapi.ip_punt_redirect(self.pg2.sw_if_index, - self.pg3.sw_if_index, - '0::0') + nh_address = self.pg3.remote_ip6 + ipr_03 = VppIpPuntRedirect(self, self.pg0.sw_if_index, + self.pg3.sw_if_index, nh_address) + ipr_13 = VppIpPuntRedirect(self, self.pg1.sw_if_index, + self.pg3.sw_if_index, nh_address) + ipr_23 = VppIpPuntRedirect(self, self.pg2.sw_if_index, + self.pg3.sw_if_index, '0::0') + ipr_03.add_vpp_config() + ipr_13.add_vpp_config() + ipr_23.add_vpp_config() # # Dump pg0 punt redirects # - punts = self.vapi.ip_punt_redirect_dump(self.pg0.sw_if_index, - is_ipv6=1) - for p in punts: - self.assertEqual(p.punt.rx_sw_if_index, self.pg0.sw_if_index) + self.assertTrue(ipr_03.query_vpp_config()) + self.assertTrue(ipr_13.query_vpp_config()) + self.assertTrue(ipr_23.query_vpp_config()) # # Dump punt redirects for all interfaces @@ -2370,5 +2524,396 @@ class TestIP6Input(VppTestCase): self.pg_enable_capture(self.pg_interfaces) self.pg_start() + +class TestIPReplace(VppTestCase): + """ IPv6 Table Replace """ + + @classmethod + def setUpClass(cls): + super(TestIPReplace, cls).setUpClass() + + @classmethod + def tearDownClass(cls): + super(TestIPReplace, cls).tearDownClass() + + def setUp(self): + super(TestIPReplace, self).setUp() + + self.create_pg_interfaces(range(4)) + + table_id = 1 + self.tables = [] + + for i in self.pg_interfaces: + i.admin_up() + i.config_ip6() + i.generate_remote_hosts(2) + self.tables.append(VppIpTable(self, table_id, + True).add_vpp_config()) + table_id += 1 + + def tearDown(self): + super(TestIPReplace, self).tearDown() + for i in self.pg_interfaces: + i.admin_down() + i.unconfig_ip6() + + def test_replace(self): + """ IP Table Replace """ + + MRouteItfFlags = VppEnum.vl_api_mfib_itf_flags_t + MRouteEntryFlags = VppEnum.vl_api_mfib_entry_flags_t + N_ROUTES = 20 + links = [self.pg0, self.pg1, self.pg2, self.pg3] + routes = [[], [], [], []] + + # the sizes of 'empty' tables + for t in self.tables: + self.assertEqual(len(t.dump()), 2) + self.assertEqual(len(t.mdump()), 5) + + # load up the tables with some routes + for ii, t in enumerate(self.tables): + for jj in range(1, N_ROUTES): + uni = VppIpRoute( + self, "2001::%d" % jj if jj != 0 else "2001::", 128, + [VppRoutePath(links[ii].remote_hosts[0].ip6, + links[ii].sw_if_index), + VppRoutePath(links[ii].remote_hosts[1].ip6, + links[ii].sw_if_index)], + table_id=t.table_id).add_vpp_config() + multi = VppIpMRoute( + self, "::", + "ff:2001::%d" % jj, 128, + MRouteEntryFlags.MFIB_API_ENTRY_FLAG_NONE, + [VppMRoutePath(self.pg0.sw_if_index, + MRouteItfFlags.MFIB_API_ITF_FLAG_ACCEPT, + proto=FibPathProto.FIB_PATH_NH_PROTO_IP6), + VppMRoutePath(self.pg1.sw_if_index, + MRouteItfFlags.MFIB_API_ITF_FLAG_FORWARD, + proto=FibPathProto.FIB_PATH_NH_PROTO_IP6), + VppMRoutePath(self.pg2.sw_if_index, + MRouteItfFlags.MFIB_API_ITF_FLAG_FORWARD, + proto=FibPathProto.FIB_PATH_NH_PROTO_IP6), + VppMRoutePath(self.pg3.sw_if_index, + MRouteItfFlags.MFIB_API_ITF_FLAG_FORWARD, + proto=FibPathProto.FIB_PATH_NH_PROTO_IP6)], + table_id=t.table_id).add_vpp_config() + routes[ii].append({'uni': uni, + 'multi': multi}) + + # + # replace the tables a few times + # + for kk in range(3): + # replace each table + for t in self.tables: + t.replace_begin() + + # all the routes are still there + for ii, t in enumerate(self.tables): + dump = t.dump() + mdump = t.mdump() + for r in routes[ii]: + self.assertTrue(find_route_in_dump(dump, r['uni'], t)) + self.assertTrue(find_mroute_in_dump(mdump, r['multi'], t)) + + # redownload the even numbered routes + for ii, t in enumerate(self.tables): + for jj in range(0, N_ROUTES, 2): + routes[ii][jj]['uni'].add_vpp_config() + routes[ii][jj]['multi'].add_vpp_config() + + # signal each table converged + for t in self.tables: + t.replace_end() + + # we should find the even routes, but not the odd + for ii, t in enumerate(self.tables): + dump = t.dump() + mdump = t.mdump() + for jj in range(0, N_ROUTES, 2): + self.assertTrue(find_route_in_dump( + dump, routes[ii][jj]['uni'], t)) + self.assertTrue(find_mroute_in_dump( + mdump, routes[ii][jj]['multi'], t)) + for jj in range(1, N_ROUTES - 1, 2): + self.assertFalse(find_route_in_dump( + dump, routes[ii][jj]['uni'], t)) + self.assertFalse(find_mroute_in_dump( + mdump, routes[ii][jj]['multi'], t)) + + # reload all the routes + for ii, t in enumerate(self.tables): + for r in routes[ii]: + r['uni'].add_vpp_config() + r['multi'].add_vpp_config() + + # all the routes are still there + for ii, t in enumerate(self.tables): + dump = t.dump() + mdump = t.mdump() + for r in routes[ii]: + self.assertTrue(find_route_in_dump(dump, r['uni'], t)) + self.assertTrue(find_mroute_in_dump(mdump, r['multi'], t)) + + # + # finally flush the tables for good measure + # + for t in self.tables: + t.flush() + self.assertEqual(len(t.dump()), 2) + self.assertEqual(len(t.mdump()), 5) + + +class TestIP6Replace(VppTestCase): + """ IPv4 Interface Address Replace """ + + @classmethod + def setUpClass(cls): + super(TestIP6Replace, cls).setUpClass() + + @classmethod + def tearDownClass(cls): + super(TestIP6Replace, cls).tearDownClass() + + def setUp(self): + super(TestIP6Replace, self).setUp() + + self.create_pg_interfaces(range(4)) + + for i in self.pg_interfaces: + i.admin_up() + + def tearDown(self): + super(TestIP6Replace, self).tearDown() + for i in self.pg_interfaces: + i.admin_down() + + def get_n_pfxs(self, intf): + return len(self.vapi.ip_address_dump(intf.sw_if_index, True)) + + def test_replace(self): + """ IP interface address replace """ + + intf_pfxs = [[], [], [], []] + + # add prefixes to each of the interfaces + for i in range(len(self.pg_interfaces)): + intf = self.pg_interfaces[i] + + # 2001:16:x::1/64 + addr = "2001:16:%d::1" % intf.sw_if_index + a = VppIpInterfaceAddress(self, intf, addr, 64).add_vpp_config() + intf_pfxs[i].append(a) + + # 2001:16:x::2/64 - a different address in the same subnet as above + addr = "2001:16:%d::2" % intf.sw_if_index + a = VppIpInterfaceAddress(self, intf, addr, 64).add_vpp_config() + intf_pfxs[i].append(a) + + # 2001:15:x::2/64 - a different address and subnet + addr = "2001:15:%d::2" % intf.sw_if_index + a = VppIpInterfaceAddress(self, intf, addr, 64).add_vpp_config() + intf_pfxs[i].append(a) + + # a dump should n_address in it + for intf in self.pg_interfaces: + self.assertEqual(self.get_n_pfxs(intf), 3) + + # + # remove all the address thru a replace + # + self.vapi.sw_interface_address_replace_begin() + self.vapi.sw_interface_address_replace_end() + for intf in self.pg_interfaces: + self.assertEqual(self.get_n_pfxs(intf), 0) + + # + # add all the interface addresses back + # + for p in intf_pfxs: + for v in p: + v.add_vpp_config() + for intf in self.pg_interfaces: + self.assertEqual(self.get_n_pfxs(intf), 3) + + # + # replace again, but this time update/re-add the address on the first + # two interfaces + # + self.vapi.sw_interface_address_replace_begin() + + for p in intf_pfxs[:2]: + for v in p: + v.add_vpp_config() + + self.vapi.sw_interface_address_replace_end() + + # on the first two the address still exist, + # on the other two they do not + for intf in self.pg_interfaces[:2]: + self.assertEqual(self.get_n_pfxs(intf), 3) + for p in intf_pfxs[:2]: + for v in p: + self.assertTrue(v.query_vpp_config()) + for intf in self.pg_interfaces[2:]: + self.assertEqual(self.get_n_pfxs(intf), 0) + + # + # add all the interface addresses back on the last two + # + for p in intf_pfxs[2:]: + for v in p: + v.add_vpp_config() + for intf in self.pg_interfaces: + self.assertEqual(self.get_n_pfxs(intf), 3) + + # + # replace again, this time add different prefixes on all the interfaces + # + self.vapi.sw_interface_address_replace_begin() + + pfxs = [] + for intf in self.pg_interfaces: + # 2001:18:x::1/64 + addr = "2001:18:%d::1" % intf.sw_if_index + pfxs.append(VppIpInterfaceAddress(self, intf, addr, + 64).add_vpp_config()) + + self.vapi.sw_interface_address_replace_end() + + # only .18 should exist on each interface + for intf in self.pg_interfaces: + self.assertEqual(self.get_n_pfxs(intf), 1) + for pfx in pfxs: + self.assertTrue(pfx.query_vpp_config()) + + # + # remove everything + # + self.vapi.sw_interface_address_replace_begin() + self.vapi.sw_interface_address_replace_end() + for intf in self.pg_interfaces: + self.assertEqual(self.get_n_pfxs(intf), 0) + + # + # add prefixes to each interface. post-begin add the prefix from + # interface X onto interface Y. this would normally be an error + # since it would generate a 'duplicate address' warning. but in + # this case, since what is newly downloaded is sane, it's ok + # + for intf in self.pg_interfaces: + # 2001:18:x::1/64 + addr = "2001:18:%d::1" % intf.sw_if_index + VppIpInterfaceAddress(self, intf, addr, 64).add_vpp_config() + + self.vapi.sw_interface_address_replace_begin() + + pfxs = [] + for intf in self.pg_interfaces: + # 2001:18:x::1/64 + addr = "2001:18:%d::1" % (intf.sw_if_index + 1) + pfxs.append(VppIpInterfaceAddress(self, intf, + addr, 64).add_vpp_config()) + + self.vapi.sw_interface_address_replace_end() + + self.logger.info(self.vapi.cli("sh int addr")) + + for intf in self.pg_interfaces: + self.assertEqual(self.get_n_pfxs(intf), 1) + for pfx in pfxs: + self.assertTrue(pfx.query_vpp_config()) + + +class TestIP6LinkLocal(VppTestCase): + """ IPv6 Link Local """ + + @classmethod + def setUpClass(cls): + super(TestIP6LinkLocal, cls).setUpClass() + + @classmethod + def tearDownClass(cls): + super(TestIP6LinkLocal, cls).tearDownClass() + + def setUp(self): + super(TestIP6LinkLocal, self).setUp() + + self.create_pg_interfaces(range(2)) + + for i in self.pg_interfaces: + i.admin_up() + + def tearDown(self): + super(TestIP6LinkLocal, self).tearDown() + for i in self.pg_interfaces: + i.admin_down() + + def test_ip6_ll(self): + """ IPv6 Link Local """ + + # + # two APIs to add a link local address. + # 1 - just like any other prefix + # 2 - with the special set LL API + # + + # + # First with the API to set a 'normal' prefix + # + ll1 = "fe80:1::1" + ll2 = "fe80:2::2" + ll3 = "fe80:3::3" + + VppIpInterfaceAddress(self, self.pg0, ll1, 128).add_vpp_config() + + # + # should be able to ping the ll + # + p_echo_request_1 = (Ether(src=self.pg0.remote_mac, + dst=self.pg0.local_mac) / + IPv6(src=ll2, + dst=ll1) / + ICMPv6EchoRequest()) + + self.send_and_expect(self.pg0, [p_echo_request_1], self.pg0) + + # + # change the link-local on pg0 + # + v_ll3 = VppIpInterfaceAddress(self, self.pg0, + ll3, 128).add_vpp_config() + + p_echo_request_3 = (Ether(src=self.pg0.remote_mac, + dst=self.pg0.local_mac) / + IPv6(src=ll2, + dst=ll3) / + ICMPv6EchoRequest()) + + self.send_and_expect(self.pg0, [p_echo_request_3], self.pg0) + + # + # set a normal v6 prefix on the link + # + self.pg0.config_ip6() + + self.send_and_expect(self.pg0, [p_echo_request_3], self.pg0) + + # the link-local cannot be removed + with self.vapi.assert_negative_api_retval(): + v_ll3.remove_vpp_config() + + # + # Use the specific link-local API on pg1 + # + VppIp6LinkLocalAddress(self, self.pg1, ll1).add_vpp_config() + self.send_and_expect(self.pg1, [p_echo_request_1], self.pg1) + + VppIp6LinkLocalAddress(self, self.pg1, ll3).add_vpp_config() + self.send_and_expect(self.pg1, [p_echo_request_3], self.pg1) + + if __name__ == '__main__': unittest.main(testRunner=VppTestRunner)