ip: Router ID included in flow hash
[vpp.git] / test / test_ip6.py
index 63d3e41..1b3b9d0 100644 (file)
@@ -19,13 +19,15 @@ 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, \
+    VppMRoutePath, VppMplsIpBind, \
     VppMplsRoute, VppMplsTable, VppIpTable, FibPathType, FibPathProto, \
-    VppIpInterfaceAddress, find_route_in_dump, find_mroute_in_dump
+    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
@@ -160,6 +162,7 @@ class TestIPv6ND(VppTestCase):
         self.assertEqual(ip.dst, dip)
 
 
+@tag_run_solo
 class TestIPv6(TestIPv6ND):
     """ IPv6 Test Case """
 
@@ -978,6 +981,80 @@ class TestIPv6(TestIPv6ND):
         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 """
 
@@ -1050,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 """
@@ -1071,6 +1166,7 @@ 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):
@@ -1088,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):
@@ -1748,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
@@ -1756,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,
@@ -1850,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)
@@ -1951,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,
@@ -1973,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])
@@ -1986,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
@@ -2111,9 +2222,9 @@ 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)
 
@@ -2122,7 +2233,9 @@ class TestIP6Punt(VppTestCase):
         #
         policer = VppPolicer(self, "ip6-punt", 400, 0, 10, 0, rate_type=1)
         policer.add_vpp_config()
-        self.vapi.ip_punt_police(policer.policer_index, is_ip6=1)
+        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)
@@ -2140,32 +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)
+        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"""
@@ -2173,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
@@ -2442,7 +2547,6 @@ class TestIPReplace(VppTestCase):
         for i in self.pg_interfaces:
             i.admin_up()
             i.config_ip6()
-            i.resolve_arp()
             i.generate_remote_hosts(2)
             self.tables.append(VppIpTable(self, table_id,
                                           True).add_vpp_config())
@@ -2452,11 +2556,13 @@ class TestIPReplace(VppTestCase):
         super(TestIPReplace, self).tearDown()
         for i in self.pg_interfaces:
             i.admin_down()
-            i.unconfig_ip4()
+            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 = [[], [], [], []]
@@ -2479,18 +2585,18 @@ class TestIPReplace(VppTestCase):
                 multi = VppIpMRoute(
                     self, "::",
                     "ff:2001::%d" % jj, 128,
-                    MRouteEntryFlags.MFIB_ENTRY_FLAG_NONE,
+                    MRouteEntryFlags.MFIB_API_ENTRY_FLAG_NONE,
                     [VppMRoutePath(self.pg0.sw_if_index,
-                                   MRouteItfFlags.MFIB_ITF_FLAG_ACCEPT,
+                                   MRouteItfFlags.MFIB_API_ITF_FLAG_ACCEPT,
                                    proto=FibPathProto.FIB_PATH_NH_PROTO_IP6),
                      VppMRoutePath(self.pg1.sw_if_index,
-                                   MRouteItfFlags.MFIB_ITF_FLAG_FORWARD,
+                                   MRouteItfFlags.MFIB_API_ITF_FLAG_FORWARD,
                                    proto=FibPathProto.FIB_PATH_NH_PROTO_IP6),
                      VppMRoutePath(self.pg2.sw_if_index,
-                                   MRouteItfFlags.MFIB_ITF_FLAG_FORWARD,
+                                   MRouteItfFlags.MFIB_API_ITF_FLAG_FORWARD,
                                    proto=FibPathProto.FIB_PATH_NH_PROTO_IP6),
                      VppMRoutePath(self.pg3.sw_if_index,
-                                   MRouteItfFlags.MFIB_ITF_FLAG_FORWARD,
+                                   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,
@@ -2560,5 +2666,254 @@ class TestIPReplace(VppTestCase):
             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)