from framework import VppTestCase, VppTestRunner
from util import ppp
from vpp_ip_route import VppIpRoute, VppRoutePath, VppIpMRoute, \
- VppMRoutePath, MRouteItfFlags, MRouteEntryFlags, VppMplsIpBind, \
+ VppMRoutePath, VppMplsIpBind, \
VppMplsTable, VppIpTable, FibPathType, find_route, \
VppIpInterfaceAddress, find_route_in_dump, find_mroute_in_dump
from vpp_sub_interface import VppSubInterface, VppDot1QSubint, VppDot1ADSubint
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 """
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 """
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
self,
"0.0.0.0",
"232.1.1.1", 32,
- 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_232_1_1_1.add_vpp_config()
pu = (Ether(src=self.pg1.remote_mac,
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 = [[], [], [], []]
multi = VppIpMRoute(
self, "0.0.0.0",
"239.0.0.%d" % jj, 32,
- 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),
VppMRoutePath(self.pg1.sw_if_index,
- MRouteItfFlags.MFIB_ITF_FLAG_FORWARD),
+ MRouteItfFlags.MFIB_API_ITF_FLAG_FORWARD),
VppMRoutePath(self.pg2.sw_if_index,
- MRouteItfFlags.MFIB_ITF_FLAG_FORWARD),
+ MRouteItfFlags.MFIB_API_ITF_FLAG_FORWARD),
VppMRoutePath(self.pg3.sw_if_index,
- MRouteItfFlags.MFIB_ITF_FLAG_FORWARD)],
+ MRouteItfFlags.MFIB_API_ITF_FLAG_FORWARD)],
table_id=t.table_id).add_vpp_config()
routes[ii].append({'uni': uni,
'multi': multi})
# 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__':
unittest.main(testRunner=VppTestRunner)
Code Review / vpp.git / blobdiff