VppMRoutePath, MRouteItfFlags, MRouteEntryFlags, VppMplsIpBind, \
VppMplsTable, VppIpTable
from vpp_sub_interface import VppSubInterface, VppDot1QSubint, VppDot1ADSubint
+from vpp_papi import VppEnum
+
+NUM_PKTS = 67
class TestIPv4(VppTestCase):
def tearDown(self):
"""Run standard test teardown and log ``show ip arp``."""
super(TestIPv4, self).tearDown()
- if not self.vpp_dead:
- self.logger.info(self.vapi.cli("show ip arp"))
- # info(self.vapi.cli("show ip fib")) # many entries
+
+ def show_commands_at_teardown(self):
+ self.logger.info(self.vapi.cli("show ip arp"))
+ # info(self.vapi.cli("show ip fib")) # many entries
def config_fib_entries(self, count):
"""For each interface add to the FIB table *count* routes to
self.pg1.sw_if_index)])
r1.add_vpp_config()
- rx = self.send_and_expect(self.pg0, p * 65, self.pg1)
+ rx = self.send_and_expect(self.pg0, p * NUM_PKTS, self.pg1)
#
# insert a more specific as a drop
r2 = VppIpRoute(self, "1.1.1.1", 32, [], is_drop=1)
r2.add_vpp_config()
- self.send_and_assert_no_replies(self.pg0, p * 65, "Drop Route")
+ self.send_and_assert_no_replies(self.pg0, p * NUM_PKTS, "Drop Route")
r2.remove_vpp_config()
- rx = self.send_and_expect(self.pg0, p * 65, self.pg1)
+ rx = self.send_and_expect(self.pg0, p * NUM_PKTS, self.pg1)
class TestIPDisabled(VppTestCase):
src_ip_pkts = []
src_mpls_pkts = []
- for ii in range(65):
+ for ii in range(NUM_PKTS):
port_ip_hdr = (IP(dst="10.0.0.1", src="20.0.0.1") /
UDP(sport=1234, dport=1234 + ii) /
Raw('\xa5' * 100))
IP(dst=self.pg1.remote_ip4,
src=self.pg0.remote_ip4) /
UDP(sport=1234, dport=1234) /
- Raw('\xa5' * 100)) * 65
+ Raw('\xa5' * 100)) * NUM_PKTS
#
# Expect that packets sent on VLAN-0 are forwarded on the
def test_ip_punt(self):
""" IP punt police and redirect """
+ # use UDP packet that have a port we need to explicitly
+ # register to get punted.
+ pt_l4 = VppEnum.vl_api_punt_type_t.PUNT_API_TYPE_L4
+ af_ip4 = VppEnum.vl_api_address_family_t.ADDRESS_IP4
+ udp_proto = VppEnum.vl_api_ip_proto_t.IP_API_PROTO_UDP
+ punt_udp = {
+ 'type': pt_l4,
+ 'punt': {
+ 'l4': {
+ 'af': af_ip4,
+ 'protocol': udp_proto,
+ 'port': 1234,
+ }
+ }
+ }
+
+ self.vapi.set_punt(is_add=1, punt=punt_udp)
+
p = (Ether(src=self.pg0.remote_mac,
dst=self.pg0.local_mac) /
IP(src=self.pg0.remote_ip4, dst=self.pg0.local_ip4) /
- TCP(sport=1234, dport=1234) /
+ UDP(sport=1234, dport=1234) /
Raw('\xa5' * 100))
pkts = p * 1025
UDP(sport=1234, dport=1234) /
Raw('\xa5' * 100))
- rx = self.send_and_expect(self.pg0, p_short * 65, self.pg1)
+ rx = self.send_and_expect(self.pg0, p_short * NUM_PKTS, self.pg1)
#
# Packet too long - this is dropped
UDP(sport=1234, dport=1234) /
Raw('\xa5' * 100))
- rx = self.send_and_assert_no_replies(self.pg0, p_long * 65,
+ rx = self.send_and_assert_no_replies(self.pg0, p_long * NUM_PKTS,
"too long")
#
UDP(sport=1234, dport=1234) /
Raw('\xa5' * 100))
- rx = self.send_and_assert_no_replies(self.pg0, p_chksum * 65,
+ rx = self.send_and_assert_no_replies(self.pg0, p_chksum * NUM_PKTS,
"bad checksum")
#
UDP(sport=1234, dport=1234) /
Raw('\xa5' * 100))
- rx = self.send_and_assert_no_replies(self.pg0, p_ver * 65,
+ rx = self.send_and_assert_no_replies(self.pg0, p_ver * NUM_PKTS,
"funky version")
#
UDP(sport=1234, dport=1234) /
Raw('\xa5' * 100))
- rx = self.send_and_assert_no_replies(self.pg0, p_frag * 65,
+ rx = self.send_and_assert_no_replies(self.pg0, p_frag * NUM_PKTS,
"frag offset")
#
UDP(sport=1234, dport=1234) /
Raw('\xa5' * 100))
- rx = self.send_and_expect(self.pg0, p_ttl * 65, self.pg0)
+ rx = self.send_and_expect(self.pg0, p_ttl * NUM_PKTS, self.pg0)
rx = rx[0]
icmp = rx[ICMP]
self.vapi.sw_interface_set_mtu(self.pg1.sw_if_index, [1500, 0, 0, 0])
- rx = self.send_and_expect(self.pg0, p_mtu * 65, self.pg0)
+ rx = self.send_and_expect(self.pg0, p_mtu * NUM_PKTS, self.pg0)
rx = rx[0]
icmp = rx[ICMP]
self.assertEqual(icmp.dst, self.pg1.remote_ip4)
self.vapi.sw_interface_set_mtu(self.pg1.sw_if_index, [2500, 0, 0, 0])
- rx = self.send_and_expect(self.pg0, p_mtu * 65, self.pg1)
+ rx = self.send_and_expect(self.pg0, p_mtu * NUM_PKTS, self.pg1)
# Reset MTU for subsequent tests
self.vapi.sw_interface_set_mtu(self.pg1.sw_if_index, [9000, 0, 0, 0])
#
# test packet is L2 broadcast
#
- rx = self.send_and_expect(self.pg1, p0 * 65, self.pg0)
+ rx = self.send_and_expect(self.pg1, p0 * NUM_PKTS, self.pg0)
self.assertTrue(rx[0][Ether].dst, "ff:ff:ff:ff:ff:ff")
- self.send_and_assert_no_replies(self.pg0, p1 * 65,
+ self.send_and_assert_no_replies(self.pg0, p1 * NUM_PKTS,
"directed broadcast disabled")
#
#
self.vapi.sw_interface_set_ip_directed_broadcast(
self.pg0.sw_if_index, 0)
- self.send_and_assert_no_replies(self.pg1, p0 * 65,
+ self.send_and_assert_no_replies(self.pg1, p0 * NUM_PKTS,
"directed broadcast disabled")
self.vapi.sw_interface_set_ip_directed_broadcast(
self.pg0.sw_if_index, 1)
- rx = self.send_and_expect(self.pg1, p0 * 65, self.pg0)
+ rx = self.send_and_expect(self.pg1, p0 * NUM_PKTS, self.pg0)
self.pg0.unconfig_ip4()
self.pg1.unconfig_ip4()
Raw('\xa5' * 2000))
self.logger.info(self.vapi.cli("sh ip fib mtrie"))
- rx = self.send_and_expect(self.pg0, p_8 * 65, self.pg2)
- rx = self.send_and_expect(self.pg0, p_24 * 65, self.pg1)
+ rx = self.send_and_expect(self.pg0, p_8 * NUM_PKTS, self.pg2)
+ rx = self.send_and_expect(self.pg0, p_24 * NUM_PKTS, self.pg1)
class TestIPv4Frag(VppTestCase):