X-Git-Url: https://gerrit.fd.io/r/gitweb?a=blobdiff_plain;f=test%2Ftest_ip6.py;h=f4b9ef8c647914bd1898e6756a3926dd9786c658;hb=282872127;hp=36532cedf128ecde78ae0a5911dd2c4445782558;hpb=efd7bc2b1c8db160933ed3e9ab3cde0d07aaf863;p=vpp.git diff --git a/test/test_ip6.py b/test/test_ip6.py index 36532cedf12..f4b9ef8c647 100644 --- a/test/test_ip6.py +++ b/test/test_ip6.py @@ -1,6 +1,6 @@ #!/usr/bin/env python3 -import socket +from socket import inet_pton, inet_ntop import unittest from parameterized import parameterized @@ -10,10 +10,10 @@ from scapy.contrib.mpls import MPLS from scapy.layers.inet6 import IPv6, ICMPv6ND_NS, ICMPv6ND_RS, \ ICMPv6ND_RA, ICMPv6NDOptMTU, ICMPv6NDOptSrcLLAddr, ICMPv6NDOptPrefixInfo, \ ICMPv6ND_NA, ICMPv6NDOptDstLLAddr, ICMPv6DestUnreach, icmp6types, \ - ICMPv6TimeExceeded, ICMPv6EchoRequest, ICMPv6EchoReply, IPv6ExtHdrHopByHop + ICMPv6TimeExceeded, ICMPv6EchoRequest, ICMPv6EchoReply, \ + IPv6ExtHdrHopByHop, ICMPv6MLReport2, ICMPv6MLDMultAddrRec from scapy.layers.l2 import Ether, Dot1Q from scapy.packet import Raw -from scapy.utils import inet_pton, inet_ntop from scapy.utils6 import in6_getnsma, in6_getnsmac, in6_ptop, in6_islladdr, \ in6_mactoifaceid from six import moves @@ -23,8 +23,8 @@ from util import ppp, ip6_normalize, mk_ll_addr from vpp_ip import DpoProto from vpp_ip_route import VppIpRoute, VppRoutePath, find_route, VppIpMRoute, \ VppMRoutePath, MRouteItfFlags, MRouteEntryFlags, VppMplsIpBind, \ - VppMplsRoute, VppMplsTable, VppIpTable, FibPathType, \ - VppIpInterfaceAddress + VppMplsRoute, VppMplsTable, VppIpTable, FibPathType, FibPathProto, \ + VppIpInterfaceAddress, find_route_in_dump, find_mroute_in_dump from vpp_neighbor import find_nbr, VppNeighbor from vpp_pg_interface import is_ipv6_misc from vpp_sub_interface import VppSubInterface, VppDot1QSubint @@ -138,6 +138,7 @@ class TestIPv6ND(VppTestCase): def send_and_expect_ns(self, tx_intf, rx_intf, pkts, tgt_ip, filter_out_fn=is_ipv6_misc): + self.vapi.cli("clear trace") tx_intf.add_stream(pkts) self.pg_enable_capture(self.pg_interfaces) self.pg_start() @@ -222,7 +223,6 @@ class TestIPv6(TestIPv6ND): """Run standard test teardown and log ``show ip6 neighbors``.""" for i in self.interfaces: i.unconfig_ip6() - i.ip6_disable() i.admin_down() for i in self.sub_interfaces: i.remove_vpp_config() @@ -578,9 +578,12 @@ class TestIPv6(TestIPv6ND): self.pg0.remote_ip6, self.pg1.remote_hosts[1].ip6) - def validate_ra(self, intf, rx, dst_ip=None, mtu=9000, pi_opt=None): + def validate_ra(self, intf, rx, dst_ip=None, src_ip=None, + mtu=9000, pi_opt=None): if not dst_ip: dst_ip = intf.remote_ip6 + if not src_ip: + src_ip = mk_ll_addr(intf.local_mac) # unicasted packets must come to the unicast mac self.assertEqual(rx[Ether].dst, intf.remote_mac) @@ -595,8 +598,7 @@ class TestIPv6(TestIPv6ND): # and come from the router's link local self.assertTrue(in6_islladdr(rx[IPv6].src)) - self.assertEqual(in6_ptop(rx[IPv6].src), - in6_ptop(mk_ll_addr(intf.local_mac))) + self.assertEqual(in6_ptop(rx[IPv6].src), in6_ptop(src_ip)) # it should contain the links MTU ra = rx[ICMPv6ND_RA] @@ -635,7 +637,9 @@ class TestIPv6(TestIPv6ND): def send_and_expect_ra(self, intf, pkts, remark, dst_ip=None, filter_out_fn=is_ipv6_misc, - opt=None): + opt=None, + src_ip=None): + self.vapi.cli("clear trace") intf.add_stream(pkts) self.pg_enable_capture(self.pg_interfaces) self.pg_start() @@ -643,7 +647,7 @@ class TestIPv6(TestIPv6ND): self.assertEqual(len(rx), 1) rx = rx[0] - self.validate_ra(intf, rx, dst_ip, pi_opt=opt) + self.validate_ra(intf, rx, dst_ip, src_ip=src_ip, pi_opt=opt) def test_rs(self): """ IPv6 Router Solicitation Exceptions @@ -666,8 +670,7 @@ class TestIPv6(TestIPv6ND): # - expect an RA in return # p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) / - IPv6( - dst=self.pg0.local_ip6, src=self.pg0.remote_ip6) / + IPv6(dst=self.pg0.local_ip6, src=self.pg0.remote_ip6) / ICMPv6ND_RS()) pkts = [p] self.send_and_expect_ra(self.pg0, pkts, "Genuine RS") @@ -925,16 +928,53 @@ class TestIPv6(TestIPv6ND): self.pg1.local_ip6_prefix_len), is_no=1) + # + # change the link's link local, so we know that works too. + # + self.vapi.sw_interface_ip6_set_link_local_address( + sw_if_index=self.pg0.sw_if_index, + ip="fe80::88") + self.pg0.ip6_ra_config(send_unicast=1) self.send_and_expect_ra(self.pg0, p, "RA with Prefix reverted to defaults", - dst_ip=ll) + dst_ip=ll, + src_ip="fe80::88") # # Reset the periodic advertisements back to default values # self.pg0.ip6_ra_config(no=1, suppress=1, send_unicast=0) + def test_mld(self): + """ MLD Report """ + # + # test one MLD is sent after applying an IPv6 Address on an interface + # + self.pg_enable_capture(self.pg_interfaces) + self.pg_start() + + subitf = VppDot1QSubint(self, self.pg1, 99) + + subitf.admin_up() + subitf.config_ip6() + + rxs = self.pg1._get_capture(timeout=2, filter_out_fn=None) + + # + # hunt for the MLD on vlan 99 + # + for rx in rxs: + # make sure ipv6 packets with hop by hop options have + # correct checksums + self.assert_packet_checksums_valid(rx) + if rx.haslayer(IPv6ExtHdrHopByHop) and \ + rx.haslayer(Dot1Q) and \ + rx[Dot1Q].vlan == 99: + mld = rx[ICMPv6MLReport2] + + self.assertEqual(mld.records_number, 4) + class TestIPv6IfAddrRoute(VppTestCase): """ IPv6 Interface Addr Route Test Case """ @@ -1035,7 +1075,6 @@ class TestICMPv6Echo(VppTestCase): super(TestICMPv6Echo, self).tearDown() for i in self.pg_interfaces: i.unconfig_ip6() - i.ip6_disable() i.admin_down() def test_icmpv6_echo(self): @@ -1159,7 +1198,7 @@ class TestIPv6RD(TestIPv6ND): def test_rd_receive_router_advertisement(self): """ Verify events triggered by received RA packets """ - self.vapi.want_ip6_ra_events() + self.vapi.want_ip6_ra_events(enable=1) prefix_info_1 = ICMPv6NDOptPrefixInfo( prefix="1::2", @@ -1265,6 +1304,17 @@ class TestIPv6RDControlPlane(TestIPv6ND): list.append(str(entry.route.prefix.network_address)) return list + def wait_for_no_default_route(self, n_tries=50, s_time=1): + while (n_tries): + fib = self.vapi.ip_route_dump(0, True) + default_routes = self.get_default_routes(fib) + if 0 == len(default_routes): + return True + n_tries = n_tries - 1 + self.sleep(s_time) + + return False + def test_all(self): """ Test handling of SLAAC addresses and default routes """ @@ -1364,9 +1414,7 @@ class TestIPv6RDControlPlane(TestIPv6ND): self.sleep_on_vpp_time(1) # check that default route is deleted - fib = self.vapi.ip_route_dump(0, True) - default_routes = self.get_default_routes(fib) - self.assertEqual(len(default_routes), 0) + self.assertTrue(self.wait_for_no_default_route()) # check FIB still contains the SLAAC address addresses = set(self.get_interface_addresses(fib, self.pg0)) @@ -1443,7 +1491,7 @@ class IPv6NDProxyTest(TestIPv6ND): # Add proxy support for the host # self.vapi.ip6nd_proxy_add_del( - ip=inet_pton(AF_INET6, self.pg0._remote_hosts[2].ip6), + is_add=1, ip=inet_pton(AF_INET6, self.pg0._remote_hosts[2].ip6), sw_if_index=self.pg1.sw_if_index) # @@ -1510,7 +1558,7 @@ class IPv6NDProxyTest(TestIPv6ND): lladdr=self.pg0._remote_hosts[2].mac)) self.vapi.ip6nd_proxy_add_del( - ip=inet_pton(AF_INET6, self.pg0._remote_hosts[3].ip6), + is_add=1, ip=inet_pton(AF_INET6, self.pg0._remote_hosts[3].ip6), sw_if_index=self.pg2.sw_if_index) self.send_and_expect_na(self.pg2, ns_pg2, @@ -1551,10 +1599,10 @@ class IPv6NDProxyTest(TestIPv6ND): # self.vapi.ip6nd_proxy_add_del( ip=inet_pton(AF_INET6, self.pg0._remote_hosts[2].ip6), - sw_if_index=self.pg1.sw_if_index, is_del=1) + sw_if_index=self.pg1.sw_if_index, is_add=0) self.vapi.ip6nd_proxy_add_del( ip=inet_pton(AF_INET6, self.pg0._remote_hosts[3].ip6), - sw_if_index=self.pg2.sw_if_index, is_del=1) + sw_if_index=self.pg2.sw_if_index, is_add=0) self.assertFalse(find_nbr(self, self.pg2.sw_if_index, @@ -2370,5 +2418,146 @@ class TestIP6Input(VppTestCase): self.pg_enable_capture(self.pg_interfaces) self.pg_start() + +class TestIPReplace(VppTestCase): + """ IPv6 Table Replace """ + + @classmethod + def setUpClass(cls): + super(TestIPReplace, cls).setUpClass() + + @classmethod + def tearDownClass(cls): + super(TestIPReplace, cls).tearDownClass() + + def setUp(self): + super(TestIPReplace, self).setUp() + + self.create_pg_interfaces(range(4)) + + table_id = 1 + self.tables = [] + + for i in self.pg_interfaces: + i.admin_up() + i.config_ip6() + i.resolve_arp() + 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_ip4() + + def test_replace(self): + """ IP Table Replace """ + + 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_ENTRY_FLAG_NONE, + [VppMRoutePath(self.pg0.sw_if_index, + MRouteItfFlags.MFIB_ITF_FLAG_ACCEPT, + proto=FibPathProto.FIB_PATH_NH_PROTO_IP6), + VppMRoutePath(self.pg1.sw_if_index, + MRouteItfFlags.MFIB_ITF_FLAG_FORWARD, + proto=FibPathProto.FIB_PATH_NH_PROTO_IP6), + VppMRoutePath(self.pg2.sw_if_index, + MRouteItfFlags.MFIB_ITF_FLAG_FORWARD, + proto=FibPathProto.FIB_PATH_NH_PROTO_IP6), + VppMRoutePath(self.pg3.sw_if_index, + MRouteItfFlags.MFIB_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) + + if __name__ == '__main__': unittest.main(testRunner=VppTestRunner)