-#!/usr/bin/env python
+#!/usr/bin/env python3
import binascii
import random
import socket
from vpp_ip_route import VppIpRoute, VppRoutePath, VppIpMRoute, \
VppMRoutePath, MRouteItfFlags, MRouteEntryFlags, VppMplsIpBind, \
VppMplsTable, VppIpTable, FibPathType, find_route, \
- VppIpInterfaceAddress
-from vpp_ip import VppIpAddress
+ 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
super(TestIPv4, self).tearDown()
def show_commands_at_teardown(self):
- self.logger.info(self.vapi.cli("show ip arp"))
+ 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):
: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)
"""
# create two addresses, verify route not present
- if_addr1 = VppIpInterfaceAddress(self, self.pg0,
- VppIpAddress("10.10.10.10"), 24)
- if_addr2 = VppIpInterfaceAddress(self, self.pg0,
- VppIpAddress("10.10.10.20"), 24)
- self.assertFalse(if_addr1.query_vpp_config()) # 10.10.10.0/24
+ 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))
# configure first address, verify route present
if_addr1.add_vpp_config()
- self.assertTrue(if_addr1.query_vpp_config()) # 10.10.10.0/24
+ 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))
# configure second address, delete first, verify route not removed
if_addr2.add_vpp_config()
if_addr1.remove_vpp_config()
- self.assertTrue(if_addr1.query_vpp_config()) # 10.10.10.0/24
+ 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))
# delete second address, verify route removed
if_addr2.remove_vpp_config()
- self.assertFalse(if_addr1.query_vpp_config()) # 10.10.10.0/24
+ 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))
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) /
pkts = []
for _ in range(count):
- dst_addr = random.choice(routes).prefix.address
+ 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 verify_route_dump(self, routes):
for r in routes:
- self.assertTrue(find_route(self, r.prefix.address, r.prefix.len))
+ self.assertTrue(find_route(self,
+ r.prefix.network_address,
+ r.prefix.prefixlen))
def verify_not_in_route_dump(self, routes):
for r in routes:
- self.assertFalse(find_route(self, r.prefix.address, r.prefix.len))
+ self.assertFalse(find_route(self,
+ r.prefix.network_address,
+ r.prefix.prefixlen))
@classmethod
def setUpClass(cls):
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()
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,
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
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(
- sw_if_index=self.pg0.sw_if_index, address=ip_addr_n,
- address_length=16)
+ 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(
- sw_if_index=self.pg0.sw_if_index, address=ip_addr_n,
- address_length=16, is_add=0)
+ 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(
- sw_if_index=self.pg0.sw_if_index, address=ip_addr_n,
- address_length=31)
+ 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)
# remove the sub-net and we are forwarding via the cover again
self.vapi.sw_interface_add_del_address(
- sw_if_index=self.pg0.sw_if_index, address=ip_addr_n,
- address_length=31, is_add=0)
+ 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()
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))
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,
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,
dst=self.pg0.local_mac) /
IP(dst="10.0.0.3", src="20.0.0.2") /
UDP(sport=1234, dport=1234 + ii) /
- Raw('\xa5' * 100))
+ Raw(b'\xa5' * 100))
self.send_and_expect_one_itf(self.pg0, port_pkts, self.pg4)
IP(dst=self.pg1.remote_ip4,
src=self.pg0.remote_ip4) /
UDP(sport=1234, dport=1234) /
- Raw('\xa5' * 100)) * NUM_PKTS
+ Raw(b'\xa5' * 100)) * NUM_PKTS
#
# Expect that packets sent on VLAN-0 are forwarded on the
dst=self.pg0.local_mac) /
IP(src=self.pg0.remote_ip4, dst=self.pg0.local_ip4) /
UDP(sport=1234, dport=1234) /
- Raw('\xa5' * 100))
+ Raw(b'\xa5' * 100))
pkts = p * 1025
#
# add a policer
#
- policer = self.vapi.policer_add_del(b"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")
# remove the policer. back to full rx
#
self.vapi.ip_punt_police(policer.policer_index, is_add=0)
- self.vapi.policer_add_del(b"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)
#
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
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")
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 * NUM_PKTS, self.pg1)
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 * 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 * 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 * 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 * 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 * NUM_PKTS, self.pg0)
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])
IP(src="0.0.0.0",
dst=self.pg0.local_ip4) /
ICMP(id=4, seq=4) /
- Raw(load='\x0a' * 18))
+ 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,
IP(src="255.255.255.255",
dst=self.pg0.local_ip4) /
ICMP(id=4, seq=4) /
- Raw(load='\x0a' * 18))
+ Raw(load=b'\x0a' * 18))
rx = self.send_and_assert_no_replies(self.pg0, p_s0 * 17)
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()
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 * NUM_PKTS, self.pg2)
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 """
+
+ 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_ENTRY_FLAG_NONE,
+ [VppMRoutePath(self.pg0.sw_if_index,
+ MRouteItfFlags.MFIB_ITF_FLAG_ACCEPT),
+ VppMRoutePath(self.pg1.sw_if_index,
+ MRouteItfFlags.MFIB_ITF_FLAG_FORWARD),
+ VppMRoutePath(self.pg2.sw_if_index,
+ MRouteItfFlags.MFIB_ITF_FLAG_FORWARD),
+ VppMRoutePath(self.pg3.sw_if_index,
+ MRouteItfFlags.MFIB_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
+ 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()
+
+ # 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