from parameterized import parameterized
import scapy.compat
import scapy.layers.inet6 as inet6
+from scapy.layers.inet import UDP
from scapy.contrib.mpls import MPLS
from scapy.layers.inet6 import IPv6, ICMPv6ND_NS, ICMPv6ND_RS, \
ICMPv6ND_RA, ICMPv6NDOptMTU, ICMPv6NDOptSrcLLAddr, ICMPv6NDOptPrefixInfo, \
in6_mactoifaceid
from six import moves
-from framework import VppTestCase, VppTestRunner
+from framework import VppTestCase, VppTestRunner, tag_run_solo
from util import ppp, ip6_normalize, mk_ll_addr
-from vpp_ip import DpoProto
+from vpp_papi import VppEnum
+from vpp_ip import DpoProto, VppIpPuntPolicer, VppIpPuntRedirect, VppIpPathMtu
from vpp_ip_route import VppIpRoute, VppRoutePath, find_route, VppIpMRoute, \
- VppMRoutePath, MRouteItfFlags, MRouteEntryFlags, VppMplsIpBind, \
+ VppMRoutePath, VppMplsIpBind, \
VppMplsRoute, VppMplsTable, VppIpTable, FibPathType, FibPathProto, \
VppIpInterfaceAddress, find_route_in_dump, find_mroute_in_dump, \
VppIp6LinkLocalAddress
from vpp_neighbor import find_nbr, VppNeighbor
+from vpp_ipip_tun_interface import VppIpIpTunInterface
from vpp_pg_interface import is_ipv6_misc
from vpp_sub_interface import VppSubInterface, VppDot1QSubint
-from vpp_policer import VppPolicer
+from vpp_policer import VppPolicer, PolicerAction
from ipaddress import IPv6Network, IPv6Address
AF_INET6 = socket.AF_INET6
self.assertEqual(ip.dst, dip)
+@tag_run_solo
class TestIPv6(TestIPv6ND):
""" IPv6 Test Case """
for i in self.pg_interfaces:
i.admin_up()
i.config_ip6()
+ i.resolve_ndp(link_layer=True)
i.resolve_ndp()
def tearDown(self):
- Check outgoing ICMPv6 Echo Reply message on pg0 interface.
"""
- icmpv6_id = 0xb
- icmpv6_seq = 5
- icmpv6_data = b'\x0a' * 18
- p_echo_request = (Ether(src=self.pg0.remote_mac,
- dst=self.pg0.local_mac) /
- IPv6(src=self.pg0.remote_ip6,
- dst=self.pg0.local_ip6) /
- ICMPv6EchoRequest(
- id=icmpv6_id,
- seq=icmpv6_seq,
- data=icmpv6_data))
-
- self.pg0.add_stream(p_echo_request)
+ # test both with global and local ipv6 addresses
+ dsts = (self.pg0.local_ip6, self.pg0.local_ip6_ll)
+ id = 0xb
+ seq = 5
+ data = b'\x0a' * 18
+ p = list()
+ for dst in dsts:
+ p.append((Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
+ IPv6(src=self.pg0.remote_ip6, dst=dst) /
+ ICMPv6EchoRequest(id=id, seq=seq, data=data)))
+
+ self.pg0.add_stream(p)
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
-
- rx = self.pg0.get_capture(1)
- rx = rx[0]
- ether = rx[Ether]
- ipv6 = rx[IPv6]
- icmpv6 = rx[ICMPv6EchoReply]
-
- self.assertEqual(ether.src, self.pg0.local_mac)
- self.assertEqual(ether.dst, self.pg0.remote_mac)
-
- self.assertEqual(ipv6.src, self.pg0.local_ip6)
- self.assertEqual(ipv6.dst, self.pg0.remote_ip6)
-
- self.assertEqual(
- icmp6types[icmpv6.type], "Echo Reply")
- self.assertEqual(icmpv6.id, icmpv6_id)
- self.assertEqual(icmpv6.seq, icmpv6_seq)
- self.assertEqual(icmpv6.data, icmpv6_data)
+ rxs = self.pg0.get_capture(len(dsts))
+
+ for rx, dst in zip(rxs, dsts):
+ ether = rx[Ether]
+ ipv6 = rx[IPv6]
+ icmpv6 = rx[ICMPv6EchoReply]
+ self.assertEqual(ether.src, self.pg0.local_mac)
+ self.assertEqual(ether.dst, self.pg0.remote_mac)
+ self.assertEqual(ipv6.src, dst)
+ self.assertEqual(ipv6.dst, self.pg0.remote_ip6)
+ self.assertEqual(icmp6types[icmpv6.type], "Echo Reply")
+ self.assertEqual(icmpv6.id, id)
+ self.assertEqual(icmpv6.seq, seq)
+ self.assertEqual(icmpv6.data, data)
class TestIPv6RD(TestIPv6ND):
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,
"::",
"ffef::1", 128,
- 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_ff_01.add_vpp_config()
pu = (Ether(src=self.pg1.remote_mac,
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)
# 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])
- self.send_and_expect_load_balancing(self.pg0, port_mpls_neos_pkts,
- [self.pg1, self.pg2])
+ rx = self.send_and_expect_load_balancing(self.pg0, port_mpls_neos_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]))
#
# The packets with Entropy label in should not load-balance,
# - 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,
- is_ipv6=1)
+ self.vapi.set_ip_flow_hash(vrf_id=0, src=1, dst=1, proto=1,
+ sport=0, dport=0, is_ipv6=1)
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,
- is_ipv6=1)
+ proto=1, is_ipv6=1)
#
# Recursive prefixes
self.send_and_expect_one_itf(self.pg0, port_pkts, self.pg3)
-class TestIP6Punt(VppTestCase):
- """ IPv6 Punt Police/Redirect """
-
- @classmethod
- def setUpClass(cls):
- super(TestIP6Punt, cls).setUpClass()
-
- @classmethod
- def tearDownClass(cls):
- super(TestIP6Punt, cls).tearDownClass()
-
- def setUp(self):
- super(TestIP6Punt, self).setUp()
+class IP6PuntSetup(object):
+ """ Setup for IPv6 Punt Police/Redirect """
+ def punt_setup(self):
self.create_pg_interfaces(range(4))
for i in self.pg_interfaces:
i.config_ip6()
i.resolve_ndp()
- def tearDown(self):
- super(TestIP6Punt, self).tearDown()
+ self.pkt = (Ether(src=self.pg0.remote_mac,
+ dst=self.pg0.local_mac) /
+ IPv6(src=self.pg0.remote_ip6,
+ dst=self.pg0.local_ip6) /
+ inet6.TCP(sport=1234, dport=1234) /
+ Raw(b'\xa5' * 100))
+
+ def punt_teardown(self):
for i in self.pg_interfaces:
i.unconfig_ip6()
i.admin_down()
+
+class TestIP6Punt(IP6PuntSetup, VppTestCase):
+ """ IPv6 Punt Police/Redirect """
+
+ def setUp(self):
+ super(TestIP6Punt, self).setUp()
+ super(TestIP6Punt, self).punt_setup()
+
+ def tearDown(self):
+ super(TestIP6Punt, self).punt_teardown()
+ super(TestIP6Punt, self).tearDown()
+
def test_ip_punt(self):
""" IP6 punt police and redirect """
- p = (Ether(src=self.pg0.remote_mac,
- dst=self.pg0.local_mac) /
- IPv6(src=self.pg0.remote_ip6,
- dst=self.pg0.local_ip6) /
- inet6.TCP(sport=1234, dport=1234) /
- Raw(b'\xa5' * 100))
-
- pkts = p * 1025
+ pkts = self.pkt * 1025
#
# Configure a punt redirect via pg1.
#
nh_addr = self.pg1.remote_ip6
- 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, "ip6-punt", 400, 0, 10, 0, rate_type=1)
policer.add_vpp_config()
- self.vapi.ip_punt_police(policer.policer_index, is_ip6=1)
+ 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)
# 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
#
- self.vapi.ip_punt_police(policer.policer_index, is_add=0, is_ip6=1)
+ 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):
""" IP6 punt redirect dump"""
#
# Configure a punt redirects
#
- nh_addr = self.pg3.remote_ip6
- self.vapi.ip_punt_redirect(self.pg0.sw_if_index,
- self.pg3.sw_if_index,
- nh_addr)
- self.vapi.ip_punt_redirect(self.pg1.sw_if_index,
- self.pg3.sw_if_index,
- nh_addr)
- self.vapi.ip_punt_redirect(self.pg2.sw_if_index,
- self.pg3.sw_if_index,
- '0::0')
+ 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, '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,
- is_ipv6=1)
- 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
self.assertEqual(str(punts[2].punt.nh), '::')
+class TestIP6PuntHandoff(IP6PuntSetup, VppTestCase):
+ """ IPv6 Punt Police/Redirect """
+ worker_config = "workers 2"
+
+ def setUp(self):
+ super(TestIP6PuntHandoff, self).setUp()
+ super(TestIP6PuntHandoff, self).punt_setup()
+
+ def tearDown(self):
+ super(TestIP6PuntHandoff, self).punt_teardown()
+ super(TestIP6PuntHandoff, self).tearDown()
+
+ def test_ip_punt(self):
+ """ IP6 punt policer thread handoff """
+ pkts = self.pkt * NUM_PKTS
+
+ #
+ # 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()
+
+ action_tx = PolicerAction(
+ VppEnum.vl_api_sse2_qos_action_type_t.SSE2_QOS_ACTION_API_TRANSMIT,
+ 0)
+ #
+ # This policer drops no packets, we are just
+ # testing that they get to the right thread.
+ #
+ policer = VppPolicer(self, "ip6-punt", 400, 0, 10, 0, 1,
+ 0, 0, False, action_tx, action_tx, action_tx)
+ policer.add_vpp_config()
+ ip_punt_policer = VppIpPuntPolicer(self, policer.policer_index,
+ is_ip6=True)
+ ip_punt_policer.add_vpp_config()
+
+ for worker in [0, 1]:
+ self.send_and_expect(self.pg0, pkts, self.pg1, worker=worker)
+ if worker == 0:
+ self.logger.debug(self.vapi.cli("show trace max 100"))
+
+ # Combined stats, all threads
+ 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)
+
+ # Worker 0, should have done all the policing
+ stats0 = policer.get_stats(worker=0)
+ self.assertEqual(stats, stats0)
+
+ # Worker 1, should have handed everything off
+ stats1 = policer.get_stats(worker=1)
+ self.assertEqual(stats1['conform_packets'], 0)
+ self.assertEqual(stats1['exceed_packets'], 0)
+ self.assertEqual(stats1['violate_packets'], 0)
+
+ # Bind the policer to worker 1 and repeat
+ policer.bind_vpp_config(1, True)
+ for worker in [0, 1]:
+ self.send_and_expect(self.pg0, pkts, self.pg1, worker=worker)
+ self.logger.debug(self.vapi.cli("show trace max 100"))
+
+ # The 2 workers should now have policed the same amount
+ stats = policer.get_stats()
+ stats0 = policer.get_stats(worker=0)
+ stats1 = policer.get_stats(worker=1)
+
+ self.assertGreater(stats0['conform_packets'], 0)
+ self.assertEqual(stats0['exceed_packets'], 0)
+ self.assertGreater(stats0['violate_packets'], 0)
+
+ self.assertGreater(stats1['conform_packets'], 0)
+ self.assertEqual(stats1['exceed_packets'], 0)
+ self.assertGreater(stats1['violate_packets'], 0)
+
+ self.assertEqual(stats0['conform_packets'] + stats1['conform_packets'],
+ stats['conform_packets'])
+
+ self.assertEqual(stats0['violate_packets'] + stats1['violate_packets'],
+ stats['violate_packets'])
+
+ # Unbind the policer and repeat
+ policer.bind_vpp_config(1, False)
+ for worker in [0, 1]:
+ self.send_and_expect(self.pg0, pkts, self.pg1, worker=worker)
+ self.logger.debug(self.vapi.cli("show trace max 100"))
+
+ # The policer should auto-bind to worker 0 when packets arrive
+ stats = policer.get_stats()
+ stats0new = policer.get_stats(worker=0)
+ stats1new = policer.get_stats(worker=1)
+
+ self.assertGreater(stats0new['conform_packets'],
+ stats0['conform_packets'])
+ self.assertEqual(stats0new['exceed_packets'], 0)
+ self.assertGreater(stats0new['violate_packets'],
+ stats0['violate_packets'])
+
+ self.assertEqual(stats1, stats1new)
+
+ #
+ # Clean up
+ #
+ ip_punt_policer.remove_vpp_config()
+ policer.remove_vpp_config()
+ ip_punt_redirect.remove_vpp_config()
+
+
class TestIPDeag(VppTestCase):
""" IPv6 Deaggregate Routes """
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, "::",
"ff:2001::%d" % jj, 128,
- 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,
proto=FibPathProto.FIB_PATH_NH_PROTO_IP6),
VppMRoutePath(self.pg1.sw_if_index,
- MRouteItfFlags.MFIB_ITF_FLAG_FORWARD,
+ MRouteItfFlags.MFIB_API_ITF_FLAG_FORWARD,
proto=FibPathProto.FIB_PATH_NH_PROTO_IP6),
VppMRoutePath(self.pg2.sw_if_index,
- MRouteItfFlags.MFIB_ITF_FLAG_FORWARD,
+ MRouteItfFlags.MFIB_API_ITF_FLAG_FORWARD,
proto=FibPathProto.FIB_PATH_NH_PROTO_IP6),
VppMRoutePath(self.pg3.sw_if_index,
- MRouteItfFlags.MFIB_ITF_FLAG_FORWARD,
+ MRouteItfFlags.MFIB_API_ITF_FLAG_FORWARD,
proto=FibPathProto.FIB_PATH_NH_PROTO_IP6)],
table_id=t.table_id).add_vpp_config()
routes[ii].append({'uni': uni,
self.send_and_expect(self.pg1, [p_echo_request_3], self.pg1)
+class TestIPv6PathMTU(VppTestCase):
+ """ IPv6 Path MTU """
+
+ def setUp(self):
+ super(TestIPv6PathMTU, self).setUp()
+
+ self.create_pg_interfaces(range(2))
+
+ # setup all interfaces
+ for i in self.pg_interfaces:
+ i.admin_up()
+ i.config_ip6()
+ i.resolve_ndp()
+
+ def tearDown(self):
+ super(TestIPv6PathMTU, self).tearDown()
+ for i in self.pg_interfaces:
+ i.unconfig_ip6()
+ i.admin_down()
+
+ def test_path_mtu_local(self):
+ """ Path MTU for attached neighbour """
+
+ self.vapi.cli("set log class ip level debug")
+ #
+ # The goal here is not test that fragmentation works correctly,
+ # that's done elsewhere, the intent is to ensure that the Path MTU
+ # settings are honoured.
+ #
+
+ #
+ # IPv6 will only frag locally generated packets, so use tunnelled
+ # packets post encap
+ #
+ tun = VppIpIpTunInterface(
+ self,
+ self.pg1,
+ self.pg1.local_ip6,
+ self.pg1.remote_ip6)
+ tun.add_vpp_config()
+ tun.admin_up()
+ tun.config_ip6()
+
+ # set the interface MTU to a reasonable value
+ self.vapi.sw_interface_set_mtu(self.pg1.sw_if_index,
+ [2800, 0, 0, 0])
+
+ p_2k = (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' * 1000))
+ p_1k = (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' * 600))
+
+ nbr = VppNeighbor(self,
+ self.pg1.sw_if_index,
+ self.pg1.remote_mac,
+ self.pg1.remote_ip6).add_vpp_config()
+
+ # this is now the interface MTU frags
+ self.send_and_expect(self.pg0, [p_2k], self.pg1, n_rx=2)
+ self.send_and_expect(self.pg0, [p_1k], self.pg1)
+
+ # drop the path MTU for this neighbour to below the interface MTU
+ # expect more frags
+ pmtu = VppIpPathMtu(self, self.pg1.remote_ip6, 1300).add_vpp_config()
+
+ # print/format the adj delegate and trackers
+ self.logger.info(self.vapi.cli("sh ip pmtu"))
+ self.logger.info(self.vapi.cli("sh adj 7"))
+
+ self.send_and_expect(self.pg0, [p_2k], self.pg1, n_rx=3)
+ self.send_and_expect(self.pg0, [p_1k], self.pg1, n_rx=2)
+
+ # increase the path MTU to more than the interface
+ # expect to use the interface MTU
+ pmtu.modify(8192)
+
+ self.send_and_expect(self.pg0, [p_2k], self.pg1, n_rx=2)
+ self.send_and_expect(self.pg0, [p_1k], self.pg1)
+
+ # go back to an MTU from the path
+ pmtu.modify(1300)
+
+ self.send_and_expect(self.pg0, [p_2k], self.pg1, n_rx=3)
+ self.send_and_expect(self.pg0, [p_1k], self.pg1, n_rx=2)
+
+ # raise the interface's MTU
+ # should still use that of the path
+ self.vapi.sw_interface_set_mtu(self.pg1.sw_if_index,
+ [2000, 0, 0, 0])
+ self.send_and_expect(self.pg0, [p_2k], self.pg1, n_rx=3)
+ self.send_and_expect(self.pg0, [p_1k], self.pg1, n_rx=2)
+
+ # set path high and interface low
+ pmtu.modify(2000)
+ self.vapi.sw_interface_set_mtu(self.pg1.sw_if_index,
+ [1300, 0, 0, 0])
+ self.send_and_expect(self.pg0, [p_2k], self.pg1, n_rx=3)
+ self.send_and_expect(self.pg0, [p_1k], self.pg1, n_rx=2)
+
+ # remove the path MTU
+ self.vapi.sw_interface_set_mtu(self.pg1.sw_if_index,
+ [2800, 0, 0, 0])
+ pmtu.modify(0)
+
+ self.send_and_expect(self.pg0, [p_2k], self.pg1, n_rx=2)
+ self.send_and_expect(self.pg0, [p_1k], self.pg1)
+
+ def test_path_mtu_remote(self):
+ """ Path MTU for remote neighbour """
+
+ self.vapi.cli("set log class ip level debug")
+ #
+ # The goal here is not test that fragmentation works correctly,
+ # that's done elsewhere, the intent is to ensure that the Path MTU
+ # settings are honoured.
+ #
+ tun_dst = "2001::1"
+
+ route = VppIpRoute(
+ self, tun_dst, 64,
+ [VppRoutePath(self.pg1.remote_ip6,
+ self.pg1.sw_if_index)]).add_vpp_config()
+
+ #
+ # IPv6 will only frag locally generated packets, so use tunnelled
+ # packets post encap
+ #
+ tun = VppIpIpTunInterface(
+ self,
+ self.pg1,
+ self.pg1.local_ip6,
+ tun_dst)
+ tun.add_vpp_config()
+ tun.admin_up()
+ tun.config_ip6()
+
+ # set the interface MTU to a reasonable value
+ self.vapi.sw_interface_set_mtu(self.pg1.sw_if_index,
+ [2800, 0, 0, 0])
+
+ p_2k = (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' * 1000))
+ p_1k = (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' * 600))
+
+ nbr = VppNeighbor(self,
+ self.pg1.sw_if_index,
+ self.pg1.remote_mac,
+ self.pg1.remote_ip6).add_vpp_config()
+
+ # this is now the interface MTU frags
+ self.send_and_expect(self.pg0, [p_2k], self.pg1, n_rx=2)
+ self.send_and_expect(self.pg0, [p_1k], self.pg1)
+
+ # drop the path MTU for this neighbour to below the interface MTU
+ # expect more frags
+ pmtu = VppIpPathMtu(self, tun_dst, 1300).add_vpp_config()
+
+ # print/format the fib entry/dpo
+ self.logger.info(self.vapi.cli("sh ip6 fib 2001::1"))
+
+ self.send_and_expect(self.pg0, [p_2k], self.pg1, n_rx=3)
+ self.send_and_expect(self.pg0, [p_1k], self.pg1, n_rx=2)
+
+ # increase the path MTU to more than the interface
+ # expect to use the interface MTU
+ pmtu.modify(8192)
+
+ self.send_and_expect(self.pg0, [p_2k], self.pg1, n_rx=2)
+ self.send_and_expect(self.pg0, [p_1k], self.pg1)
+
+ # go back to an MTU from the path
+ pmtu.modify(1300)
+
+ self.send_and_expect(self.pg0, [p_2k], self.pg1, n_rx=3)
+ self.send_and_expect(self.pg0, [p_1k], self.pg1, n_rx=2)
+
+ # raise the interface's MTU
+ # should still use that of the path
+ self.vapi.sw_interface_set_mtu(self.pg1.sw_if_index,
+ [2000, 0, 0, 0])
+ self.send_and_expect(self.pg0, [p_2k], self.pg1, n_rx=3)
+ self.send_and_expect(self.pg0, [p_1k], self.pg1, n_rx=2)
+
+ # turn the tun_dst into an attached neighbour
+ route.modify([VppRoutePath("::",
+ self.pg1.sw_if_index)])
+ nbr2 = VppNeighbor(self,
+ self.pg1.sw_if_index,
+ self.pg1.remote_mac,
+ tun_dst).add_vpp_config()
+
+ self.send_and_expect(self.pg0, [p_2k], self.pg1, n_rx=3)
+ self.send_and_expect(self.pg0, [p_1k], self.pg1, n_rx=2)
+
+ # add back to not attached
+ nbr2.remove_vpp_config()
+ route.modify([VppRoutePath(self.pg1.remote_ip6,
+ self.pg1.sw_if_index)])
+
+ # set path high and interface low
+ pmtu.modify(2000)
+ self.vapi.sw_interface_set_mtu(self.pg1.sw_if_index,
+ [1300, 0, 0, 0])
+ self.send_and_expect(self.pg0, [p_2k], self.pg1, n_rx=3)
+ self.send_and_expect(self.pg0, [p_1k], self.pg1, n_rx=2)
+
+ # remove the path MTU
+ self.vapi.sw_interface_set_mtu(self.pg1.sw_if_index,
+ [2800, 0, 0, 0])
+ pmtu.remove_vpp_config()
+ self.send_and_expect(self.pg0, [p_2k], self.pg1, n_rx=2)
+ self.send_and_expect(self.pg0, [p_1k], self.pg1)
+
+
if __name__ == '__main__':
unittest.main(testRunner=VppTestRunner)
Code Review / vpp.git / blobdiff