+ ns_pg2 = (Ether(dst=in6_getnsmac(nsma), src=self.pg2.remote_mac) /
+ IPv6(dst=d,
+ src=self.pg0._remote_hosts[3].ip6) /
+ ICMPv6ND_NS(tgt=self.pg0.local_ip6) /
+ ICMPv6NDOptSrcLLAddr(
+ lladdr=self.pg0._remote_hosts[2].mac))
+
+ self.vapi.ip6nd_proxy_add_del(
+ 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,
+ "NS to proxy entry other interface",
+ dst_ip=self.pg0._remote_hosts[3].ip6,
+ tgt_ip=self.pg0.local_ip6)
+
+ self.assertTrue(find_nbr(self,
+ self.pg2.sw_if_index,
+ self.pg0._remote_hosts[3].ip6))
+
+ #
+ # hosts can communicate. pg2->pg1
+ #
+ t2 = (Ether(dst=self.pg2.local_mac,
+ src=self.pg0.remote_hosts[3].mac) /
+ IPv6(dst=self.pg0._remote_hosts[2].ip6,
+ src=self.pg0._remote_hosts[3].ip6) /
+ inet6.UDP(sport=10000, dport=20000) /
+ Raw(b'\xa5' * 100))
+
+ self.pg2.add_stream(t2)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ rx = self.pg1.get_capture(1)
+ rx = rx[0]
+
+ self.assertEqual(rx[Ether].dst, self.pg0._remote_hosts[2].mac)
+ self.assertEqual(rx[Ether].src, self.pg1.local_mac)
+
+ self.assertEqual(rx[IPv6].src,
+ t2[IPv6].src)
+ self.assertEqual(rx[IPv6].dst,
+ t2[IPv6].dst)
+
+ #
+ # remove the proxy configs
+ #
+ 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_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_add=0)
+
+ self.assertFalse(find_nbr(self,
+ self.pg2.sw_if_index,
+ self.pg0._remote_hosts[3].ip6))
+ self.assertFalse(find_nbr(self,
+ self.pg1.sw_if_index,
+ self.pg0._remote_hosts[2].ip6))
+
+ #
+ # no longer proxy-ing...
+ #
+ self.send_and_assert_no_replies(self.pg0, ns_pg0, "Proxy unconfigured")
+ self.send_and_assert_no_replies(self.pg1, ns_pg1, "Proxy unconfigured")
+ self.send_and_assert_no_replies(self.pg2, ns_pg2, "Proxy unconfigured")
+
+ #
+ # no longer forwarding. traffic generates NS out of the glean/main
+ # interface
+ #
+ self.pg2.add_stream(t2)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+
+ rx = self.pg0.get_capture(1)
+
+ self.assertTrue(rx[0].haslayer(ICMPv6ND_NS))
+
+
+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()
+
+ # create 2 pg interfaces
+ self.create_pg_interfaces(range(1))
+
+ for i in self.pg_interfaces:
+ i.admin_up()
+ i.config_ip6()
+ i.resolve_ndp()
+
+ def tearDown(self):
+ super(TestIPNull, self).tearDown()
+ for i in self.pg_interfaces:
+ i.unconfig_ip6()
+ i.admin_down()
+
+ def test_ip_null(self):
+ """ IP NULL route """
+
+ p = (Ether(src=self.pg0.remote_mac,
+ dst=self.pg0.local_mac) /
+ IPv6(src=self.pg0.remote_ip6, dst="2001::1") /
+ inet6.UDP(sport=1234, dport=1234) /
+ Raw(b'\xa5' * 100))
+
+ #
+ # A route via IP NULL that will reply with ICMP unreachables
+ #
+ 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)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+
+ rx = self.pg0.get_capture(1)
+ rx = rx[0]
+ icmp = rx[ICMPv6DestUnreach]
+
+ # 0 = "No route to destination"
+ self.assertEqual(icmp.code, 0)
+
+ # ICMP is rate limited. pause a bit
+ self.sleep(1)
+
+ #
+ # A route via IP NULL that will reply with ICMP prohibited
+ #
+ 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)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+
+ rx = self.pg0.get_capture(1)
+ rx = rx[0]
+ icmp = rx[ICMPv6DestUnreach]
+
+ # 1 = "Communication with destination administratively prohibited"
+ self.assertEqual(icmp.code, 1)
+
+
+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 enabled
+ self.pg0.admin_up()
+ self.pg0.config_ip6()
+ self.pg0.resolve_ndp()
+
+ # PG 1 is not IP enabled
+ self.pg1.admin_up()
+
+ def tearDown(self):
+ super(TestIPDisabled, self).tearDown()
+ for i in self.pg_interfaces:
+ i.unconfig_ip4()
+ i.admin_down()
+
+ 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
+ #
+ route_ff_01 = VppIpMRoute(
+ self,
+ "::",
+ "ffef::1", 128,
+ MRouteEntryFlags.MFIB_API_ENTRY_FLAG_NONE,
+ [VppMRoutePath(self.pg1.sw_if_index,
+ MRouteItfFlags.MFIB_API_ITF_FLAG_ACCEPT),
+ VppMRoutePath(self.pg0.sw_if_index,
+ MRouteItfFlags.MFIB_API_ITF_FLAG_FORWARD)])
+ route_ff_01.add_vpp_config()
+
+ pu = (Ether(src=self.pg1.remote_mac,
+ dst=self.pg1.local_mac) /
+ IPv6(src="2001::1", dst=self.pg0.remote_ip6) /
+ inet6.UDP(sport=1234, dport=1234) /
+ Raw(b'\xa5' * 100))
+ pm = (Ether(src=self.pg1.remote_mac,
+ dst=self.pg1.local_mac) /
+ IPv6(src="2001::1", dst="ffef::1") /
+ inet6.UDP(sport=1234, dport=1234) /
+ Raw(b'\xa5' * 100))
+
+ #
+ # PG1 does not forward IP traffic
+ #
+ self.send_and_assert_no_replies(self.pg1, pu, "IPv6 disabled")
+ self.send_and_assert_no_replies(self.pg1, pm, "IPv6 disabled")
+
+ #
+ # IP enable PG1
+ #
+ self.pg1.config_ip6()
+
+ #
+ # Now we get packets through
+ #
+ self.pg1.add_stream(pu)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ rx = self.pg0.get_capture(1)
+
+ self.pg1.add_stream(pm)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ rx = self.pg0.get_capture(1)
+
+ #
+ # Disable PG1
+ #
+ self.pg1.unconfig_ip6()
+
+ #
+ # PG1 does not forward IP traffic
+ #
+ self.send_and_assert_no_replies(self.pg1, pu, "IPv6 disabled")
+ self.send_and_assert_no_replies(self.pg1, pm, "IPv6 disabled")
+
+
+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()
+
+ self.create_pg_interfaces(range(5))
+
+ mpls_tbl = VppMplsTable(self, 0)
+ mpls_tbl.add_vpp_config()
+
+ for i in self.pg_interfaces:
+ i.admin_up()
+ i.config_ip6()
+ i.resolve_ndp()
+ i.enable_mpls()
+
+ def tearDown(self):
+ for i in self.pg_interfaces:
+ i.unconfig_ip6()
+ i.admin_down()
+ i.disable_mpls()
+ super(TestIP6LoadBalance, self).tearDown()
+
+ def pg_send(self, input, pkts):
+ self.vapi.cli("clear trace")
+ input.add_stream(pkts)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+
+ 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)
+ rx = itf.get_capture(len(pkts))
+
+ def test_ip6_load_balance(self):
+ """ IPv6 Load-Balancing """
+
+ #
+ # An array of packets that differ only in the destination port
+ # - IP only
+ # - MPLS EOS
+ # - MPLS non-EOS
+ # - MPLS non-EOS with an entropy label
+ #
+ port_ip_pkts = []
+ port_mpls_pkts = []
+ port_mpls_neos_pkts = []
+ port_ent_pkts = []
+
+ #
+ # An array of packets that differ only in the source address
+ #
+ src_ip_pkts = []
+ src_mpls_pkts = []
+
+ for ii in range(NUM_PKTS):
+ port_ip_hdr = (
+ IPv6(dst="3000::1", src="3000:1::1") /
+ inet6.UDP(sport=1234, dport=1234 + ii) /
+ Raw(b'\xa5' * 100))
+ port_ip_pkts.append((Ether(src=self.pg0.remote_mac,
+ dst=self.pg0.local_mac) /
+ port_ip_hdr))
+ port_mpls_pkts.append((Ether(src=self.pg0.remote_mac,
+ dst=self.pg0.local_mac) /
+ MPLS(label=66, ttl=2) /
+ port_ip_hdr))
+ port_mpls_neos_pkts.append((Ether(src=self.pg0.remote_mac,
+ dst=self.pg0.local_mac) /
+ MPLS(label=67, ttl=2) /
+ MPLS(label=77, ttl=2) /
+ port_ip_hdr))
+ port_ent_pkts.append((Ether(src=self.pg0.remote_mac,
+ dst=self.pg0.local_mac) /
+ MPLS(label=67, ttl=2) /
+ MPLS(label=14, ttl=2) /
+ MPLS(label=999, ttl=2) /
+ port_ip_hdr))
+ src_ip_hdr = (
+ IPv6(dst="3000::1", src="3000:1::%d" % ii) /
+ inet6.UDP(sport=1234, dport=1234) /
+ Raw(b'\xa5' * 100))
+ src_ip_pkts.append((Ether(src=self.pg0.remote_mac,
+ dst=self.pg0.local_mac) /
+ src_ip_hdr))
+ src_mpls_pkts.append((Ether(src=self.pg0.remote_mac,
+ dst=self.pg0.local_mac) /
+ MPLS(label=66, ttl=2) /
+ src_ip_hdr))
+
+ #
+ # A route for the IP packets
+ #
+ route_3000_1 = VppIpRoute(self, "3000::1", 128,
+ [VppRoutePath(self.pg1.remote_ip6,
+ self.pg1.sw_if_index),
+ VppRoutePath(self.pg2.remote_ip6,
+ self.pg2.sw_if_index)])
+ route_3000_1.add_vpp_config()
+
+ #
+ # a local-label for the EOS packets
+ #
+ binding = VppMplsIpBind(self, 66, "3000::1", 128, is_ip6=1)
+ binding.add_vpp_config()
+
+ #
+ # An MPLS route for the non-EOS packets
+ #
+ route_67 = VppMplsRoute(self, 67, 0,
+ [VppRoutePath(self.pg1.remote_ip6,
+ self.pg1.sw_if_index,
+ labels=[67]),
+ VppRoutePath(self.pg2.remote_ip6,
+ self.pg2.sw_if_index,
+ labels=[67])])
+ route_67.add_vpp_config()
+
+ #
+ # inject the packet on pg0 - expect load-balancing across the 2 paths
+ # - since the default hash config is to use IP src,dst and port
+ # 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 with 64 different packets we do expect some
+ # balancing. So instead just ensure there is traffic on each link.
+ #
+ 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])
+ 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,
+ # since the Entropy value is fixed.
+ #
+ self.send_and_expect_one_itf(self.pg0, port_ent_pkts, self.pg1)
+
+ #
+ # change the flow hash config so it's only IP src,dst
+ # - now only the stream with differing source address will
+ # load-balance
+ #
+ 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])
+ self.send_and_expect_load_balancing(self.pg0, src_mpls_pkts,
+ [self.pg1, self.pg2])
+ self.send_and_expect_one_itf(self.pg0, port_ip_pkts, self.pg2)
+
+ #
+ # 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,
+ proto=1, is_ipv6=1)
+
+ #
+ # Recursive prefixes
+ # - testing that 2 stages of load-balancing occurs and there is no
+ # polarisation (i.e. only 2 of 4 paths are used)
+ #
+ port_pkts = []
+ src_pkts = []
+
+ for ii in range(257):
+ port_pkts.append((Ether(src=self.pg0.remote_mac,
+ dst=self.pg0.local_mac) /
+ IPv6(dst="4000::1",
+ src="4000:1::1") /
+ inet6.UDP(sport=1234,
+ dport=1234 + ii) /
+ Raw(b'\xa5' * 100)))
+ src_pkts.append((Ether(src=self.pg0.remote_mac,
+ dst=self.pg0.local_mac) /
+ IPv6(dst="4000::1",
+ src="4000:1::%d" % ii) /
+ inet6.UDP(sport=1234, dport=1234) /
+ Raw(b'\xa5' * 100)))
+
+ route_3000_2 = VppIpRoute(self, "3000::2", 128,
+ [VppRoutePath(self.pg3.remote_ip6,
+ self.pg3.sw_if_index),
+ VppRoutePath(self.pg4.remote_ip6,
+ self.pg4.sw_if_index)])
+ route_3000_2.add_vpp_config()
+
+ route_4000_1 = VppIpRoute(self, "4000::1", 128,
+ [VppRoutePath("3000::1",
+ 0xffffffff),
+ VppRoutePath("3000::2",
+ 0xffffffff)])
+ route_4000_1.add_vpp_config()
+
+ #
+ # inject the packet on pg0 - expect load-balancing across all 4 paths
+ #
+ self.vapi.cli("clear trace")
+ self.send_and_expect_load_balancing(self.pg0, port_pkts,
+ [self.pg1, self.pg2,
+ self.pg3, self.pg4])
+ self.send_and_expect_load_balancing(self.pg0, src_pkts,
+ [self.pg1, self.pg2,
+ self.pg3, self.pg4])
+
+ #
+ # Recursive prefixes
+ # - testing that 2 stages of load-balancing no choices
+ #
+ port_pkts = []
+
+ for ii in range(257):
+ port_pkts.append((Ether(src=self.pg0.remote_mac,
+ dst=self.pg0.local_mac) /
+ IPv6(dst="6000::1",
+ src="6000:1::1") /
+ inet6.UDP(sport=1234,
+ dport=1234 + ii) /
+ Raw(b'\xa5' * 100)))
+
+ route_5000_2 = VppIpRoute(self, "5000::2", 128,
+ [VppRoutePath(self.pg3.remote_ip6,
+ self.pg3.sw_if_index)])
+ route_5000_2.add_vpp_config()
+
+ route_6000_1 = VppIpRoute(self, "6000::1", 128,
+ [VppRoutePath("5000::2",
+ 0xffffffff)])
+ route_6000_1.add_vpp_config()
+
+ #
+ # inject the packet on pg0 - expect load-balancing across all 4 paths
+ #
+ self.vapi.cli("clear trace")
+ self.send_and_expect_one_itf(self.pg0, port_pkts, self.pg3)
+
+
+class IP6PuntSetup(object):
+ """ Setup for IPv6 Punt Police/Redirect """
+
+ def punt_setup(self):
+ self.create_pg_interfaces(range(4))
+
+ for i in self.pg_interfaces:
+ i.admin_up()
+ i.config_ip6()
+ i.resolve_ndp()
+
+ self.pkt = (Ether(src=self.pg0.remote_mac,
+ dst=self.pg0.local_mac) /
+ IPv6(src=self.pg0.remote_ip6,
+ dst=self.pg0.local_ip6) /
+ inet6.TCP(sport=1234, dport=1234) /
+ Raw(b'\xa5' * 100))
+
+ def punt_teardown(self):
+ for i in self.pg_interfaces:
+ i.unconfig_ip6()
+ i.admin_down()
+
+
+class TestIP6Punt(IP6PuntSetup, VppTestCase):
+ """ IPv6 Punt Police/Redirect """
+
+ def setUp(self):
+ super(TestIP6Punt, self).setUp()
+ super(TestIP6Punt, self).punt_setup()
+
+ def tearDown(self):
+ super(TestIP6Punt, self).punt_teardown()
+ super(TestIP6Punt, self).tearDown()
+
+ def test_ip_punt(self):
+ """ IP6 punt police and redirect """
+
+ pkts = self.pkt * 1025
+
+ #
+ # Configure a punt redirect via pg1.
+ #
+ nh_addr = self.pg1.remote_ip6
+ 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 = 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)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+
+ #
+ # 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)
+ stats = policer.get_stats()
+
+ # Single rate policer - expect conform, violate but no exceed
+ self.assertGreater(stats['conform_packets'], 0)
+ self.assertEqual(stats['exceed_packets'], 0)
+ self.assertGreater(stats['violate_packets'], 0)
+
+ self.assertGreater(len(rx), 0)
+ self.assertLess(len(rx), len(pkts))
+
+ #
+ # remove the policer. back to full rx
+ #
+ 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.
+ #
+ 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
+ #
+ 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)
+ ip_punt_redirect.remove_vpp_config()
+
+ def test_ip_punt_dump(self):
+ """ IP6 punt redirect dump"""
+
+ #
+ # Configure a punt redirects
+ #
+ 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
+ #
+ 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
+ #
+ punts = self.vapi.ip_punt_redirect_dump(0xffffffff, is_ipv6=1)
+ 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, self.pg3.remote_ip6)
+ self.assertEqual(str(punts[2].punt.nh), '::')
+
+
+class TestIP6PuntHandoff(IP6PuntSetup, VppTestCase):
+ """ IPv6 Punt Police/Redirect """
+ vpp_worker_count = 2
+
+ def setUp(self):
+ super(TestIP6PuntHandoff, self).setUp()
+ super(TestIP6PuntHandoff, self).punt_setup()
+
+ def tearDown(self):
+ super(TestIP6PuntHandoff, self).punt_teardown()
+ super(TestIP6PuntHandoff, self).tearDown()
+
+ def test_ip_punt(self):
+ """ IP6 punt policer thread handoff """
+ pkts = self.pkt * NUM_PKTS
+
+ #
+ # Configure a punt redirect via pg1.
+ #
+ nh_addr = self.pg1.remote_ip6
+ ip_punt_redirect = VppIpPuntRedirect(self, self.pg0.sw_if_index,
+ self.pg1.sw_if_index, nh_addr)
+ ip_punt_redirect.add_vpp_config()
+
+ action_tx = PolicerAction(
+ VppEnum.vl_api_sse2_qos_action_type_t.SSE2_QOS_ACTION_API_TRANSMIT,
+ 0)
+ #
+ # This policer drops no packets, we are just
+ # testing that they get to the right thread.
+ #
+ policer = VppPolicer(self, "ip6-punt", 400, 0, 10, 0, 1,
+ 0, 0, False, action_tx, action_tx, action_tx)
+ policer.add_vpp_config()
+ ip_punt_policer = VppIpPuntPolicer(self, policer.policer_index,
+ is_ip6=True)
+ ip_punt_policer.add_vpp_config()
+
+ for worker in [0, 1]:
+ self.send_and_expect(self.pg0, pkts, self.pg1, worker=worker)
+ if worker == 0:
+ self.logger.debug(self.vapi.cli("show trace max 100"))
+
+ # Combined stats, all threads
+ stats = policer.get_stats()
+
+ # Single rate policer - expect conform, violate but no exceed
+ self.assertGreater(stats['conform_packets'], 0)
+ self.assertEqual(stats['exceed_packets'], 0)
+ self.assertGreater(stats['violate_packets'], 0)
+
+ # Worker 0, should have done all the policing
+ stats0 = policer.get_stats(worker=0)
+ self.assertEqual(stats, stats0)
+
+ # Worker 1, should have handed everything off
+ stats1 = policer.get_stats(worker=1)
+ self.assertEqual(stats1['conform_packets'], 0)
+ self.assertEqual(stats1['exceed_packets'], 0)
+ self.assertEqual(stats1['violate_packets'], 0)
+
+ # Bind the policer to worker 1 and repeat
+ policer.bind_vpp_config(1, True)
+ for worker in [0, 1]:
+ self.send_and_expect(self.pg0, pkts, self.pg1, worker=worker)
+ self.logger.debug(self.vapi.cli("show trace max 100"))
+
+ # The 2 workers should now have policed the same amount
+ stats = policer.get_stats()
+ stats0 = policer.get_stats(worker=0)
+ stats1 = policer.get_stats(worker=1)
+
+ self.assertGreater(stats0['conform_packets'], 0)
+ self.assertEqual(stats0['exceed_packets'], 0)
+ self.assertGreater(stats0['violate_packets'], 0)
+
+ self.assertGreater(stats1['conform_packets'], 0)
+ self.assertEqual(stats1['exceed_packets'], 0)
+ self.assertGreater(stats1['violate_packets'], 0)
+
+ self.assertEqual(stats0['conform_packets'] + stats1['conform_packets'],
+ stats['conform_packets'])
+
+ self.assertEqual(stats0['violate_packets'] + stats1['violate_packets'],
+ stats['violate_packets'])
+
+ # Unbind the policer and repeat
+ policer.bind_vpp_config(1, False)
+ for worker in [0, 1]:
+ self.send_and_expect(self.pg0, pkts, self.pg1, worker=worker)
+ self.logger.debug(self.vapi.cli("show trace max 100"))
+
+ # The policer should auto-bind to worker 0 when packets arrive
+ stats = policer.get_stats()
+ stats0new = policer.get_stats(worker=0)
+ stats1new = policer.get_stats(worker=1)
+
+ self.assertGreater(stats0new['conform_packets'],
+ stats0['conform_packets'])
+ self.assertEqual(stats0new['exceed_packets'], 0)
+ self.assertGreater(stats0new['violate_packets'],
+ stats0['violate_packets'])
+
+ self.assertEqual(stats1, stats1new)
+
+ #
+ # Clean up
+ #
+ ip_punt_policer.remove_vpp_config()
+ policer.remove_vpp_config()
+ ip_punt_redirect.remove_vpp_config()
+
+
+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()
+
+ self.create_pg_interfaces(range(3))
+
+ for i in self.pg_interfaces:
+ i.admin_up()
+ i.config_ip6()
+ i.resolve_ndp()
+
+ def tearDown(self):
+ super(TestIPDeag, self).tearDown()
+ for i in self.pg_interfaces:
+ i.unconfig_ip6()
+ i.admin_down()
+
+ def test_ip_deag(self):
+ """ IP Deag Routes """
+
+ #
+ # Create a table to be used for:
+ # 1 - another destination address lookup
+ # 2 - a source address lookup
+ #
+ table_dst = VppIpTable(self, 1, is_ip6=1)
+ table_src = VppIpTable(self, 2, is_ip6=1)
+ table_dst.add_vpp_config()
+ table_src.add_vpp_config()
+
+ #
+ # Add a route in the default table to point to a deag/
+ # second lookup in each of these tables
+ #
+ route_to_dst = VppIpRoute(self, "1::1", 128,
+ [VppRoutePath("::",
+ 0xffffffff,
+ 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()
+
+ #
+ # packets to these destination are dropped, since they'll
+ # hit the respective default routes in the second table
+ #
+ p_dst = (Ether(src=self.pg0.remote_mac,
+ dst=self.pg0.local_mac) /
+ IPv6(src="5::5", dst="1::1") /
+ inet6.TCP(sport=1234, dport=1234) /
+ Raw(b'\xa5' * 100))
+ p_src = (Ether(src=self.pg0.remote_mac,
+ dst=self.pg0.local_mac) /
+ IPv6(src="2::2", dst="1::2") /
+ inet6.TCP(sport=1234, dport=1234) /
+ Raw(b'\xa5' * 100))
+ pkts_dst = p_dst * 257
+ pkts_src = p_src * 257
+
+ self.send_and_assert_no_replies(self.pg0, pkts_dst,
+ "IP in dst table")
+ self.send_and_assert_no_replies(self.pg0, pkts_src,
+ "IP in src table")
+
+ #
+ # add a route in the dst table to forward via pg1
+ #
+ route_in_dst = VppIpRoute(self, "1::1", 128,
+ [VppRoutePath(self.pg1.remote_ip6,
+ self.pg1.sw_if_index)],
+ table_id=1)
+ route_in_dst.add_vpp_config()
+
+ self.send_and_expect(self.pg0, pkts_dst, self.pg1)
+
+ #
+ # add a route in the src table to forward via pg2
+ #
+ route_in_src = VppIpRoute(self, "2::2", 128,
+ [VppRoutePath(self.pg2.remote_ip6,
+ self.pg2.sw_if_index)],
+ table_id=2)
+ route_in_src.add_vpp_config()
+ self.send_and_expect(self.pg0, pkts_src, self.pg2)
+
+ #
+ # loop in the lookup DP
+ #
+ route_loop = VppIpRoute(self, "3::3", 128,
+ [VppRoutePath("::",
+ 0xffffffff)])
+ route_loop.add_vpp_config()
+
+ p_l = (Ether(src=self.pg0.remote_mac,
+ dst=self.pg0.local_mac) /
+ IPv6(src="3::4", dst="3::3") /
+ inet6.TCP(sport=1234, dport=1234) /
+ Raw(b'\xa5' * 100))
+
+ self.send_and_assert_no_replies(self.pg0, p_l * 257,
+ "IP lookup loop")
+
+
+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()
+
+ self.create_pg_interfaces(range(2))
+
+ for i in self.pg_interfaces:
+ i.admin_up()
+ i.config_ip6()
+ i.resolve_ndp()
+
+ def tearDown(self):
+ super(TestIP6Input, self).tearDown()
+ for i in self.pg_interfaces:
+ i.unconfig_ip6()
+ i.admin_down()
+
+ def test_ip_input_icmp_reply(self):
+ """ IP6 Input Exception - Return ICMP (3,0) """
+ #
+ # hop limit - ICMP replies
+ #
+ p_version = (Ether(src=self.pg0.remote_mac,
+ dst=self.pg0.local_mac) /
+ IPv6(src=self.pg0.remote_ip6,
+ dst=self.pg1.remote_ip6,
+ hlim=1) /
+ inet6.UDP(sport=1234, dport=1234) /
+ Raw(b'\xa5' * 100))
+
+ rx = self.send_and_expect(self.pg0, p_version * NUM_PKTS, self.pg0)
+ rx = rx[0]
+ icmp = rx[ICMPv6TimeExceeded]
+
+ # 0: "hop limit exceeded in transit",
+ self.assertEqual((icmp.type, icmp.code), (3, 0))
+
+ icmpv6_data = '\x0a' * 18
+ all_0s = "::"
+ all_1s = "FFFF:FFFF:FFFF:FFFF:FFFF:FFFF:FFFF:FFFF"
+
+ @parameterized.expand([
+ # Name, src, dst, l4proto, msg, timeout
+ ("src='iface', dst='iface'", None, None,
+ inet6.UDP(sport=1234, dport=1234), "funky version", None),
+ ("src='All 0's', dst='iface'", all_0s, None,
+ ICMPv6EchoRequest(id=0xb, seq=5, data=icmpv6_data), None, 0.1),
+ ("src='iface', dst='All 0's'", None, all_0s,
+ ICMPv6EchoRequest(id=0xb, seq=5, data=icmpv6_data), None, 0.1),
+ ("src='All 1's', dst='iface'", all_1s, None,
+ ICMPv6EchoRequest(id=0xb, seq=5, data=icmpv6_data), None, 0.1),
+ ("src='iface', dst='All 1's'", None, all_1s,
+ ICMPv6EchoRequest(id=0xb, seq=5, data=icmpv6_data), None, 0.1),
+ ("src='All 1's', dst='All 1's'", all_1s, all_1s,
+ ICMPv6EchoRequest(id=0xb, seq=5, data=icmpv6_data), None, 0.1),
+
+ ])
+ def test_ip_input_no_replies(self, name, src, dst, l4, msg, timeout):
+
+ self._testMethodDoc = 'IPv6 Input Exception - %s' % name
+
+ p_version = (Ether(src=self.pg0.remote_mac,
+ dst=self.pg0.local_mac) /
+ IPv6(src=src or self.pg0.remote_ip6,
+ dst=dst or self.pg1.remote_ip6,
+ version=3) /
+ l4 /
+ Raw(b'\xa5' * 100))
+
+ 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(b'\xa5' * 100))
+
+ self.pg0.add_stream(p)
+ 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"
+
+ VppNeighbor(self,
+ self.pg0.sw_if_index,
+ self.pg0.remote_mac,
+ ll2).add_vpp_config()
+
+ 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)
+
+ def test_ip6_ll_p2p(self):
+ """ IPv6 Link Local P2P (GRE)"""
+
+ self.pg0.config_ip4()
+ self.pg0.resolve_arp()
+ gre_if = VppGreInterface(self,
+ self.pg0.local_ip4,
+ self.pg0.remote_ip4).add_vpp_config()
+ gre_if.admin_up()
+
+ ll1 = "fe80:1::1"
+ ll2 = "fe80:2::2"
+
+ VppIpInterfaceAddress(self, gre_if, ll1, 128).add_vpp_config()
+
+ self.logger.info(self.vapi.cli("sh ip6-ll gre0 fe80:2::2"))
+
+ p_echo_request_1 = (Ether(src=self.pg0.remote_mac,
+ dst=self.pg0.local_mac) /
+ IP(src=self.pg0.remote_ip4,
+ dst=self.pg0.local_ip4) /
+ GRE() /
+ IPv6(src=ll2, dst=ll1) /
+ ICMPv6EchoRequest())
+ self.send_and_expect(self.pg0, [p_echo_request_1], self.pg0)
+
+ self.pg0.unconfig_ip4()
+ gre_if.remove_vpp_config()
+
+ def test_ip6_ll_p2mp(self):
+ """ IPv6 Link Local P2MP (GRE)"""
+
+ self.pg0.config_ip4()
+ self.pg0.resolve_arp()
+
+ gre_if = VppGreInterface(
+ self,
+ self.pg0.local_ip4,
+ "0.0.0.0",
+ mode=(VppEnum.vl_api_tunnel_mode_t.
+ TUNNEL_API_MODE_MP)).add_vpp_config()
+ gre_if.admin_up()
+
+ ll1 = "fe80:1::1"
+ ll2 = "fe80:2::2"
+
+ VppIpInterfaceAddress(self, gre_if, ll1, 128).add_vpp_config()
+
+ p_echo_request_1 = (Ether(src=self.pg0.remote_mac,
+ dst=self.pg0.local_mac) /
+ IP(src=self.pg0.remote_ip4,
+ dst=self.pg0.local_ip4) /
+ GRE() /
+ IPv6(src=ll2, dst=ll1) /
+ ICMPv6EchoRequest())
+
+ # no route back at this point
+ self.send_and_assert_no_replies(self.pg0, [p_echo_request_1])
+
+ # add teib entry for the peer
+ teib = VppTeib(self, gre_if, ll2, self.pg0.remote_ip4)
+ teib.add_vpp_config()
+
+ self.logger.info(self.vapi.cli("sh ip6-ll gre0 %s" % ll2))
+ self.send_and_expect(self.pg0, [p_echo_request_1], self.pg0)
+
+ # teardown
+ self.pg0.unconfig_ip4()
+
+
+class TestIPv6PathMTU(VppTestCase):
+ """ IPv6 Path MTU """
+
+ def setUp(self):
+ super(TestIPv6PathMTU, self).setUp()
+
+ self.create_pg_interfaces(range(2))
+
+ # setup all interfaces
+ for i in self.pg_interfaces:
+ i.admin_up()
+ i.config_ip6()
+ i.resolve_ndp()
+
+ def tearDown(self):
+ super(TestIPv6PathMTU, self).tearDown()
+ for i in self.pg_interfaces:
+ i.unconfig_ip6()
+ i.admin_down()
+
+ def test_path_mtu_local(self):
+ """ Path MTU for attached neighbour """
+
+ self.vapi.cli("set log class ip level debug")
+ #
+ # The goal here is not test that fragmentation works correctly,
+ # that's done elsewhere, the intent is to ensure that the Path MTU
+ # settings are honoured.
+ #
+
+ #
+ # IPv6 will only frag locally generated packets, so use tunnelled
+ # packets post encap
+ #
+ tun = VppIpIpTunInterface(
+ self,
+ self.pg1,
+ self.pg1.local_ip6,
+ self.pg1.remote_ip6)
+ tun.add_vpp_config()
+ tun.admin_up()
+ tun.config_ip6()
+
+ # set the interface MTU to a reasonable value
+ self.vapi.sw_interface_set_mtu(self.pg1.sw_if_index,
+ [2800, 0, 0, 0])
+
+ p_2k = (Ether(dst=self.pg0.local_mac,
+ src=self.pg0.remote_mac) /
+ IPv6(src=self.pg0.remote_ip6,
+ dst=tun.remote_ip6) /
+ UDP(sport=1234, dport=5678) /
+ Raw(b'0xa' * 1000))
+ p_1k = (Ether(dst=self.pg0.local_mac,
+ src=self.pg0.remote_mac) /
+ IPv6(src=self.pg0.remote_ip6,
+ dst=tun.remote_ip6) /
+ UDP(sport=1234, dport=5678) /
+ Raw(b'0xa' * 600))
+
+ nbr = VppNeighbor(self,
+ self.pg1.sw_if_index,
+ self.pg1.remote_mac,
+ self.pg1.remote_ip6).add_vpp_config()
+
+ # this is now the interface MTU frags
+ self.send_and_expect(self.pg0, [p_2k], self.pg1, n_rx=2)
+ self.send_and_expect(self.pg0, [p_1k], self.pg1)
+
+ # drop the path MTU for this neighbour to below the interface MTU
+ # expect more frags
+ pmtu = VppIpPathMtu(self, self.pg1.remote_ip6, 1300).add_vpp_config()
+
+ # print/format the adj delegate and trackers
+ self.logger.info(self.vapi.cli("sh ip pmtu"))
+ self.logger.info(self.vapi.cli("sh adj 7"))
+
+ self.send_and_expect(self.pg0, [p_2k], self.pg1, n_rx=3)
+ self.send_and_expect(self.pg0, [p_1k], self.pg1, n_rx=2)
+
+ # increase the path MTU to more than the interface
+ # expect to use the interface MTU
+ pmtu.modify(8192)
+
+ self.send_and_expect(self.pg0, [p_2k], self.pg1, n_rx=2)
+ self.send_and_expect(self.pg0, [p_1k], self.pg1)
+
+ # go back to an MTU from the path
+ pmtu.modify(1300)
+
+ self.send_and_expect(self.pg0, [p_2k], self.pg1, n_rx=3)
+ self.send_and_expect(self.pg0, [p_1k], self.pg1, n_rx=2)
+
+ # raise the interface's MTU
+ # should still use that of the path
+ self.vapi.sw_interface_set_mtu(self.pg1.sw_if_index,
+ [2000, 0, 0, 0])
+ self.send_and_expect(self.pg0, [p_2k], self.pg1, n_rx=3)
+ self.send_and_expect(self.pg0, [p_1k], self.pg1, n_rx=2)
+
+ # set path high and interface low
+ pmtu.modify(2000)
+ self.vapi.sw_interface_set_mtu(self.pg1.sw_if_index,
+ [1300, 0, 0, 0])
+ self.send_and_expect(self.pg0, [p_2k], self.pg1, n_rx=3)
+ self.send_and_expect(self.pg0, [p_1k], self.pg1, n_rx=2)
+
+ # remove the path MTU
+ self.vapi.sw_interface_set_mtu(self.pg1.sw_if_index,
+ [2800, 0, 0, 0])
+ pmtu.modify(0)
+
+ self.send_and_expect(self.pg0, [p_2k], self.pg1, n_rx=2)
+ self.send_and_expect(self.pg0, [p_1k], self.pg1)
+
+ def test_path_mtu_remote(self):
+ """ Path MTU for remote neighbour """
+
+ self.vapi.cli("set log class ip level debug")
+ #
+ # The goal here is not test that fragmentation works correctly,
+ # that's done elsewhere, the intent is to ensure that the Path MTU
+ # settings are honoured.
+ #
+ tun_dst = "2001::1"
+
+ route = VppIpRoute(
+ self, tun_dst, 64,
+ [VppRoutePath(self.pg1.remote_ip6,
+ self.pg1.sw_if_index)]).add_vpp_config()
+
+ #
+ # IPv6 will only frag locally generated packets, so use tunnelled
+ # packets post encap
+ #
+ tun = VppIpIpTunInterface(
+ self,
+ self.pg1,
+ self.pg1.local_ip6,
+ tun_dst)
+ tun.add_vpp_config()
+ tun.admin_up()
+ tun.config_ip6()
+
+ # set the interface MTU to a reasonable value
+ self.vapi.sw_interface_set_mtu(self.pg1.sw_if_index,
+ [2800, 0, 0, 0])
+
+ p_2k = (Ether(dst=self.pg0.local_mac,
+ src=self.pg0.remote_mac) /
+ IPv6(src=self.pg0.remote_ip6,
+ dst=tun.remote_ip6) /
+ UDP(sport=1234, dport=5678) /
+ Raw(b'0xa' * 1000))
+ p_1k = (Ether(dst=self.pg0.local_mac,
+ src=self.pg0.remote_mac) /
+ IPv6(src=self.pg0.remote_ip6,
+ dst=tun.remote_ip6) /
+ UDP(sport=1234, dport=5678) /
+ Raw(b'0xa' * 600))
+
+ nbr = VppNeighbor(self,
+ self.pg1.sw_if_index,
+ self.pg1.remote_mac,
+ self.pg1.remote_ip6).add_vpp_config()
+
+ # this is now the interface MTU frags
+ self.send_and_expect(self.pg0, [p_2k], self.pg1, n_rx=2)
+ self.send_and_expect(self.pg0, [p_1k], self.pg1)
+
+ # drop the path MTU for this neighbour to below the interface MTU
+ # expect more frags
+ pmtu = VppIpPathMtu(self, tun_dst, 1300).add_vpp_config()
+
+ # print/format the fib entry/dpo
+ self.logger.info(self.vapi.cli("sh ip6 fib 2001::1"))
+
+ self.send_and_expect(self.pg0, [p_2k], self.pg1, n_rx=3)
+ self.send_and_expect(self.pg0, [p_1k], self.pg1, n_rx=2)
+
+ # increase the path MTU to more than the interface
+ # expect to use the interface MTU
+ pmtu.modify(8192)
+
+ self.send_and_expect(self.pg0, [p_2k], self.pg1, n_rx=2)
+ self.send_and_expect(self.pg0, [p_1k], self.pg1)
+
+ # go back to an MTU from the path
+ pmtu.modify(1300)
+
+ self.send_and_expect(self.pg0, [p_2k], self.pg1, n_rx=3)
+ self.send_and_expect(self.pg0, [p_1k], self.pg1, n_rx=2)
+
+ # raise the interface's MTU
+ # should still use that of the path
+ self.vapi.sw_interface_set_mtu(self.pg1.sw_if_index,
+ [2000, 0, 0, 0])
+ self.send_and_expect(self.pg0, [p_2k], self.pg1, n_rx=3)
+ self.send_and_expect(self.pg0, [p_1k], self.pg1, n_rx=2)
+
+ # turn the tun_dst into an attached neighbour
+ route.modify([VppRoutePath("::",
+ self.pg1.sw_if_index)])
+ nbr2 = VppNeighbor(self,
+ self.pg1.sw_if_index,
+ self.pg1.remote_mac,
+ tun_dst).add_vpp_config()
+
+ self.send_and_expect(self.pg0, [p_2k], self.pg1, n_rx=3)
+ self.send_and_expect(self.pg0, [p_1k], self.pg1, n_rx=2)
+
+ # add back to not attached
+ nbr2.remove_vpp_config()
+ route.modify([VppRoutePath(self.pg1.remote_ip6,
+ self.pg1.sw_if_index)])
+
+ # set path high and interface low
+ pmtu.modify(2000)
+ self.vapi.sw_interface_set_mtu(self.pg1.sw_if_index,
+ [1300, 0, 0, 0])
+ self.send_and_expect(self.pg0, [p_2k], self.pg1, n_rx=3)
+ self.send_and_expect(self.pg0, [p_1k], self.pg1, n_rx=2)
+
+ # remove the path MTU
+ self.vapi.sw_interface_set_mtu(self.pg1.sw_if_index,
+ [2800, 0, 0, 0])
+ pmtu.remove_vpp_config()
+ self.send_and_expect(self.pg0, [p_2k], self.pg1, n_rx=2)
+ self.send_and_expect(self.pg0, [p_1k], self.pg1)
+
+
+class TestIPFibSource(VppTestCase):
+ """ IPv6 Table FibSource """
+
+ @classmethod
+ def setUpClass(cls):
+ super(TestIPFibSource, cls).setUpClass()
+
+ @classmethod
+ def tearDownClass(cls):
+ super(TestIPFibSource, cls).tearDownClass()
+
+ def setUp(self):
+ super(TestIPFibSource, self).setUp()
+
+ self.create_pg_interfaces(range(2))
+
+ for i in self.pg_interfaces:
+ i.admin_up()
+ i.config_ip6()
+ i.resolve_arp()
+ i.generate_remote_hosts(2)
+ i.configure_ipv6_neighbors()
+
+ def tearDown(self):
+ super(TestIPFibSource, self).tearDown()
+ for i in self.pg_interfaces:
+ i.admin_down()
+ i.unconfig_ip4()
+
+ def test_fib_source(self):
+ """ IP Table FibSource """
+
+ routes = self.vapi.ip_route_v2_dump(0, True)
+
+ # 2 interfaces (4 routes) + 2 specials + 4 neighbours = 10 routes
+ self.assertEqual(len(routes), 10)
+
+ # dump all the sources in the FIB
+ sources = self.vapi.fib_source_dump()
+ for source in sources:
+ if (source.src.name == "API"):
+ api_source = source.src
+ if (source.src.name == "interface"):
+ intf_source = source.src
+ if (source.src.name == "adjacency"):
+ adj_source = source.src
+ if (source.src.name == "special"):
+ special_source = source.src
+ if (source.src.name == "default-route"):
+ dr_source = source.src
+
+ # dump the individual route types
+ routes = self.vapi.ip_route_v2_dump(0, True, src=adj_source.id)
+ self.assertEqual(len(routes), 4)
+ routes = self.vapi.ip_route_v2_dump(0, True, src=intf_source.id)
+ self.assertEqual(len(routes), 4)
+ routes = self.vapi.ip_route_v2_dump(0, True, src=special_source.id)
+ self.assertEqual(len(routes), 1)
+ routes = self.vapi.ip_route_v2_dump(0, True, src=dr_source.id)
+ self.assertEqual(len(routes), 1)
+
+ # add a new soure that'a better than the API
+ self.vapi.fib_source_add(src={'name': "bgp",
+ "priority": api_source.priority - 1})
+
+ # dump all the sources to check our new one is there
+ sources = self.vapi.fib_source_dump()
+
+ for source in sources:
+ if (source.src.name == "bgp"):
+ bgp_source = source.src
+
+ self.assertTrue(bgp_source)
+ self.assertEqual(bgp_source.priority,
+ api_source.priority - 1)
+
+ # add a route with the default API source
+ r1 = VppIpRouteV2(
+ self, "2001::1", 128,
+ [VppRoutePath(self.pg0.remote_ip6,
+ self.pg0.sw_if_index)]).add_vpp_config()
+
+ r2 = VppIpRouteV2(self, "2001::1", 128,
+ [VppRoutePath(self.pg1.remote_ip6,
+ self.pg1.sw_if_index)],
+ src=bgp_source.id).add_vpp_config()
+
+ # ensure the BGP source takes priority
+ p = (Ether(src=self.pg0.remote_mac,
+ dst=self.pg0.local_mac) /
+ IPv6(src=self.pg0.remote_ip6, dst="2001::1") /
+ inet6.UDP(sport=1234, dport=1234) /
+ Raw(b'\xa5' * 100))
+
+ self.send_and_expect(self.pg0, [p], self.pg1)
+
+ r2.remove_vpp_config()
+ r1.remove_vpp_config()
+
+ self.assertFalse(find_route(self, "2001::1", 128))
+
+
+class TestIPxAF(VppTestCase):
+ """ IP cross AF """
+
+ @classmethod
+ def setUpClass(cls):
+ super(TestIPxAF, cls).setUpClass()
+
+ @classmethod
+ def tearDownClass(cls):
+ super(TestIPxAF, cls).tearDownClass()
+
+ def setUp(self):
+ super(TestIPxAF, self).setUp()
+
+ self.create_pg_interfaces(range(2))
+
+ for i in self.pg_interfaces:
+ i.admin_up()
+ i.config_ip6()
+ i.config_ip4()
+ i.resolve_arp()
+ i.resolve_ndp()
+
+ def tearDown(self):
+ super(TestIPxAF, self).tearDown()
+ for i in self.pg_interfaces:
+ i.admin_down()
+ i.unconfig_ip4()
+ i.unconfig_ip6()
+
+ def test_x_af(self):
+ """ Cross AF routing """
+
+ N_PKTS = 63
+ # a v4 route via a v6 attached next-hop
+ VppIpRoute(
+ self, "1.1.1.1", 32,
+ [VppRoutePath(self.pg1.remote_ip6,
+ self.pg1.sw_if_index)]).add_vpp_config()
+
+ p = (Ether(src=self.pg0.remote_mac,
+ dst=self.pg0.local_mac) /
+ IP(src=self.pg0.remote_ip4, dst="1.1.1.1") /
+ UDP(sport=1234, dport=1234) /
+ Raw(b'\xa5' * 100))
+ rxs = self.send_and_expect(self.pg0, p * N_PKTS, self.pg1)
+
+ for rx in rxs:
+ self.assertEqual(rx[IP].dst, "1.1.1.1")
+
+ # a v6 route via a v4 attached next-hop
+ VppIpRoute(
+ self, "2001::1", 128,
+ [VppRoutePath(self.pg1.remote_ip4,
+ self.pg1.sw_if_index)]).add_vpp_config()
+
+ p = (Ether(src=self.pg0.remote_mac,
+ dst=self.pg0.local_mac) /
+ IPv6(src=self.pg0.remote_ip6, dst="2001::1") /
+ UDP(sport=1234, dport=1234) /
+ Raw(b'\xa5' * 100))
+ rxs = self.send_and_expect(self.pg0, p * N_PKTS, self.pg1)
+
+ for rx in rxs:
+ self.assertEqual(rx[IPv6].dst, "2001::1")
+
+ # a recursive v4 route via a v6 next-hop (from above)
+ VppIpRoute(
+ self, "2.2.2.2", 32,
+ [VppRoutePath("2001::1",
+ 0xffffffff)]).add_vpp_config()
+
+ p = (Ether(src=self.pg0.remote_mac,
+ dst=self.pg0.local_mac) /
+ IP(src=self.pg0.remote_ip4, dst="2.2.2.2") /
+ UDP(sport=1234, dport=1234) /
+ Raw(b'\xa5' * 100))
+ rxs = self.send_and_expect(self.pg0, p * N_PKTS, self.pg1)
+
+ # a recursive v4 route via a v6 next-hop
+ VppIpRoute(
+ self, "2.2.2.3", 32,
+ [VppRoutePath(self.pg1.remote_ip6,
+ 0xffffffff)]).add_vpp_config()
+
+ p = (Ether(src=self.pg0.remote_mac,
+ dst=self.pg0.local_mac) /
+ IP(src=self.pg0.remote_ip4, dst="2.2.2.3") /
+ UDP(sport=1234, dport=1234) /
+ Raw(b'\xa5' * 100))
+ rxs = self.send_and_expect(self.pg0, p * N_PKTS, self.pg1)
+
+ # a recursive v6 route via a v4 next-hop
+ VppIpRoute(
+ self, "3001::1", 128,
+ [VppRoutePath(self.pg1.remote_ip4,
+ 0xffffffff)]).add_vpp_config()
+
+ p = (Ether(src=self.pg0.remote_mac,
+ dst=self.pg0.local_mac) /
+ IPv6(src=self.pg0.remote_ip6, dst="3001::1") /
+ UDP(sport=1234, dport=1234) /
+ Raw(b'\xa5' * 100))
+ rxs = self.send_and_expect(self.pg0, p * N_PKTS, self.pg1)
+
+ for rx in rxs:
+ self.assertEqual(rx[IPv6].dst, "3001::1")
+
+ VppIpRoute(
+ self, "3001::2", 128,
+ [VppRoutePath("1.1.1.1",
+ 0xffffffff)]).add_vpp_config()
+
+ p = (Ether(src=self.pg0.remote_mac,
+ dst=self.pg0.local_mac) /
+ IPv6(src=self.pg0.remote_ip6, dst="3001::2") /
+ UDP(sport=1234, dport=1234) /
+ Raw(b'\xa5' * 100))
+ rxs = self.send_and_expect(self.pg0, p * N_PKTS, self.pg1)
+
+ for rx in rxs:
+ self.assertEqual(rx[IPv6].dst, "3001::2")
+
+
+class TestIPv6Punt(VppTestCase):
+ """ IPv6 Punt Police/Redirect """
+
+ def setUp(self):
+ super(TestIPv6Punt, self).setUp()
+ self.create_pg_interfaces(range(4))
+
+ for i in self.pg_interfaces:
+ i.admin_up()
+ i.config_ip6()
+ i.resolve_ndp()
+
+ def tearDown(self):
+ super(TestIPv6Punt, self).tearDown()
+ for i in self.pg_interfaces:
+ i.unconfig_ip6()
+ i.admin_down()
+
+ def test_ip6_punt(self):
+ """ IPv6 punt police and redirect """
+
+ # use UDP packet that have a port we need to explicitly
+ # register to get punted.
+ pt_l4 = VppEnum.vl_api_punt_type_t.PUNT_API_TYPE_L4
+ af_ip6 = VppEnum.vl_api_address_family_t.ADDRESS_IP6
+ udp_proto = VppEnum.vl_api_ip_proto_t.IP_API_PROTO_UDP
+ punt_udp = {
+ 'type': pt_l4,
+ 'punt': {
+ 'l4': {
+ 'af': af_ip6,
+ 'protocol': udp_proto,
+ 'port': 7654,
+ }
+ }
+ }
+
+ self.vapi.set_punt(is_add=1, punt=punt_udp)
+
+ pkts = (Ether(src=self.pg0.remote_mac,
+ dst=self.pg0.local_mac) /
+ IPv6(src=self.pg0.remote_ip6, dst=self.pg0.local_ip6) /
+ UDP(sport=1234, dport=7654) /
+ Raw(b'\xa5' * 100)) * 1025
+
+ #
+ # Configure a punt redirect via pg1.
+ #
+ nh_addr = self.pg1.remote_ip6
+ 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 = 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)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+
+ #
+ # 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)
+
+ stats = policer.get_stats()
+
+ # Single rate policer - expect conform, violate but no exceed
+ self.assertGreater(stats['conform_packets'], 0)
+ self.assertEqual(stats['exceed_packets'], 0)
+ self.assertGreater(stats['violate_packets'], 0)
+
+ self.assertGreater(len(rx), 0)
+ self.assertLess(len(rx), len(pkts))
+
+ #
+ # remove the policer. back to full rx
+ #
+ 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.
+ #
+ 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
+ #
+ 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)
+ ip_punt_redirect.remove_vpp_config()
+
+ def test_ip6_punt_dump(self):
+ """ IPv6 punt redirect dump"""
+
+ #
+ # Configure a punt redirects
+ #
+ 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, "::")
+ ipr_03.add_vpp_config()
+ ipr_13.add_vpp_config()
+ ipr_23.add_vpp_config()
+
+ #
+ # Dump pg0 punt redirects
+ #
+ 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
+ #
+ punts = self.vapi.ip_punt_redirect_dump(sw_if_index=0xffffffff,
+ is_ipv6=True)
+ 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, self.pg3.remote_ip6)
+ self.assertEqual(str(punts[2].punt.nh), '::')