X-Git-Url: https://gerrit.fd.io/r/gitweb?a=blobdiff_plain;f=test%2Ftest_ip6.py;h=c86cb777aedbeb55620dc399efbd79d9cb359a91;hb=e63a2d44d16774a88763c5f6368a3f7210c64ddc;hp=dd29041c8987c37bff0205f77a2ac5587970d56a;hpb=5d280d5b51ace57f73ac1b43caf6c37c6ae11529;p=vpp.git diff --git a/test/test_ip6.py b/test/test_ip6.py index dd29041c898..c86cb777aed 100644 --- a/test/test_ip6.py +++ b/test/test_ip6.py @@ -676,6 +676,8 @@ class TestIPv6(TestIPv6ND): Test scenario: """ + self.pg0.ip6_ra_config(no=1, suppress=1) + # # Before we begin change the IPv6 RA responses to use the unicast # address - that way we will not confuse them with the periodic @@ -771,7 +773,7 @@ class TestIPv6(TestIPv6ND): # If we happen to pick up the periodic RA at this point then so be it, # it's not an error. # - self.pg0.ip6_ra_config(send_unicast=1, suppress=0) + self.pg0.ip6_ra_config(send_unicast=1) p = (Ether(dst=dmac, src=self.pg0.remote_mac) / IPv6(dst="ff02::2", src="::") / ICMPv6ND_RS()) @@ -979,7 +981,8 @@ class TestIPv6(TestIPv6ND): # # Reset the periodic advertisements back to default values # - self.pg0.ip6_ra_config(no=1, suppress=1, send_unicast=0) + self.pg0.ip6_ra_config(suppress=1) + self.pg0.ip6_ra_config(no=1, send_unicast=1) def test_mld(self): """ MLD Report """ @@ -1754,19 +1757,19 @@ class IPv6NDProxyTest(TestIPv6ND): self.assertTrue(rx[0].haslayer(ICMPv6ND_NS)) -class TestIPNull(VppTestCase): +class TestIP6Null(VppTestCase): """ IPv6 routes via NULL """ @classmethod def setUpClass(cls): - super(TestIPNull, cls).setUpClass() + super(TestIP6Null, cls).setUpClass() @classmethod def tearDownClass(cls): - super(TestIPNull, cls).tearDownClass() + super(TestIP6Null, cls).tearDownClass() def setUp(self): - super(TestIPNull, self).setUp() + super(TestIP6Null, self).setUp() # create 2 pg interfaces self.create_pg_interfaces(range(1)) @@ -1777,7 +1780,7 @@ class TestIPNull(VppTestCase): i.resolve_ndp() def tearDown(self): - super(TestIPNull, self).tearDown() + super(TestIP6Null, self).tearDown() for i in self.pg_interfaces: i.unconfig_ip6() i.admin_down() @@ -1835,19 +1838,19 @@ class TestIPNull(VppTestCase): self.assertEqual(icmp.code, 1) -class TestIPDisabled(VppTestCase): +class TestIP6Disabled(VppTestCase): """ IPv6 disabled """ @classmethod def setUpClass(cls): - super(TestIPDisabled, cls).setUpClass() + super(TestIP6Disabled, cls).setUpClass() @classmethod def tearDownClass(cls): - super(TestIPDisabled, cls).tearDownClass() + super(TestIP6Disabled, cls).tearDownClass() def setUp(self): - super(TestIPDisabled, self).setUp() + super(TestIP6Disabled, self).setUp() # create 2 pg interfaces self.create_pg_interfaces(range(2)) @@ -1861,7 +1864,7 @@ class TestIPDisabled(VppTestCase): self.pg1.admin_up() def tearDown(self): - super(TestIPDisabled, self).tearDown() + super(TestIP6Disabled, self).tearDown() for i in self.pg_interfaces: i.unconfig_ip4() i.admin_down() @@ -1965,25 +1968,6 @@ class TestIP6LoadBalance(VppTestCase): 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 """ @@ -2107,7 +2091,7 @@ class TestIP6LoadBalance(VppTestCase): # 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) + self.send_and_expect_only(self.pg0, port_ent_pkts, self.pg1) # # change the flow hash config so it's only IP src,dst @@ -2121,7 +2105,7 @@ class TestIP6LoadBalance(VppTestCase): [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) + self.send_and_expect_only(self.pg0, port_ip_pkts, self.pg2) # # change the flow hash config back to defaults @@ -2206,7 +2190,7 @@ class TestIP6LoadBalance(VppTestCase): # 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) + self.send_and_expect_only(self.pg0, port_pkts, self.pg3) class IP6PuntSetup(object): @@ -2459,19 +2443,19 @@ class TestIP6PuntHandoff(IP6PuntSetup, VppTestCase): ip_punt_redirect.remove_vpp_config() -class TestIPDeag(VppTestCase): +class TestIP6Deag(VppTestCase): """ IPv6 Deaggregate Routes """ @classmethod def setUpClass(cls): - super(TestIPDeag, cls).setUpClass() + super(TestIP6Deag, cls).setUpClass() @classmethod def tearDownClass(cls): - super(TestIPDeag, cls).tearDownClass() + super(TestIP6Deag, cls).tearDownClass() def setUp(self): - super(TestIPDeag, self).setUp() + super(TestIP6Deag, self).setUp() self.create_pg_interfaces(range(3)) @@ -2481,7 +2465,7 @@ class TestIPDeag(VppTestCase): i.resolve_ndp() def tearDown(self): - super(TestIPDeag, self).tearDown() + super(TestIP6Deag, self).tearDown() for i in self.pg_interfaces: i.unconfig_ip6() i.admin_down() @@ -2618,12 +2602,14 @@ class TestIP6Input(VppTestCase): 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] + rxs = self.send_and_expect_some(self.pg0, + p_version * NUM_PKTS, + self.pg0) - # 0: "hop limit exceeded in transit", - self.assertEqual((icmp.type, icmp.code), (3, 0)) + for rx in rxs: + icmp = rx[ICMPv6TimeExceeded] + # 0: "hop limit exceeded in transit", + self.assertEqual((icmp.type, icmp.code), (3, 0)) icmpv6_data = '\x0a' * 18 all_0s = "::" @@ -2676,19 +2662,19 @@ class TestIP6Input(VppTestCase): self.pg_start() -class TestIPReplace(VppTestCase): +class TestIP6Replace(VppTestCase): """ IPv6 Table Replace """ @classmethod def setUpClass(cls): - super(TestIPReplace, cls).setUpClass() + super(TestIP6Replace, cls).setUpClass() @classmethod def tearDownClass(cls): - super(TestIPReplace, cls).tearDownClass() + super(TestIP6Replace, cls).tearDownClass() def setUp(self): - super(TestIPReplace, self).setUp() + super(TestIP6Replace, self).setUp() self.create_pg_interfaces(range(4)) @@ -2704,7 +2690,7 @@ class TestIPReplace(VppTestCase): table_id += 1 def tearDown(self): - super(TestIPReplace, self).tearDown() + super(TestIP6Replace, self).tearDown() for i in self.pg_interfaces: i.admin_down() i.unconfig_ip6() @@ -2817,19 +2803,19 @@ class TestIPReplace(VppTestCase): self.assertEqual(len(t.mdump()), 5) -class TestIP6Replace(VppTestCase): - """ IPv4 Interface Address Replace """ +class TestIP6AddrReplace(VppTestCase): + """ IPv6 Interface Address Replace """ @classmethod def setUpClass(cls): - super(TestIP6Replace, cls).setUpClass() + super(TestIP6AddrReplace, cls).setUpClass() @classmethod def tearDownClass(cls): - super(TestIP6Replace, cls).tearDownClass() + super(TestIP6AddrReplace, cls).tearDownClass() def setUp(self): - super(TestIP6Replace, self).setUp() + super(TestIP6AddrReplace, self).setUp() self.create_pg_interfaces(range(4)) @@ -2837,7 +2823,7 @@ class TestIP6Replace(VppTestCase): i.admin_up() def tearDown(self): - super(TestIP6Replace, self).tearDown() + super(TestIP6AddrReplace, self).tearDown() for i in self.pg_interfaces: i.admin_down() @@ -3187,6 +3173,12 @@ class TestIPv6PathMTU(VppTestCase): self.vapi.sw_interface_set_mtu(self.pg1.sw_if_index, [2800, 0, 0, 0]) + p_6k = (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' * 2000)) p_2k = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) / IPv6(src=self.pg0.remote_ip6, @@ -3206,6 +3198,7 @@ class TestIPv6PathMTU(VppTestCase): self.pg1.remote_ip6).add_vpp_config() # this is now the interface MTU frags + self.send_and_expect(self.pg0, [p_6k], self.pg1, n_rx=4) self.send_and_expect(self.pg0, [p_2k], self.pg1, n_rx=2) self.send_and_expect(self.pg0, [p_1k], self.pg1) @@ -3597,5 +3590,149 @@ class TestIPxAF(VppTestCase): 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), '::') + + if __name__ == '__main__': unittest.main(testRunner=VppTestRunner)