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_ip import VppIpPuntPolicer, VppIpPuntRedirect
from vpp_sub_interface import VppSubInterface, VppDot1QSubint, VppDot1ADSubint
from vpp_papi import VppEnum
from vpp_neighbor import VppNeighbor
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,
super(TestIPLoadBalance, self).tearDown()
def send_and_expect_load_balancing(self, input, pkts, outputs):
+ self.vapi.cli("clear trace")
input.add_stream(pkts)
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
for oo in outputs:
rx = oo._get_capture(1)
self.assertNotEqual(0, len(rx))
- for r in rx:
- rxs.append(r)
+ rxs.append(rx)
return rxs
def send_and_expect_one_itf(self, input, pkts, itf):
self.pg_start()
rx = itf.get_capture(len(pkts))
+ def total_len(self, rxs):
+ n = 0
+ for rx in rxs:
+ n += len(rx)
+ return n
+
def test_ip_load_balance(self):
""" IP Load-Balancing """
+ fhc = VppEnum.vl_api_ip_flow_hash_config_t
+ af = VppEnum.vl_api_address_family_t
+
#
# An array of packets that differ only in the destination port
#
# be guaranteed. But with 64 different packets we do expect some
# balancing. So instead just ensure there is traffic on each link.
#
- self.send_and_expect_load_balancing(self.pg0, port_ip_pkts,
- [self.pg1, self.pg2])
+ 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, src_mpls_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]))
#
# 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, sport=0, dport=0)
+ self.vapi.set_ip_flow_hash_v2(
+ af=af.ADDRESS_IP4,
+ table_id=0,
+ flow_hash_config=(fhc.IP_API_FLOW_HASH_SRC_IP |
+ fhc.IP_API_FLOW_HASH_DST_IP |
+ fhc.IP_API_FLOW_HASH_PROTO))
self.send_and_expect_load_balancing(self.pg0, src_ip_pkts,
[self.pg1, 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)
+ self.vapi.set_ip_flow_hash(vrf_id=0, src=1, dst=1,
+ proto=1, sport=1, dport=1)
#
# Recursive prefixes
self.pg3, self.pg4])
#
- # bring down pg1 expect LB to adjust to use only those that are pu
+ # bring down pg1 expect LB to adjust to use only those that are up
#
self.pg1.link_down()
rx = self.send_and_expect_load_balancing(self.pg0, src_pkts,
[self.pg2, self.pg3,
self.pg4])
- self.assertEqual(len(src_pkts), len(rx))
+ self.assertEqual(len(src_pkts), self.total_len(rx))
#
- # bring down pg2 expect LB to adjust to use only those that are pu
+ # bring down pg2 expect LB to adjust to use only those that are up
#
self.pg2.link_down()
rx = self.send_and_expect_load_balancing(self.pg0, src_pkts,
[self.pg3, self.pg4])
- self.assertEqual(len(src_pkts), len(rx))
+ self.assertEqual(len(src_pkts), self.total_len(rx))
#
# bring the links back up - expect LB over all again
rx = self.send_and_expect_load_balancing(self.pg0, src_pkts,
[self.pg1, self.pg2,
self.pg3, self.pg4])
- self.assertEqual(len(src_pkts), len(rx))
+ self.assertEqual(len(src_pkts), self.total_len(rx))
#
# The same link-up/down but this time admin state
self.pg2.admin_down()
rx = self.send_and_expect_load_balancing(self.pg0, src_pkts,
[self.pg3, self.pg4])
- self.assertEqual(len(src_pkts), len(rx))
+ self.assertEqual(len(src_pkts), self.total_len(rx))
self.pg1.admin_up()
self.pg2.admin_up()
self.pg1.resolve_arp()
rx = self.send_and_expect_load_balancing(self.pg0, src_pkts,
[self.pg1, self.pg2,
self.pg3, self.pg4])
- self.assertEqual(len(src_pkts), len(rx))
+ self.assertEqual(len(src_pkts), self.total_len(rx))
#
# Recursive prefixes
rx = self.send_and_expect_load_balancing(self.pg0, port_pkts,
[self.pg3, self.pg4])
- self.assertEqual(len(src_pkts), len(rx))
+ self.assertEqual(len(src_pkts), self.total_len(rx))
class TestIPVlan0(VppTestCase):
# Configure a punt redirect via pg1.
#
nh_addr = self.pg1.remote_ip4
- self.vapi.ip_punt_redirect(self.pg0.sw_if_index,
- self.pg1.sw_if_index,
- nh_addr)
+ 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)
#
policer = VppPolicer(self, "ip4-punt", 400, 0, 10, 0, rate_type=1)
policer.add_vpp_config()
- self.vapi.ip_punt_police(policer.policer_index)
+ ip_punt_policer = VppIpPuntPolicer(self, policer.policer_index)
+ ip_punt_policer.add_vpp_config()
self.vapi.cli("clear trace")
self.pg0.add_stream(pkts)
#
# remove the policer. back to full rx
#
- self.vapi.ip_punt_police(policer.policer_index, is_add=0)
+ 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.
#
- self.vapi.ip_punt_redirect(self.pg0.sw_if_index,
- self.pg1.sw_if_index,
- nh_addr,
- is_add=0)
+ 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
#
- self.vapi.ip_punt_redirect(0xffffffff,
- self.pg1.sw_if_index,
- nh_addr)
+ 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)
-
- self.vapi.ip_punt_redirect(0xffffffff,
- self.pg1.sw_if_index,
- nh_addr,
- is_add=0)
+ ip_punt_redirect.remove_vpp_config()
def test_ip_punt_dump(self):
""" IP4 punt redirect dump"""
# Configure a punt redirects
#
nh_address = self.pg3.remote_ip4
- self.vapi.ip_punt_redirect(self.pg0.sw_if_index,
- self.pg3.sw_if_index,
- nh_address)
- self.vapi.ip_punt_redirect(self.pg1.sw_if_index,
- self.pg3.sw_if_index,
- nh_address)
- self.vapi.ip_punt_redirect(self.pg2.sw_if_index,
- self.pg3.sw_if_index,
- '0.0.0.0')
+ 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.0.0")
+ ipr_03.add_vpp_config()
+ ipr_13.add_vpp_config()
+ ipr_23.add_vpp_config()
#
# Dump pg0 punt redirects
#
- punts = self.vapi.ip_punt_redirect_dump(self.pg0.sw_if_index)
- for p in punts:
- self.assertEqual(p.punt.rx_sw_if_index, self.pg0.sw_if_index)
+ 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
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})
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()
+ r8 = VppIpRoute(self, "127.0.0.0", 8,
+ [VppRoutePath("127.0.0.1",
+ lo.sw_if_index)]).add_vpp_config()
+ r8.remove_vpp_config()
+ r8.add_vpp_config()
+ r8.remove_vpp_config()
# remove the default route
r.remove_vpp_config()
+ # remove the interface prefix
+ a.remove_vpp_config()
+
class TestIP4Replace(VppTestCase):
""" IPv4 Interface Address Replace """