-#!/usr/bin/env python
+#!/usr/bin/env python3
import binascii
import random
import socket
import unittest
+import scapy.compat
from scapy.contrib.mpls import MPLS
from scapy.layers.inet import IP, UDP, TCP, ICMP, icmptypes, icmpcodes
from scapy.layers.l2 import Ether, Dot1Q, ARP
from framework import VppTestCase, VppTestRunner
from util import ppp
from vpp_ip_route import VppIpRoute, VppRoutePath, VppIpMRoute, \
- VppMRoutePath, MRouteItfFlags, MRouteEntryFlags, VppMplsIpBind, \
- VppMplsTable, VppIpTable, VppIpAddress
+ VppMRoutePath, VppMplsIpBind, \
+ VppMplsTable, VppIpTable, FibPathType, find_route, \
+ VppIpInterfaceAddress, find_route_in_dump, find_mroute_in_dump
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
class TestIPv4(VppTestCase):
""" IPv4 Test Case """
+ @classmethod
+ def setUpClass(cls):
+ super(TestIPv4, cls).setUpClass()
+
+ @classmethod
+ def tearDownClass(cls):
+ super(TestIPv4, cls).tearDownClass()
+
def setUp(self):
"""
Perform test setup before test case.
i.resolve_arp()
# config 2M FIB entries
- self.config_fib_entries(200)
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 config_fib_entries(self, count):
- """For each interface add to the FIB table *count* routes to
- "10.0.0.1/32" destination with interface's local address as next-hop
- address.
-
- :param int count: Number of FIB entries.
- - *TODO:* check if the next-hop address shouldn't be remote address
- instead of local address.
- """
- n_int = len(self.interfaces)
- percent = 0
- counter = 0.0
- dest_addr = socket.inet_pton(socket.AF_INET, "10.0.0.1")
- dest_addr_len = 32
- for i in self.interfaces:
- next_hop_address = i.local_ip4n
- for j in range(count / n_int):
- self.vapi.ip_add_del_route(
- dest_addr, dest_addr_len, next_hop_address)
- counter += 1
- if counter / count * 100 > percent:
- self.logger.info("Configure %d FIB entries .. %d%% done" %
- (count, percent))
- percent += 1
+ def show_commands_at_teardown(self):
+ 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):
"""Add load, set destination IP and extend packet to required packet
:param int packet_size: Required packet size.
:param Scapy pkt: Packet to be modified.
"""
- dst_if_idx = packet_size / 10 % 2
+ dst_if_idx = int(packet_size / 10 % 2)
dst_if = self.flows[src_if][dst_if_idx]
info = self.create_packet_info(src_if, dst_if)
payload = self.info_to_payload(info)
try:
ip = packet[IP]
udp = packet[UDP]
- payload_info = self.payload_to_info(str(packet[Raw]))
+ payload_info = self.payload_to_info(packet[Raw])
packet_index = payload_info.index
self.assertEqual(payload_info.dst, dst_sw_if_index)
self.logger.debug(
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 """
+
+ @classmethod
+ def setUpClass(cls):
+ super(TestIPv4IfAddrRoute, cls).setUpClass()
+
+ @classmethod
+ def tearDownClass(cls):
+ super(TestIPv4IfAddrRoute, cls).tearDownClass()
+
+ def setUp(self):
+ super(TestIPv4IfAddrRoute, self).setUp()
+
+ # create 1 pg interface
+ self.create_pg_interfaces(range(1))
+
+ for i in self.pg_interfaces:
+ i.admin_up()
+ i.config_ip4()
+ i.resolve_arp()
+
+ def tearDown(self):
+ super(TestIPv4IfAddrRoute, self).tearDown()
+ for i in self.pg_interfaces:
+ i.unconfig_ip4()
+ i.admin_down()
+
+ def test_ipv4_ifaddrs_same_prefix(self):
+ """ IPv4 Interface Addresses Same Prefix test
+
+ Test scenario:
+
+ - Verify no route in FIB for prefix 10.10.10.0/24
+ - Configure IPv4 address 10.10.10.10/24 on an interface
+ - Verify route in FIB for prefix 10.10.10.0/24
+ - Configure IPv4 address 10.10.10.20/24 on an interface
+ - Delete 10.10.10.10/24 from interface
+ - Verify route in FIB for prefix 10.10.10.0/24
+ - Delete 10.10.10.20/24 from interface
+ - Verify no route in FIB for prefix 10.10.10.0/24
+ """
+
+ # create two addresses, verify route not present
+ if_addr1 = VppIpInterfaceAddress(self, self.pg0, "10.10.10.10", 24)
+ if_addr2 = VppIpInterfaceAddress(self, self.pg0, "10.10.10.20", 24)
+ self.assertFalse(if_addr1.query_vpp_config()) # 10.10.10.10/24
+ self.assertFalse(find_route(self, "10.10.10.10", 32))
+ self.assertFalse(find_route(self, "10.10.10.20", 32))
+ self.assertFalse(find_route(self, "10.10.10.255", 32))
+ self.assertFalse(find_route(self, "10.10.10.0", 32))
+
+ # configure first address, verify route present
+ if_addr1.add_vpp_config()
+ self.assertTrue(if_addr1.query_vpp_config()) # 10.10.10.10/24
+ self.assertTrue(find_route(self, "10.10.10.10", 32))
+ self.assertFalse(find_route(self, "10.10.10.20", 32))
+ self.assertTrue(find_route(self, "10.10.10.255", 32))
+ self.assertTrue(find_route(self, "10.10.10.0", 32))
+
+ # configure second address, delete first, verify route not removed
+ if_addr2.add_vpp_config()
+ if_addr1.remove_vpp_config()
+ self.assertFalse(if_addr1.query_vpp_config()) # 10.10.10.10/24
+ self.assertTrue(if_addr2.query_vpp_config()) # 10.10.10.20/24
+ self.assertFalse(find_route(self, "10.10.10.10", 32))
+ self.assertTrue(find_route(self, "10.10.10.20", 32))
+ self.assertTrue(find_route(self, "10.10.10.255", 32))
+ self.assertTrue(find_route(self, "10.10.10.0", 32))
+
+ # delete second address, verify route removed
+ if_addr2.remove_vpp_config()
+ self.assertFalse(if_addr2.query_vpp_config()) # 10.10.10.20/24
+ self.assertFalse(find_route(self, "10.10.10.10", 32))
+ self.assertFalse(find_route(self, "10.10.10.20", 32))
+ self.assertFalse(find_route(self, "10.10.10.255", 32))
+ self.assertFalse(find_route(self, "10.10.10.0", 32))
+
+ def test_ipv4_ifaddr_route(self):
+ """ IPv4 Interface Address Route test
+
+ Test scenario:
+
+ - Create loopback
+ - Configure IPv4 address on loopback
+ - Verify that address is not in the FIB
+ - Bring loopback up
+ - Verify that address is in the FIB now
+ - Bring loopback down
+ - Verify that address is not in the FIB anymore
+ - Bring loopback up
+ - Configure IPv4 address on loopback
+ - Verify that address is in the FIB now
+ """
+
+ # create a loopback and configure IPv4
+ loopbacks = self.create_loopback_interfaces(1)
+ lo_if = self.lo_interfaces[0]
+
+ lo_if.local_ip4_prefix_len = 32
+ lo_if.config_ip4()
+
+ # The intf was down when addr was added -> entry not in FIB
+ fib4_dump = self.vapi.ip_route_dump(0)
+ self.assertFalse(lo_if.is_ip4_entry_in_fib_dump(fib4_dump))
+
+ # When intf is brought up, entry is added
+ lo_if.admin_up()
+ fib4_dump = self.vapi.ip_route_dump(0)
+ self.assertTrue(lo_if.is_ip4_entry_in_fib_dump(fib4_dump))
+
+ # When intf is brought down, entry is removed
+ lo_if.admin_down()
+ fib4_dump = self.vapi.ip_route_dump(0)
+ self.assertFalse(lo_if.is_ip4_entry_in_fib_dump(fib4_dump))
+
+ # Remove addr, bring up interface, re-add -> entry in FIB
+ lo_if.unconfig_ip4()
+ lo_if.admin_up()
+ lo_if.config_ip4()
+ 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 """
+ @classmethod
+ def setUpClass(cls):
+ super(TestICMPEcho, cls).setUpClass()
+
+ @classmethod
+ def tearDownClass(cls):
+ super(TestICMPEcho, cls).tearDownClass()
+
def setUp(self):
super(TestICMPEcho, self).setUp()
icmp_id = 0xb
icmp_seq = 5
- icmp_load = '\x0a' * 18
+ icmp_load = b'\x0a' * 18
p_echo_request = (Ether(src=self.pg0.remote_mac,
dst=self.pg0.local_mac) /
IP(src=self.pg0.remote_ip4, dst=self.pg0.local_ip4) /
..note:: Python API is too slow to add many routes, needs replacement.
"""
- def config_fib_many_to_one(self, start_dest_addr, next_hop_addr, count):
+ def config_fib_many_to_one(self, start_dest_addr, next_hop_addr,
+ count, start=0):
"""
:param start_dest_addr:
:param count:
:return list: added ips with 32 prefix
"""
- added_ips = []
- dest_addr = int(binascii.hexlify(socket.inet_pton(socket.AF_INET,
- start_dest_addr)), 16)
- dest_addr_len = 32
- n_next_hop_addr = socket.inet_pton(socket.AF_INET, next_hop_addr)
- for _ in range(count):
- n_dest_addr = '{:08x}'.format(dest_addr).decode('hex')
- self.vapi.ip_add_del_route(n_dest_addr, dest_addr_len,
- n_next_hop_addr)
- added_ips.append(socket.inet_ntoa(n_dest_addr))
- dest_addr += 1
- return added_ips
-
- def unconfig_fib_many_to_one(self, start_dest_addr, next_hop_addr, count):
-
- removed_ips = []
- dest_addr = int(binascii.hexlify(socket.inet_pton(socket.AF_INET,
- start_dest_addr)), 16)
- dest_addr_len = 32
- n_next_hop_addr = socket.inet_pton(socket.AF_INET, next_hop_addr)
- for _ in range(count):
- n_dest_addr = '{:08x}'.format(dest_addr).decode('hex')
- self.vapi.ip_add_del_route(n_dest_addr, dest_addr_len,
- n_next_hop_addr, is_add=0)
- removed_ips.append(socket.inet_ntoa(n_dest_addr))
- dest_addr += 1
- return removed_ips
-
- def create_stream(self, src_if, dst_if, dst_ips, count):
+ routes = []
+ for i in range(count):
+ r = VppIpRoute(self, start_dest_addr % (i + start), 32,
+ [VppRoutePath(next_hop_addr, 0xffffffff)])
+ r.add_vpp_config()
+ routes.append(r)
+ return routes
+
+ def unconfig_fib_many_to_one(self, start_dest_addr, next_hop_addr,
+ count, start=0):
+
+ routes = []
+ for i in range(count):
+ r = VppIpRoute(self, start_dest_addr % (i + start), 32,
+ [VppRoutePath(next_hop_addr, 0xffffffff)])
+ r.remove_vpp_config()
+ routes.append(r)
+ return routes
+
+ def create_stream(self, src_if, dst_if, routes, count):
pkts = []
for _ in range(count):
- dst_addr = random.choice(dst_ips)
+ dst_addr = random.choice(routes).prefix.network_address
info = self.create_packet_info(src_if, dst_if)
payload = self.info_to_payload(info)
p = (Ether(dst=src_if.local_mac, src=src_if.remote_mac) /
- IP(src=src_if.remote_ip4, dst=dst_addr) /
+ IP(src=src_if.remote_ip4, dst=str(dst_addr)) /
UDP(sport=1234, dport=1234) /
Raw(payload))
info.data = p.copy()
def _find_ip_match(self, find_in, pkt):
for p in find_in:
- if self.payload_to_info(str(p[Raw])) == \
- self.payload_to_info(str(pkt[Raw])):
+ if self.payload_to_info(p[Raw]) == \
+ self.payload_to_info(pkt[Raw]):
if p[IP].src != pkt[IP].src:
break
if p[IP].dst != pkt[IP].dst:
return p
return None
- @staticmethod
- def _match_route_detail(route_detail, ip, address_length=32, table_id=0):
- if route_detail.address == socket.inet_pton(socket.AF_INET, ip):
- if route_detail.table_id != table_id:
- return False
- elif route_detail.address_length != address_length:
- return False
- else:
- return True
- else:
- return False
-
def verify_capture(self, dst_interface, received_pkts, expected_pkts):
self.assertEqual(len(received_pkts), len(expected_pkts))
to_verify = list(expected_pkts)
to_verify.remove(x)
self.assertListEqual(to_verify, [])
- def verify_route_dump(self, fib_dump, ips):
+ def verify_route_dump(self, routes):
+ for r in routes:
+ self.assertTrue(find_route(self,
+ r.prefix.network_address,
+ r.prefix.prefixlen))
- def _ip_in_route_dump(ip, fib_dump):
- return next((route for route in fib_dump
- if self._match_route_detail(route, ip)),
- False)
-
- for ip in ips:
- self.assertTrue(_ip_in_route_dump(ip, fib_dump),
- 'IP {} is not in fib dump.'.format(ip))
-
- def verify_not_in_route_dump(self, fib_dump, ips):
-
- def _ip_in_route_dump(ip, fib_dump):
- return next((route for route in fib_dump
- if self._match_route_detail(route, ip)),
- False)
-
- for ip in ips:
- self.assertFalse(_ip_in_route_dump(ip, fib_dump),
- 'IP {} is in fib dump.'.format(ip))
+ def verify_not_in_route_dump(self, routes):
+ for r in routes:
+ self.assertFalse(find_route(self,
+ r.prefix.network_address,
+ r.prefix.prefixlen))
@classmethod
def setUpClass(cls):
super(TestIPv4FibCrud, cls).tearDownClass()
raise
+ @classmethod
+ def tearDownClass(cls):
+ super(TestIPv4FibCrud, cls).tearDownClass()
+
def setUp(self):
super(TestIPv4FibCrud, self).setUp()
self.reset_packet_infos()
+ self.configured_routes = []
+ self.deleted_routes = []
+
def test_1_add_routes(self):
- """ Add 1k routes
+ """ Add 1k routes """
- - add 100 routes check with traffic script.
- """
- # config 1M FIB entries
+ # add 100 routes check with traffic script.
self.configured_routes.extend(self.config_fib_many_to_one(
- "10.0.0.0", self.pg0.remote_ip4, 100))
+ "10.0.0.%d", self.pg0.remote_ip4, 100))
- fib_dump = self.vapi.ip_fib_dump()
- self.verify_route_dump(fib_dump, self.configured_routes)
+ self.verify_route_dump(self.configured_routes)
self.stream_1 = self.create_stream(
self.pg1, self.pg0, self.configured_routes, 100)
- delete 10 routes check with traffic script.
"""
+ # config 1M FIB entries
+ self.configured_routes.extend(self.config_fib_many_to_one(
+ "10.0.0.%d", self.pg0.remote_ip4, 100))
self.deleted_routes.extend(self.unconfig_fib_many_to_one(
- "10.0.0.10", self.pg0.remote_ip4, 10))
+ "10.0.0.%d", self.pg0.remote_ip4, 10, start=10))
for x in self.deleted_routes:
self.configured_routes.remove(x)
- fib_dump = self.vapi.ip_fib_dump()
- self.verify_route_dump(fib_dump, self.configured_routes)
+ self.verify_route_dump(self.configured_routes)
self.stream_1 = self.create_stream(
self.pg1, self.pg0, self.configured_routes, 100)
- re-add 5 routes check with traffic script.
- add 100 routes check with traffic script.
"""
+ # config 1M FIB entries
+ self.configured_routes.extend(self.config_fib_many_to_one(
+ "10.0.0.%d", self.pg0.remote_ip4, 100))
+ self.deleted_routes.extend(self.unconfig_fib_many_to_one(
+ "10.0.0.%d", self.pg0.remote_ip4, 10, start=10))
+ for x in self.deleted_routes:
+ self.configured_routes.remove(x)
+
tmp = self.config_fib_many_to_one(
- "10.0.0.10", self.pg0.remote_ip4, 5)
+ "10.0.0.%d", self.pg0.remote_ip4, 5, start=10)
self.configured_routes.extend(tmp)
for x in tmp:
self.deleted_routes.remove(x)
self.configured_routes.extend(self.config_fib_many_to_one(
- "10.0.1.0", self.pg0.remote_ip4, 100))
+ "10.0.1.%d", self.pg0.remote_ip4, 100))
- fib_dump = self.vapi.ip_fib_dump()
- self.verify_route_dump(fib_dump, self.configured_routes)
+ self.verify_route_dump(self.configured_routes)
self.stream_1 = self.create_stream(
self.pg1, self.pg0, self.configured_routes, 300)
pkts = self.pg0.get_capture(len(self.stream_1) + len(self.stream_2))
self.verify_capture(self.pg0, pkts, self.stream_1 + self.stream_2)
- def test_4_del_routes(self):
- """ Delete 1.5k routes
-
- - delete 5 routes check with traffic script.
- - add 100 routes check with traffic script.
- """
+ # delete 5 routes check with traffic script.
+ # add 100 routes check with traffic script.
self.deleted_routes.extend(self.unconfig_fib_many_to_one(
- "10.0.0.0", self.pg0.remote_ip4, 15))
+ "10.0.0.%d", self.pg0.remote_ip4, 15))
self.deleted_routes.extend(self.unconfig_fib_many_to_one(
- "10.0.0.20", self.pg0.remote_ip4, 85))
+ "10.0.0.%d", self.pg0.remote_ip4, 85))
self.deleted_routes.extend(self.unconfig_fib_many_to_one(
- "10.0.1.0", self.pg0.remote_ip4, 100))
- fib_dump = self.vapi.ip_fib_dump()
- self.verify_not_in_route_dump(fib_dump, self.deleted_routes)
+ "10.0.1.%d", self.pg0.remote_ip4, 100))
+ self.verify_not_in_route_dump(self.deleted_routes)
class TestIPNull(VppTestCase):
""" IPv4 routes via NULL """
+ @classmethod
+ def setUpClass(cls):
+ super(TestIPNull, cls).setUpClass()
+
+ @classmethod
+ def tearDownClass(cls):
+ super(TestIPNull, cls).tearDownClass()
+
def setUp(self):
super(TestIPNull, self).setUp()
#
# A route via IP NULL that will reply with ICMP unreachables
#
- ip_unreach = VppIpRoute(self, "10.0.0.1", 32, [], is_unreach=1)
+ ip_unreach = VppIpRoute(
+ self, "10.0.0.1", 32,
+ [VppRoutePath("0.0.0.0",
+ 0xffffffff,
+ type=FibPathType.FIB_PATH_TYPE_ICMP_UNREACH)])
ip_unreach.add_vpp_config()
p_unreach = (Ether(src=self.pg0.remote_mac,
dst=self.pg0.local_mac) /
IP(src=self.pg0.remote_ip4, dst="10.0.0.1") /
UDP(sport=1234, dport=1234) /
- Raw('\xa5' * 100))
-
+ Raw(b'\xa5' * 100))
self.pg0.add_stream(p_unreach)
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
#
# A route via IP NULL that will reply with ICMP prohibited
#
- ip_prohibit = VppIpRoute(self, "10.0.0.2", 32, [], is_prohibit=1)
+ ip_prohibit = VppIpRoute(
+ self, "10.0.0.2", 32,
+ [VppRoutePath("0.0.0.0",
+ 0xffffffff,
+ type=FibPathType.FIB_PATH_TYPE_ICMP_PROHIBIT)])
ip_prohibit.add_vpp_config()
p_prohibit = (Ether(src=self.pg0.remote_mac,
dst=self.pg0.local_mac) /
IP(src=self.pg0.remote_ip4, dst="10.0.0.2") /
UDP(sport=1234, dport=1234) /
- Raw('\xa5' * 100))
+ Raw(b'\xa5' * 100))
self.pg0.add_stream(p_prohibit)
self.pg_enable_capture(self.pg_interfaces)
dst=self.pg0.local_mac) /
IP(src=self.pg0.remote_ip4, dst="1.1.1.1") /
UDP(sport=1234, dport=1234) /
- Raw('\xa5' * 100))
+ Raw(b'\xa5' * 100))
r1 = VppIpRoute(self, "1.1.1.0", 24,
[VppRoutePath(self.pg1.remote_ip4,
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 = VppIpRoute(self, "1.1.1.1", 32,
+ [VppRoutePath("0.0.0.0",
+ 0xffffffff,
+ type=FibPathType.FIB_PATH_TYPE_DROP)])
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):
""" IPv4 disabled """
+ @classmethod
+ def setUpClass(cls):
+ super(TestIPDisabled, cls).setUpClass()
+
+ @classmethod
+ def tearDownClass(cls):
+ super(TestIPDisabled, cls).tearDownClass()
+
def setUp(self):
super(TestIPDisabled, self).setUp()
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,
dst=self.pg1.local_mac) /
IP(src="10.10.10.10", dst=self.pg0.remote_ip4) /
UDP(sport=1234, dport=1234) /
- Raw('\xa5' * 100))
+ Raw(b'\xa5' * 100))
pm = (Ether(src=self.pg1.remote_mac,
dst=self.pg1.local_mac) /
IP(src="10.10.10.10", dst="232.1.1.1") /
UDP(sport=1234, dport=1234) /
- Raw('\xa5' * 100))
+ Raw(b'\xa5' * 100))
#
# PG1 does not forward IP traffic
class TestIPSubNets(VppTestCase):
""" IPv4 Subnets """
+ @classmethod
+ def setUpClass(cls):
+ super(TestIPSubNets, cls).setUpClass()
+
+ @classmethod
+ def tearDownClass(cls):
+ super(TestIPSubNets, cls).tearDownClass()
+
def setUp(self):
super(TestIPSubNets, self).setUp()
# create a 2 pg interfaces
self.create_pg_interfaces(range(2))
- # pg0 we will use to experiemnt
+ # pg0 we will use to experiment
self.pg0.admin_up()
# pg1 is setup normally
dst=self.pg1.local_mac) /
IP(dst="10.10.10.10", src=self.pg0.local_ip4) /
UDP(sport=1234, dport=1234) /
- Raw('\xa5' * 100))
+ Raw(b'\xa5' * 100))
self.pg1.add_stream(p)
self.pg_enable_capture(self.pg_interfaces)
#
ip_addr_n = socket.inet_pton(socket.AF_INET, "10.10.10.10")
- self.vapi.sw_interface_add_del_address(self.pg0.sw_if_index,
- ip_addr_n,
- 16)
+ self.vapi.sw_interface_add_del_address(
+ sw_if_index=self.pg0.sw_if_index,
+ prefix="10.10.10.10/16")
pn = (Ether(src=self.pg1.remote_mac,
dst=self.pg1.local_mac) /
IP(dst="10.10.0.0", src=self.pg0.local_ip4) /
UDP(sport=1234, dport=1234) /
- Raw('\xa5' * 100))
+ Raw(b'\xa5' * 100))
pb = (Ether(src=self.pg1.remote_mac,
dst=self.pg1.local_mac) /
IP(dst="10.10.255.255", src=self.pg0.local_ip4) /
UDP(sport=1234, dport=1234) /
- Raw('\xa5' * 100))
+ Raw(b'\xa5' * 100))
self.send_and_assert_no_replies(self.pg1, pn, "IP Network address")
self.send_and_assert_no_replies(self.pg1, pb, "IP Broadcast address")
# remove the sub-net and we are forwarding via the cover again
- self.vapi.sw_interface_add_del_address(self.pg0.sw_if_index,
- ip_addr_n,
- 16,
- is_add=0)
+ self.vapi.sw_interface_add_del_address(
+ sw_if_index=self.pg0.sw_if_index,
+ prefix="10.10.10.10/16",
+ is_add=0)
+
self.pg1.add_stream(pn)
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
#
ip_addr_n = socket.inet_pton(socket.AF_INET, "10.10.10.10")
- self.vapi.sw_interface_add_del_address(self.pg0.sw_if_index,
- ip_addr_n,
- 31)
+ self.vapi.sw_interface_add_del_address(
+ sw_if_index=self.pg0.sw_if_index,
+ prefix="10.10.10.10/31")
pn = (Ether(src=self.pg1.remote_mac,
dst=self.pg1.local_mac) /
IP(dst="10.10.10.11", src=self.pg0.local_ip4) /
UDP(sport=1234, dport=1234) /
- Raw('\xa5' * 100))
+ Raw(b'\xa5' * 100))
self.pg1.add_stream(pn)
self.pg_enable_capture(self.pg_interfaces)
rx[ARP]
# remove the sub-net and we are forwarding via the cover again
- self.vapi.sw_interface_add_del_address(self.pg0.sw_if_index,
- ip_addr_n,
- 31,
- is_add=0)
+ self.vapi.sw_interface_add_del_address(
+ sw_if_index=self.pg0.sw_if_index,
+ prefix="10.10.10.10/31", is_add=0)
+
self.pg1.add_stream(pn)
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
class TestIPLoadBalance(VppTestCase):
""" IPv4 Load-Balancing """
+ @classmethod
+ def setUpClass(cls):
+ super(TestIPLoadBalance, cls).setUpClass()
+
+ @classmethod
+ def tearDownClass(cls):
+ super(TestIPLoadBalance, cls).tearDownClass()
+
def setUp(self):
super(TestIPLoadBalance, self).setUp()
input.add_stream(pkts)
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
+ rxs = []
for oo in outputs:
rx = oo._get_capture(1)
self.assertNotEqual(0, len(rx))
+ for r in rx:
+ rxs.append(r)
+ return rxs
def send_and_expect_one_itf(self, input, pkts, itf):
input.add_stream(pkts)
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))
+ Raw(b'\xa5' * 100))
port_ip_pkts.append((Ether(src=self.pg0.remote_mac,
dst=self.pg0.local_mac) /
port_ip_hdr))
src_ip_hdr = (IP(dst="10.0.0.1", src="20.0.0.%d" % ii) /
UDP(sport=1234, dport=1234) /
- Raw('\xa5' * 100))
+ Raw(b'\xa5' * 100))
src_ip_pkts.append((Ether(src=self.pg0.remote_mac,
dst=self.pg0.local_mac) /
src_ip_hdr))
# src,dst
# We are not going to ensure equal amounts of packets across each link,
# since the hash algorithm is statistical and therefore this can never
- # be guaranteed. But wuth 64 different packets we do expect some
+ # 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,
# - now only the stream with differing source address will
# load-balance
#
- self.vapi.set_ip_flow_hash(0, src=1, dst=1, sport=0, dport=0)
+ self.vapi.set_ip_flow_hash(vrf_id=0, src=1, dst=1, sport=0, dport=0)
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(0, src=1, dst=1, sport=1, dport=1)
+ self.vapi.set_ip_flow_hash(vrf_id=0, src=1, dst=1, sport=1, dport=1)
#
# Recursive prefixes
dst=self.pg0.local_mac) /
IP(dst="1.1.1.1", src="20.0.0.1") /
UDP(sport=1234, dport=1234 + ii) /
- Raw('\xa5' * 100)))
+ Raw(b'\xa5' * 100)))
src_pkts.append((Ether(src=self.pg0.remote_mac,
dst=self.pg0.local_mac) /
IP(dst="1.1.1.1", src="20.0.0.%d" % ii) /
UDP(sport=1234, dport=1234) /
- Raw('\xa5' * 100)))
+ Raw(b'\xa5' * 100)))
route_10_0_0_2 = VppIpRoute(self, "10.0.0.2", 32,
[VppRoutePath(self.pg3.remote_ip4,
[self.pg1, self.pg2,
self.pg3, self.pg4])
+ #
+ # bring down pg1 expect LB to adjust to use only those that are pu
+ #
+ 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))
+
+ #
+ # bring down pg2 expect LB to adjust to use only those that are pu
+ #
+ 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))
+
+ #
+ # bring the links back up - expect LB over all again
+ #
+ self.pg1.link_up()
+ self.pg2.link_up()
+
+ 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))
+
+ #
+ # The same link-up/down but this time admin state
+ #
+ self.pg1.admin_down()
+ 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.pg1.admin_up()
+ self.pg2.admin_up()
+ self.pg1.resolve_arp()
+ self.pg2.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))
+
#
# Recursive prefixes
# - testing that 2 stages of load-balancing, no choices
dst=self.pg0.local_mac) /
IP(dst="1.1.1.2", src="20.0.0.2") /
UDP(sport=1234, dport=1234 + ii) /
- Raw('\xa5' * 100)))
+ Raw(b'\xa5' * 100)))
route_10_0_0_3 = VppIpRoute(self, "10.0.0.3", 32,
[VppRoutePath(self.pg3.remote_ip4,
route_1_1_1_2.add_vpp_config()
#
- # inject the packet on pg0 - expect load-balancing across all 4 paths
+ # inject the packet on pg0 - rx only on via routes output interface
#
self.vapi.cli("clear trace")
self.send_and_expect_one_itf(self.pg0, port_pkts, self.pg3)
+ #
+ # Add a LB route in the presence of a down link - expect no
+ # packets over the down link
+ #
+ self.pg3.link_down()
+
+ route_10_0_0_3 = VppIpRoute(self, "10.0.0.3", 32,
+ [VppRoutePath(self.pg3.remote_ip4,
+ self.pg3.sw_if_index),
+ VppRoutePath(self.pg4.remote_ip4,
+ self.pg4.sw_if_index)])
+ route_10_0_0_3.add_vpp_config()
+
+ port_pkts = []
+ for ii in range(257):
+ port_pkts.append(Ether(src=self.pg0.remote_mac,
+ dst=self.pg0.local_mac) /
+ IP(dst="10.0.0.3", src="20.0.0.2") /
+ UDP(sport=1234, dport=1234 + ii) /
+ Raw(b'\xa5' * 100))
+
+ self.send_and_expect_one_itf(self.pg0, port_pkts, self.pg4)
+
+ # bring the link back up
+ self.pg3.link_up()
+
+ rx = self.send_and_expect_load_balancing(self.pg0, port_pkts,
+ [self.pg3, self.pg4])
+ self.assertEqual(len(src_pkts), len(rx))
+
class TestIPVlan0(VppTestCase):
""" IPv4 VLAN-0 """
+ @classmethod
+ def setUpClass(cls):
+ super(TestIPVlan0, cls).setUpClass()
+
+ @classmethod
+ def tearDownClass(cls):
+ super(TestIPVlan0, cls).tearDownClass()
+
def setUp(self):
super(TestIPVlan0, self).setUp()
IP(dst=self.pg1.remote_ip4,
src=self.pg0.remote_ip4) /
UDP(sport=1234, dport=1234) /
- Raw('\xa5' * 100)) * 65
+ Raw(b'\xa5' * 100)) * NUM_PKTS
#
# Expect that packets sent on VLAN-0 are forwarded on the
class TestIPPunt(VppTestCase):
""" IPv4 Punt Police/Redirect """
+ @classmethod
+ def setUpClass(cls):
+ super(TestIPPunt, cls).setUpClass()
+
+ @classmethod
+ def tearDownClass(cls):
+ super(TestIPPunt, cls).tearDownClass()
+
def setUp(self):
super(TestIPPunt, self).setUp()
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) /
- Raw('\xa5' * 100))
+ UDP(sport=1234, dport=1234) /
+ Raw(b'\xa5' * 100))
pkts = p * 1025
#
# Configure a punt redirect via pg1.
#
- nh_addr = VppIpAddress(self.pg1.remote_ip4).encode()
+ nh_addr = self.pg1.remote_ip4
self.vapi.ip_punt_redirect(self.pg0.sw_if_index,
self.pg1.sw_if_index,
nh_addr)
#
# add a policer
#
- policer = self.vapi.policer_add_del("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")
self.pg_start()
#
- # the number of packet recieved should be greater than 0,
+ # the number of packet received should be greater than 0,
# but not equal to the number sent, since some were policed
#
rx = self.pg1._get_capture(1)
self.assertLess(len(rx), len(pkts))
#
- # remove the poilcer. back to full rx
+ # remove the policer. back to full rx
#
self.vapi.ip_punt_police(policer.policer_index, is_add=0)
- self.vapi.policer_add_del("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)
#
#
# Configure a punt redirects
#
- nh_address = VppIpAddress(self.pg3.remote_ip4).encode()
+ nh_address = self.pg3.remote_ip4
self.vapi.ip_punt_redirect(self.pg0.sw_if_index,
self.pg3.sw_if_index,
nh_address)
nh_address)
self.vapi.ip_punt_redirect(self.pg2.sw_if_index,
self.pg3.sw_if_index,
- VppIpAddress('0.0.0.0').encode())
+ '0.0.0.0')
#
# Dump pg0 punt redirects
self.assertEqual(len(punts), 3)
for p in punts:
self.assertEqual(p.punt.tx_sw_if_index, self.pg3.sw_if_index)
- self.assertNotEqual(punts[1].punt.nh.un.ip4, self.pg3.remote_ip4)
- self.assertEqual(punts[2].punt.nh.un.ip4.address, '\x00'*4)
+ self.assertNotEqual(punts[1].punt.nh, self.pg3.remote_ip4)
+ self.assertEqual(str(punts[2].punt.nh), '0.0.0.0')
class TestIPDeag(VppTestCase):
""" IPv4 Deaggregate Routes """
+ @classmethod
+ def setUpClass(cls):
+ super(TestIPDeag, cls).setUpClass()
+
+ @classmethod
+ def tearDownClass(cls):
+ super(TestIPDeag, cls).tearDownClass()
+
def setUp(self):
super(TestIPDeag, self).setUp()
[VppRoutePath("0.0.0.0",
0xffffffff,
nh_table_id=1)])
- route_to_src = VppIpRoute(self, "1.1.1.2", 32,
- [VppRoutePath("0.0.0.0",
- 0xffffffff,
- nh_table_id=2,
- is_source_lookup=1)])
+ route_to_src = VppIpRoute(
+ self, "1.1.1.2", 32,
+ [VppRoutePath("0.0.0.0",
+ 0xffffffff,
+ nh_table_id=2,
+ type=FibPathType.FIB_PATH_TYPE_SOURCE_LOOKUP)])
route_to_dst.add_vpp_config()
route_to_src.add_vpp_config()
dst=self.pg0.local_mac) /
IP(src="5.5.5.5", dst="1.1.1.1") /
TCP(sport=1234, dport=1234) /
- Raw('\xa5' * 100))
+ Raw(b'\xa5' * 100))
p_src = (Ether(src=self.pg0.remote_mac,
dst=self.pg0.local_mac) /
IP(src="2.2.2.2", dst="1.1.1.2") /
TCP(sport=1234, dport=1234) /
- Raw('\xa5' * 100))
+ Raw(b'\xa5' * 100))
pkts_dst = p_dst * 257
pkts_src = p_src * 257
self.pg1.sw_if_index)],
table_id=1)
route_in_dst.add_vpp_config()
+
self.send_and_expect(self.pg0, pkts_dst, self.pg1)
#
dst=self.pg0.local_mac) /
IP(src="2.2.2.4", dst="2.2.2.3") /
TCP(sport=1234, dport=1234) /
- Raw('\xa5' * 100))
+ Raw(b'\xa5' * 100))
self.send_and_assert_no_replies(self.pg0, p_l * 257,
"IP lookup loop")
class TestIPInput(VppTestCase):
""" IPv4 Input Exceptions """
+ @classmethod
+ def setUpClass(cls):
+ super(TestIPInput, cls).setUpClass()
+
+ @classmethod
+ def tearDownClass(cls):
+ super(TestIPInput, cls).tearDownClass()
+
def setUp(self):
super(TestIPInput, self).setUp()
dst=self.pg1.remote_ip4,
len=40) /
UDP(sport=1234, dport=1234) /
- Raw('\xa5' * 100))
+ Raw(b'\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
dst=self.pg1.remote_ip4,
len=400) /
UDP(sport=1234, dport=1234) /
- Raw('\xa5' * 100))
+ Raw(b'\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")
#
dst=self.pg1.remote_ip4,
chksum=400) /
UDP(sport=1234, dport=1234) /
- Raw('\xa5' * 100))
+ Raw(b'\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")
#
dst=self.pg1.remote_ip4,
version=3) /
UDP(sport=1234, dport=1234) /
- Raw('\xa5' * 100))
+ Raw(b'\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")
#
dst=self.pg1.remote_ip4,
frag=1) /
UDP(sport=1234, dport=1234) /
- Raw('\xa5' * 100))
+ Raw(b'\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")
#
dst=self.pg1.remote_ip4,
ttl=1) /
UDP(sport=1234, dport=1234) /
- Raw('\xa5' * 100))
+ Raw(b'\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]
dst=self.pg1.remote_ip4,
ttl=10, flags='DF') /
UDP(sport=1234, dport=1234) /
- Raw('\xa5' * 2000))
+ Raw(b'\xa5' * 2000))
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])
+ #
+ # source address 0.0.0.0 and 25.255.255.255 and for-us
+ #
+ p_s0 = (Ether(src=self.pg0.remote_mac,
+ dst=self.pg0.local_mac) /
+ IP(src="0.0.0.0",
+ dst=self.pg0.local_ip4) /
+ ICMP(id=4, seq=4) /
+ Raw(load=b'\x0a' * 18))
+ rx = self.send_and_assert_no_replies(self.pg0, p_s0 * 17)
+
+ p_s0 = (Ether(src=self.pg0.remote_mac,
+ dst=self.pg0.local_mac) /
+ IP(src="255.255.255.255",
+ dst=self.pg0.local_ip4) /
+ ICMP(id=4, seq=4) /
+ Raw(load=b'\x0a' * 18))
+ rx = self.send_and_assert_no_replies(self.pg0, p_s0 * 17)
+
class TestIPDirectedBroadcast(VppTestCase):
""" IPv4 Directed Broadcast """
+ @classmethod
+ def setUpClass(cls):
+ super(TestIPDirectedBroadcast, cls).setUpClass()
+
+ @classmethod
+ def tearDownClass(cls):
+ super(TestIPDirectedBroadcast, cls).tearDownClass()
+
def setUp(self):
super(TestIPDirectedBroadcast, self).setUp()
IP(src="1.1.1.1",
dst=self.pg0._local_ip4_bcast) /
UDP(sport=1234, dport=1234) /
- Raw('\xa5' * 2000))
+ Raw(b'\xa5' * 2000))
p1 = (Ether(src=self.pg0.remote_mac,
dst=self.pg0.local_mac) /
IP(src="1.1.1.1",
dst=self.pg1._local_ip4_bcast) /
UDP(sport=1234, dport=1234) /
- Raw('\xa5' * 2000))
+ Raw(b'\xa5' * 2000))
self.pg0.config_ip4()
self.pg0.resolve_arp()
#
# 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()
class TestIPLPM(VppTestCase):
""" IPv4 longest Prefix Match """
+ @classmethod
+ def setUpClass(cls):
+ super(TestIPLPM, cls).setUpClass()
+
+ @classmethod
+ def tearDownClass(cls):
+ super(TestIPLPM, cls).tearDownClass()
+
def setUp(self):
super(TestIPLPM, self).setUp()
IP(src="1.1.1.1",
dst="10.1.1.1") /
UDP(sport=1234, dport=1234) /
- Raw('\xa5' * 2000))
+ Raw(b'\xa5' * 2000))
p_24 = (Ether(src=self.pg0.remote_mac,
dst=self.pg0.local_mac) /
IP(src="1.1.1.1",
dst="10.1.2.1") /
UDP(sport=1234, dport=1234) /
- Raw('\xa5' * 2000))
+ Raw(b'\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):
i.config_ip4()
i.resolve_arp()
+ @classmethod
+ def tearDownClass(cls):
+ super(TestIPv4Frag, cls).tearDownClass()
+
def test_frag_large_packets(self):
""" Fragmentation of large packets """
+ self.vapi.cli("adjacency counters enable")
+
p = (Ether(dst=self.src_if.local_mac, src=self.src_if.remote_mac) /
IP(src=self.src_if.remote_ip4, dst=self.dst_if.remote_ip4) /
UDP(sport=1234, dport=5678) / Raw())
self.extend_packet(p, 6000, "abcde")
saved_payload = p[Raw].load
+ nbr = VppNeighbor(self,
+ self.dst_if.sw_if_index,
+ self.dst_if.remote_mac,
+ self.dst_if.remote_ip4).add_vpp_config()
+
# Force fragmentation by setting MTU of output interface
# lower than packet size
self.vapi.sw_interface_set_mtu(self.dst_if.sw_if_index,
# cannot be larger then VPP buffer size (which is 2048)
packets = self.dst_if.get_capture(3)
+ # we should show 3 packets thru the neighbor
+ self.assertEqual(3, nbr.get_stats()['packets'])
+
# Assume VPP sends the fragments in order
- payload = ''
+ payload = b''
for p in packets:
payload_offset = p.frag * 8
if payload_offset > 0:
self.assert_equal(payload, saved_payload, "payload")
+class TestIPReplace(VppTestCase):
+ """ IPv4 Table Replace """
+
+ @classmethod
+ def setUpClass(cls):
+ super(TestIPReplace, cls).setUpClass()
+
+ @classmethod
+ def tearDownClass(cls):
+ super(TestIPReplace, cls).tearDownClass()
+
+ def setUp(self):
+ super(TestIPReplace, 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(TestIPReplace, self).tearDown()
+ for i in self.pg_interfaces:
+ i.admin_down()
+ i.unconfig_ip4()
+
+ 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 = [[], [], [], []]
+
+ # load up the tables with some routes
+ for ii, t in enumerate(self.tables):
+ for jj in range(N_ROUTES):
+ uni = VppIpRoute(
+ self, "10.0.0.%d" % jj, 32,
+ [VppRoutePath(links[ii].remote_hosts[0].ip4,
+ links[ii].sw_if_index),
+ VppRoutePath(links[ii].remote_hosts[1].ip4,
+ links[ii].sw_if_index)],
+ table_id=t.table_id).add_vpp_config()
+ multi = VppIpMRoute(
+ self, "0.0.0.0",
+ "239.0.0.%d" % jj, 32,
+ MRouteEntryFlags.MFIB_API_ENTRY_FLAG_NONE,
+ [VppMRoutePath(self.pg0.sw_if_index,
+ MRouteItfFlags.MFIB_API_ITF_FLAG_ACCEPT),
+ VppMRoutePath(self.pg1.sw_if_index,
+ MRouteItfFlags.MFIB_API_ITF_FLAG_FORWARD),
+ VppMRoutePath(self.pg2.sw_if_index,
+ MRouteItfFlags.MFIB_API_ITF_FLAG_FORWARD),
+ VppMRoutePath(self.pg3.sw_if_index,
+ MRouteItfFlags.MFIB_API_ITF_FLAG_FORWARD)],
+ table_id=t.table_id).add_vpp_config()
+ routes[ii].append({'uni': uni,
+ 'multi': multi})
+
+ #
+ # replace the tables a few times
+ #
+ for kk in range(3):
+ # replace_begin each table
+ for t in self.tables:
+ t.replace_begin()
+
+ # all the routes are still there
+ for ii, t in enumerate(self.tables):
+ dump = t.dump()
+ mdump = t.mdump()
+ for r in routes[ii]:
+ self.assertTrue(find_route_in_dump(dump, r['uni'], t))
+ self.assertTrue(find_mroute_in_dump(mdump, r['multi'], t))
+
+ # redownload the even numbered routes
+ for ii, t in enumerate(self.tables):
+ for jj in range(0, N_ROUTES, 2):
+ routes[ii][jj]['uni'].add_vpp_config()
+ routes[ii][jj]['multi'].add_vpp_config()
+
+ # signal each table replace_end
+ for t in self.tables:
+ t.replace_end()
+
+ # we should find the even routes, but not the odd
+ for ii, t in enumerate(self.tables):
+ dump = t.dump()
+ mdump = t.mdump()
+ for jj in range(0, N_ROUTES, 2):
+ self.assertTrue(find_route_in_dump(
+ dump, routes[ii][jj]['uni'], t))
+ self.assertTrue(find_mroute_in_dump(
+ mdump, routes[ii][jj]['multi'], t))
+ for jj in range(1, N_ROUTES - 1, 2):
+ self.assertFalse(find_route_in_dump(
+ dump, routes[ii][jj]['uni'], t))
+ self.assertFalse(find_mroute_in_dump(
+ mdump, routes[ii][jj]['multi'], t))
+
+ # reload all the routes
+ for ii, t in enumerate(self.tables):
+ for r in routes[ii]:
+ r['uni'].add_vpp_config()
+ r['multi'].add_vpp_config()
+
+ # all the routes are still there
+ for ii, t in enumerate(self.tables):
+ dump = t.dump()
+ mdump = t.mdump()
+ for r in routes[ii]:
+ self.assertTrue(find_route_in_dump(dump, r['uni'], t))
+ self.assertTrue(find_mroute_in_dump(mdump, r['multi'], t))
+
+ #
+ # finally flush the tables for good measure
+ #
+ for t in self.tables:
+ t.flush()
+ self.assertEqual(len(t.dump()), 5)
+ 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
+ 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 """
+
+ @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)