X-Git-Url: https://gerrit.fd.io/r/gitweb?a=blobdiff_plain;f=test%2Ftest_ip4.py;h=8c6f8c04c0cbd05ec035e911ddd90d3f3dc07f19;hb=61ae056bd;hp=7300679ff0d997f08af09007aa84fe55572a210d;hpb=9db6ada779794779158163f6293b479ae7f6ad5e;p=vpp.git diff --git a/test/test_ip4.py b/test/test_ip4.py index 7300679ff0d..8c6f8c04c0c 100644 --- a/test/test_ip4.py +++ b/test/test_ip4.py @@ -20,6 +20,8 @@ from vpp_ip_route import VppIpRoute, VppRoutePath, VppIpMRoute, \ from vpp_sub_interface import VppSubInterface, VppDot1QSubint, VppDot1ADSubint from vpp_papi import VppEnum from vpp_neighbor import VppNeighbor +from vpp_lo_interface import VppLoInterface +from vpp_policer import VppPolicer NUM_PKTS = 67 @@ -88,7 +90,7 @@ class TestIPv4(VppTestCase): super(TestIPv4, self).tearDown() def show_commands_at_teardown(self): - self.logger.info(self.vapi.cli("show ip arp")) + self.logger.info(self.vapi.cli("show ip4 neighbors")) # info(self.vapi.cli("show ip fib")) # many entries def modify_packet(self, src_if, packet_size, pkt): @@ -212,6 +214,80 @@ class TestIPv4(VppTestCase): self.verify_capture(i, pkts) +class TestIPv4RouteLookup(VppTestCase): + """ IPv4 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(TestIPv4RouteLookup, cls).setUpClass() + + @classmethod + def tearDownClass(cls): + super(TestIPv4RouteLookup, cls).tearDownClass() + + def setUp(self): + super(TestIPv4RouteLookup, self).setUp() + + drop_nh = VppRoutePath("127.0.0.1", 0xffffffff, + type=FibPathType.FIB_PATH_TYPE_DROP) + + # Add 3 routes + r = VppIpRoute(self, "1.1.0.0", 16, [drop_nh]) + r.add_vpp_config() + self.routes.append(r) + + r = VppIpRoute(self, "1.1.1.0", 24, [drop_nh]) + r.add_vpp_config() + self.routes.append(r) + + r = VppIpRoute(self, "1.1.1.1", 32, [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(TestIPv4RouteLookup, self).tearDown() + + def test_exact_match(self): + # Verify we find the host route + prefix = "1.1.1.1/32" + result = self.route_lookup(prefix, True) + assert (prefix == str(result.route.prefix)) + + # Verify we find a middle prefix route + prefix = "1.1.1.0/24" + 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("1.1.1.2/32", True) + + def test_longest_prefix_match(self): + # verify we find lpm + lpm_prefix = "1.1.1.0/24" + result = self.route_lookup("1.1.1.2/32", 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 TestIPv4IfAddrRoute(VppTestCase): """ IPv4 Interface Addr Route Test Case """ @@ -335,6 +411,24 @@ class TestIPv4IfAddrRoute(VppTestCase): fib4_dump = self.vapi.ip_route_dump(0) self.assertTrue(lo_if.is_ip4_entry_in_fib_dump(fib4_dump)) + def test_ipv4_ifaddr_del(self): + """ Delete an interface address that does not exist """ + + loopbacks = self.create_loopback_interfaces(1) + lo = self.lo_interfaces[0] + + lo.config_ip4() + 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_ip4_prefix, + is_add=0) + class TestICMPEcho(VppTestCase): """ ICMP Echo Test Case """ @@ -1389,8 +1483,8 @@ class TestIPPunt(VppTestCase): # # add a policer # - policer = self.vapi.policer_add_del(b"ip4-punt", 400, 0, 10, 0, - rate_type=1) + policer = VppPolicer(self, "ip4-punt", 400, 0, 10, 0, rate_type=1) + policer.add_vpp_config() self.vapi.ip_punt_police(policer.policer_index) self.vapi.cli("clear trace") @@ -1410,8 +1504,7 @@ class TestIPPunt(VppTestCase): # remove the policer. back to full rx # self.vapi.ip_punt_police(policer.policer_index, is_add=0) - self.vapi.policer_add_del(b"ip4-punt", 400, 0, 10, 0, - rate_type=1, is_add=0) + policer.remove_vpp_config() self.send_and_expect(self.pg0, pkts, self.pg1) # @@ -2088,7 +2181,232 @@ class TestIPReplace(VppTestCase): for t in self.tables: t.flush() self.assertEqual(len(t.dump()), 5) - self.assertEqual(len(t.mdump()), 1) + self.assertEqual(len(t.mdump()), 3) + + +class TestIPCover(VppTestCase): + """ IPv4 Table Cover """ + + @classmethod + def setUpClass(cls): + super(TestIPCover, cls).setUpClass() + + @classmethod + def tearDownClass(cls): + super(TestIPCover, cls).tearDownClass() + + def setUp(self): + super(TestIPCover, self).setUp() + + self.create_pg_interfaces(range(4)) + + table_id = 1 + self.tables = [] + + for i in self.pg_interfaces: + i.admin_up() + i.config_ip4() + i.resolve_arp() + i.generate_remote_hosts(2) + self.tables.append(VppIpTable(self, table_id).add_vpp_config()) + table_id += 1 + + def tearDown(self): + super(TestIPCover, self).tearDown() + for i in self.pg_interfaces: + i.admin_down() + i.unconfig_ip4() + + def test_cover(self): + """ IP Table Cover """ + + # add a loop back with a /32 prefix + lo = VppLoInterface(self) + lo.admin_up() + a = VppIpInterfaceAddress(self, lo, "127.0.0.1", 32).add_vpp_config() + + # add a neighbour that matches the loopback's /32 + nbr = VppNeighbor(self, + lo.sw_if_index, + lo.remote_mac, + "127.0.0.1").add_vpp_config() + + # add the default route which will be the cover for /32 + r = VppIpRoute(self, "0.0.0.0", 0, + [VppRoutePath("127.0.0.1", + lo.sw_if_index)], + register=False).add_vpp_config() + + # add/remove/add a longer mask cover + r = VppIpRoute(self, "127.0.0.0", 8, + [VppRoutePath("127.0.0.1", + lo.sw_if_index)]).add_vpp_config() + r.remove_vpp_config() + r.add_vpp_config() + + # remove the default route + r.remove_vpp_config() + + +class TestIP4Replace(VppTestCase): + """ IPv4 Interface Address Replace """ + + @classmethod + def setUpClass(cls): + super(TestIP4Replace, cls).setUpClass() + + @classmethod + def tearDownClass(cls): + super(TestIP4Replace, cls).tearDownClass() + + def setUp(self): + super(TestIP4Replace, self).setUp() + + self.create_pg_interfaces(range(4)) + + for i in self.pg_interfaces: + i.admin_up() + + def tearDown(self): + super(TestIP4Replace, 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)) + + 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] + + # 172.16.x.1/24 + addr = "172.16.%d.1" % intf.sw_if_index + a = VppIpInterfaceAddress(self, intf, addr, 24).add_vpp_config() + intf_pfxs[i].append(a) + + # 172.16.x.2/24 - a different address in the same subnet as above + addr = "172.16.%d.2" % intf.sw_if_index + a = VppIpInterfaceAddress(self, intf, addr, 24).add_vpp_config() + intf_pfxs[i].append(a) + + # 172.15.x.2/24 - a different address and subnet + addr = "172.15.%d.2" % intf.sw_if_index + a = VppIpInterfaceAddress(self, intf, addr, 24).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: + # 172.18.x.1/24 + addr = "172.18.%d.1" % intf.sw_if_index + pfxs.append(VppIpInterfaceAddress(self, intf, addr, + 24).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: + # 172.18.x.1/24 + addr = "172.18.%d.1" % intf.sw_if_index + VppIpInterfaceAddress(self, intf, addr, 24).add_vpp_config() + + self.vapi.sw_interface_address_replace_begin() + + pfxs = [] + for intf in self.pg_interfaces: + # 172.18.x.1/24 + addr = "172.18.%d.1" % (intf.sw_if_index + 1) + pfxs.append(VppIpInterfaceAddress(self, intf, + addr, 24).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()) if __name__ == '__main__':