Change-Id: I60818a22f543f9a3c3f62f9c67f5e4239e5b045a
Signed-off-by: Jan Gelety <jgelety@cisco.com>
interfaces in 5 VRFs are tested
- jumbo packets in configuration with 15 pg-ip4 interfaces leads to \
problems too
interfaces in 5 VRFs are tested
- jumbo packets in configuration with 15 pg-ip4 interfaces leads to \
problems too
- - Reset of FIB table / VRF does not remove routes from IP FIB (see Jira \
- ticket https://jira.fd.io/browse/VPP-560) so checks of reset VRF tables \
- are skipped in tests 2, 3 and 4
**config 1**
- add 15 pg-ip4 interfaces
**config 1**
- add 15 pg-ip4 interfaces
- no packet received in case of pg-ip4 interfaces not in VRF
**config 2**
- no packet received in case of pg-ip4 interfaces not in VRF
**config 2**
**test 2**
- send IP4 packets between all pg-ip4 interfaces in all VRF groups
**test 2**
- send IP4 packets between all pg-ip4 interfaces in all VRF groups
- no packet received in case of pg-ip4 interfaces not in VRF
**config 3**
- no packet received in case of pg-ip4 interfaces not in VRF
**config 3**
- - add 1 of deleted VRFs and 1 new VRF
+ - add 1 of reset VRFs and 1 new VRF
**test 3**
- send IP4 packets between all pg-ip4 interfaces in all VRF groups
**test 3**
- send IP4 packets between all pg-ip4 interfaces in all VRF groups
- no packet received in case of pg-ip4 interfaces not in VRF
**config 4**
- no packet received in case of pg-ip4 interfaces not in VRF
**config 4**
- - delete all VRFs (i.e. no VRF except VRF=0 created)
+ - reset all created VRFs
**test 4**
- send IP4 packets between all pg-ip4 interfaces in all VRF groups
**test 4**
- send IP4 packets between all pg-ip4 interfaces in all VRF groups
import unittest
import random
import unittest
import random
from scapy.packet import Raw
from scapy.layers.l2 import Ether
from scapy.packet import Raw
from scapy.layers.l2 import Ether
from framework import VppTestCase, VppTestRunner
from util import ppp
from framework import VppTestCase, VppTestRunner
from util import ppp
+from vrf import VRFState
# Create list of VRFs
cls.vrf_list = list()
# Create list of VRFs
cls.vrf_list = list()
- # Create list of deleted VRFs
- cls.vrf_deleted_list = list()
+ # Create list of reset VRFs
+ cls.vrf_reset_list = list()
# Create list of pg_interfaces in VRFs
cls.pg_in_vrf = list()
# Create list of pg_interfaces in VRFs
cls.pg_in_vrf = list()
for i in range(count):
vrf_id = i + start
pg_if = self.pg_if_by_vrf_id[vrf_id][0]
for i in range(count):
vrf_id = i + start
pg_if = self.pg_if_by_vrf_id[vrf_id][0]
- dest_addr = pg_if.remote_hosts[0].ip4n
+ dest_addr = pg_if.local_ip4n
dest_addr_len = 24
self.vapi.ip_table_add_del(vrf_id, is_add=1)
self.vapi.ip_add_del_route(
dest_addr_len = 24
self.vapi.ip_table_add_del(vrf_id, is_add=1)
self.vapi.ip_add_del_route(
self.logger.info("IPv4 VRF ID %d created" % vrf_id)
if vrf_id not in self.vrf_list:
self.vrf_list.append(vrf_id)
self.logger.info("IPv4 VRF ID %d created" % vrf_id)
if vrf_id not in self.vrf_list:
self.vrf_list.append(vrf_id)
- if vrf_id in self.vrf_deleted_list:
- self.vrf_deleted_list.remove(vrf_id)
+ if vrf_id in self.vrf_reset_list:
+ self.vrf_reset_list.remove(vrf_id)
for j in range(self.pg_ifs_per_vrf):
pg_if = self.pg_if_by_vrf_id[vrf_id][j]
pg_if.set_table_ip4(vrf_id)
for j in range(self.pg_ifs_per_vrf):
pg_if = self.pg_if_by_vrf_id[vrf_id][j]
pg_if.set_table_ip4(vrf_id)
self.logger.debug(self.vapi.ppcli("show ip fib"))
self.logger.debug(self.vapi.ppcli("show ip arp"))
self.logger.debug(self.vapi.ppcli("show ip fib"))
self.logger.debug(self.vapi.ppcli("show ip arp"))
- def delete_vrf(self, vrf_id):
+ def reset_vrf_and_remove_from_vrf_list(self, vrf_id):
- Delete required FIB table / VRF.
+ Reset required FIB table / VRF and remove it from VRF list.
- :param int vrf_id: The FIB table / VRF ID to be deleted.
+ :param int vrf_id: The FIB table / VRF ID to be reset.
"""
# self.vapi.reset_vrf(vrf_id, is_ipv6=0)
self.vapi.reset_fib(vrf_id, is_ipv6=0)
if vrf_id in self.vrf_list:
self.vrf_list.remove(vrf_id)
"""
# self.vapi.reset_vrf(vrf_id, is_ipv6=0)
self.vapi.reset_fib(vrf_id, is_ipv6=0)
if vrf_id in self.vrf_list:
self.vrf_list.remove(vrf_id)
- if vrf_id not in self.vrf_deleted_list:
- self.vrf_deleted_list.append(vrf_id)
+ if vrf_id not in self.vrf_reset_list:
+ self.vrf_reset_list.append(vrf_id)
for j in range(self.pg_ifs_per_vrf):
pg_if = self.pg_if_by_vrf_id[vrf_id][j]
pg_if.unconfig_ip4()
for j in range(self.pg_ifs_per_vrf):
pg_if = self.pg_if_by_vrf_id[vrf_id][j]
pg_if.unconfig_ip4()
self.pg_in_vrf.remove(pg_if)
if pg_if not in self.pg_not_in_vrf:
self.pg_not_in_vrf.append(pg_if)
self.pg_in_vrf.remove(pg_if)
if pg_if not in self.pg_not_in_vrf:
self.pg_not_in_vrf.append(pg_if)
- self.logger.info("IPv4 VRF ID %d reset" % vrf_id)
+ self.logger.info("IPv4 VRF ID %d reset finished" % vrf_id)
self.logger.debug(self.vapi.ppcli("show ip fib"))
self.logger.debug(self.vapi.ppcli("show ip arp"))
self.vapi.ip_table_add_del(vrf_id, is_add=0)
self.logger.debug(self.vapi.ppcli("show ip fib"))
self.logger.debug(self.vapi.ppcli("show ip arp"))
self.vapi.ip_table_add_del(vrf_id, is_add=0)
:return: 1 if the FIB table / VRF ID is configured, otherwise return 0.
"""
ip_fib_dump = self.vapi.ip_fib_dump()
:return: 1 if the FIB table / VRF ID is configured, otherwise return 0.
"""
ip_fib_dump = self.vapi.ip_fib_dump()
vrf_count = 0
for ip_fib_details in ip_fib_dump:
vrf_count = 0
for ip_fib_details in ip_fib_dump:
- if ip_fib_details[2] == vrf_id:
- vrf_count += 1
- if vrf_count == 0:
+ if ip_fib_details.table_id == vrf_id:
+ if not vrf_exist:
+ vrf_exist = True
+ addr = socket.inet_ntoa(ip_fib_details.address)
+ found = False
+ for pg_if in self.pg_if_by_vrf_id[vrf_id]:
+ if found:
+ break
+ for host in pg_if.remote_hosts:
+ if str(addr) == str(host.ip4):
+ vrf_count += 1
+ found = True
+ break
+ if not vrf_exist and vrf_count == 0:
self.logger.info("IPv4 VRF ID %d is not configured" % vrf_id)
self.logger.info("IPv4 VRF ID %d is not configured" % vrf_id)
+ return VRFState.not_configured
+ elif vrf_exist and vrf_count == 0:
+ self.logger.info("IPv4 VRF ID %d has been reset" % vrf_id)
+ return VRFState.reset
else:
self.logger.info("IPv4 VRF ID %d is configured" % vrf_id)
else:
self.logger.info("IPv4 VRF ID %d is configured" % vrf_id)
+ return VRFState.configured
def run_verify_test(self):
"""
def run_verify_test(self):
"""
raise Exception("Unknown interface: %s" % pg_if.name)
def test_ip4_vrf_01(self):
raise Exception("Unknown interface: %s" % pg_if.name)
def test_ip4_vrf_01(self):
- """ IP4 VRF Multi-instance test 1 - create 5 BDs
+ """ IP4 VRF Multi-instance test 1 - create 4 VRFs
"""
# Config 1
# Create 4 VRFs
"""
# Config 1
# Create 4 VRFs
# Verify 1
for vrf_id in self.vrf_list:
# Verify 1
for vrf_id in self.vrf_list:
- self.assertEqual(self.verify_vrf(vrf_id), 1)
+ self.assert_equal(self.verify_vrf(vrf_id),
+ VRFState.configured, VRFState)
# Test 1
self.run_verify_test()
def test_ip4_vrf_02(self):
# Test 1
self.run_verify_test()
def test_ip4_vrf_02(self):
- """ IP4 VRF Multi-instance test 2 - delete 2 VRFs
+ """ IP4 VRF Multi-instance test 2 - reset 2 VRFs
- # Delete 2 VRFs
- self.delete_vrf(1)
- self.delete_vrf(2)
+ # Reset 2 VRFs
+ self.reset_vrf_and_remove_from_vrf_list(1)
+ self.reset_vrf_and_remove_from_vrf_list(2)
- # for vrf_id in self.vrf_deleted_list:
- # self.assertEqual(self.verify_vrf(vrf_id), 0)
+ for vrf_id in self.vrf_reset_list:
+ self.assert_equal(self.verify_vrf(vrf_id),
+ VRFState.reset, VRFState)
for vrf_id in self.vrf_list:
for vrf_id in self.vrf_list:
- self.assertEqual(self.verify_vrf(vrf_id), 1)
+ self.assert_equal(self.verify_vrf(vrf_id),
+ VRFState.configured, VRFState)
# Test 2
self.run_verify_test()
# Test 2
self.run_verify_test()
""" IP4 VRF Multi-instance 3 - add 2 VRFs
"""
# Config 3
""" IP4 VRF Multi-instance 3 - add 2 VRFs
"""
# Config 3
- # Add 1 of deleted VRFs and 1 new VRF
+ # Add 1 of reset VRFs and 1 new VRF
self.create_vrf_and_assign_interfaces(1)
self.create_vrf_and_assign_interfaces(1, start=5)
# Verify 3
self.create_vrf_and_assign_interfaces(1)
self.create_vrf_and_assign_interfaces(1, start=5)
# Verify 3
- # for vrf_id in self.vrf_deleted_list:
- # self.assertEqual(self.verify_vrf(vrf_id), 0)
+ for vrf_id in self.vrf_reset_list:
+ self.assert_equal(self.verify_vrf(vrf_id),
+ VRFState.reset, VRFState)
for vrf_id in self.vrf_list:
for vrf_id in self.vrf_list:
- self.assertEqual(self.verify_vrf(vrf_id), 1)
+ self.assert_equal(self.verify_vrf(vrf_id),
+ VRFState.configured, VRFState)
# Test 3
self.run_verify_test()
def test_ip4_vrf_04(self):
# Test 3
self.run_verify_test()
def test_ip4_vrf_04(self):
- """ IP4 VRF Multi-instance test 4 - delete 4 VRFs
+ """ IP4 VRF Multi-instance test 4 - reset 4 VRFs
- # Delete all VRFs (i.e. no VRF except VRF=0 created)
+ # Reset all VRFs (i.e. no VRF except VRF=0 configured)
for i in range(len(self.vrf_list)):
for i in range(len(self.vrf_list)):
- self.delete_vrf(self.vrf_list[0])
+ self.reset_vrf_and_remove_from_vrf_list(self.vrf_list[0])
- # for vrf_id in self.vrf_deleted_list:
- # self.assertEqual(self.verify_vrf(vrf_id), 0)
- for vrf_id in self.vrf_list:
- self.assertEqual(self.verify_vrf(vrf_id), 1)
+ for vrf_id in self.vrf_reset_list:
+ self.assert_equal(self.verify_vrf(vrf_id),
+ VRFState.reset, VRFState)
+ vrf_list_length = len(self.vrf_list)
+ self.assertEqual(
+ vrf_list_length, 0,
+ "List of configured VRFs is not empty: %s != 0" % vrf_list_length)
# Test 4
self.run_verify_test()
# Test 4
self.run_verify_test()
RouterAlert, IPv6ExtHdrHopByHop
from scapy.utils6 import in6_ismaddr, in6_isllsnmaddr, in6_getAddrType
from scapy.pton_ntop import inet_ntop
RouterAlert, IPv6ExtHdrHopByHop
from scapy.utils6 import in6_ismaddr, in6_isllsnmaddr, in6_getAddrType
from scapy.pton_ntop import inet_ntop
-from scapy.data import IPV6_ADDR_UNICAST
from framework import VppTestCase, VppTestRunner
from util import ppp
from framework import VppTestCase, VppTestRunner
from util import ppp
-
-# VRF status constants
-VRF_NOT_CONFIGURED = 0
-VRF_CONFIGURED = 1
-VRF_RESET = 2
+from vrf import VRFState
for i in range(count):
vrf_id = i + start
pg_if = self.pg_if_by_vrf_id[vrf_id][0]
for i in range(count):
vrf_id = i + start
pg_if = self.pg_if_by_vrf_id[vrf_id][0]
- dest_addr = pg_if.remote_hosts[0].ip6n
+ dest_addr = pg_if.local_ip6n
dest_addr_len = 64
self.vapi.ip_table_add_del(vrf_id, is_add=1, is_ipv6=1)
self.vapi.ip_add_del_route(
dest_addr_len = 64
self.vapi.ip_table_add_del(vrf_id, is_add=1, is_ipv6=1)
self.vapi.ip_add_del_route(
self.logger.debug(self.vapi.ppcli("show ip6 fib"))
self.logger.debug(self.vapi.ppcli("show ip6 neighbors"))
self.logger.debug(self.vapi.ppcli("show ip6 fib"))
self.logger.debug(self.vapi.ppcli("show ip6 neighbors"))
- def reset_vrf(self, vrf_id):
+ def reset_vrf_and_remove_from_vrf_list(self, vrf_id):
- Reset required FIB table / VRF.
+ Reset required FIB table / VRF and remove it from VRF list.
:param int vrf_id: The FIB table / VRF ID to be reset.
"""
:param int vrf_id: The FIB table / VRF ID to be reset.
"""
self.pg_in_vrf.remove(pg_if)
if pg_if not in self.pg_not_in_vrf:
self.pg_not_in_vrf.append(pg_if)
self.pg_in_vrf.remove(pg_if)
if pg_if not in self.pg_not_in_vrf:
self.pg_not_in_vrf.append(pg_if)
- self.logger.info("IPv6 VRF ID %d reset" % vrf_id)
+ self.logger.info("IPv6 VRF ID %d reset finished" % vrf_id)
self.logger.debug(self.vapi.ppcli("show ip6 fib"))
self.logger.debug(self.vapi.ppcli("show ip6 neighbors"))
self.vapi.ip_table_add_del(vrf_id, is_add=0, is_ipv6=1)
self.logger.debug(self.vapi.ppcli("show ip6 fib"))
self.logger.debug(self.vapi.ppcli("show ip6 neighbors"))
self.vapi.ip_table_add_del(vrf_id, is_add=0, is_ipv6=1)
if not vrf_exist:
vrf_exist = True
addr = inet_ntop(socket.AF_INET6, ip6_fib_details.address)
if not vrf_exist:
vrf_exist = True
addr = inet_ntop(socket.AF_INET6, ip6_fib_details.address)
- addrtype = in6_getAddrType(addr)
- vrf_count += 1 if addrtype == IPV6_ADDR_UNICAST else 0
+ found = False
+ for pg_if in self.pg_if_by_vrf_id[vrf_id]:
+ if found:
+ break
+ for host in pg_if.remote_hosts:
+ if str(addr) == str(host.ip6):
+ vrf_count += 1
+ found = True
+ break
if not vrf_exist and vrf_count == 0:
self.logger.info("IPv6 VRF ID %d is not configured" % vrf_id)
if not vrf_exist and vrf_count == 0:
self.logger.info("IPv6 VRF ID %d is not configured" % vrf_id)
- return VRF_NOT_CONFIGURED
+ return VRFState.not_configured
elif vrf_exist and vrf_count == 0:
self.logger.info("IPv6 VRF ID %d has been reset" % vrf_id)
elif vrf_exist and vrf_count == 0:
self.logger.info("IPv6 VRF ID %d has been reset" % vrf_id)
else:
self.logger.info("IPv6 VRF ID %d is configured" % vrf_id)
else:
self.logger.info("IPv6 VRF ID %d is configured" % vrf_id)
+ return VRFState.configured
def run_verify_test(self):
"""
def run_verify_test(self):
"""
# Verify 1
for vrf_id in self.vrf_list:
# Verify 1
for vrf_id in self.vrf_list:
- self.assertEqual(self.verify_vrf(vrf_id), VRF_CONFIGURED)
+ self.assert_equal(self.verify_vrf(vrf_id),
+ VRFState.configured, VRFState)
# Test 1
self.run_verify_test()
# Test 1
self.run_verify_test()
"""
# Config 2
# Delete 2 VRFs
"""
# Config 2
# Delete 2 VRFs
- self.reset_vrf(1)
- self.reset_vrf(2)
+ self.reset_vrf_and_remove_from_vrf_list(1)
+ self.reset_vrf_and_remove_from_vrf_list(2)
# Verify 2
for vrf_id in self.vrf_reset_list:
# Verify 2
for vrf_id in self.vrf_reset_list:
- self.assertEqual(self.verify_vrf(vrf_id), VRF_RESET)
+ self.assert_equal(self.verify_vrf(vrf_id),
+ VRFState.reset, VRFState)
for vrf_id in self.vrf_list:
for vrf_id in self.vrf_list:
- self.assertEqual(self.verify_vrf(vrf_id), VRF_CONFIGURED)
+ self.assert_equal(self.verify_vrf(vrf_id),
+ VRFState.configured, VRFState)
# Test 2
self.run_verify_test()
# Reset routes learned from ICMPv6 Neighbor Discovery
for vrf_id in self.vrf_reset_list:
# Test 2
self.run_verify_test()
# Reset routes learned from ICMPv6 Neighbor Discovery
for vrf_id in self.vrf_reset_list:
+ self.reset_vrf_and_remove_from_vrf_list(vrf_id)
def test_ip6_vrf_03(self):
""" IP6 VRF Multi-instance 3 - add 2 VRFs
def test_ip6_vrf_03(self):
""" IP6 VRF Multi-instance 3 - add 2 VRFs
# Verify 3
for vrf_id in self.vrf_reset_list:
# Verify 3
for vrf_id in self.vrf_reset_list:
- self.assertEqual(self.verify_vrf(vrf_id), VRF_RESET)
+ self.assert_equal(self.verify_vrf(vrf_id),
+ VRFState.reset, VRFState)
for vrf_id in self.vrf_list:
for vrf_id in self.vrf_list:
- self.assertEqual(self.verify_vrf(vrf_id), VRF_CONFIGURED)
+ self.assert_equal(self.verify_vrf(vrf_id),
+ VRFState.configured, VRFState)
# Test 3
self.run_verify_test()
# Reset routes learned from ICMPv6 Neighbor Discovery
for vrf_id in self.vrf_reset_list:
# Test 3
self.run_verify_test()
# Reset routes learned from ICMPv6 Neighbor Discovery
for vrf_id in self.vrf_reset_list:
+ self.reset_vrf_and_remove_from_vrf_list(vrf_id)
def test_ip6_vrf_04(self):
""" IP6 VRF Multi-instance test 4 - reset 4 VRFs
"""
# Config 4
def test_ip6_vrf_04(self):
""" IP6 VRF Multi-instance test 4 - reset 4 VRFs
"""
# Config 4
- # Reset all VRFs (i.e. no VRF except VRF=0 created)
+ # Reset all VRFs (i.e. no VRF except VRF=0 configured)
for i in range(len(self.vrf_list)):
for i in range(len(self.vrf_list)):
- self.reset_vrf(self.vrf_list[0])
+ self.reset_vrf_and_remove_from_vrf_list(self.vrf_list[0])
# Verify 4
for vrf_id in self.vrf_reset_list:
# Verify 4
for vrf_id in self.vrf_reset_list:
- self.assertEqual(self.verify_vrf(vrf_id), VRF_RESET)
- for vrf_id in self.vrf_list:
- self.assertEqual(self.verify_vrf(vrf_id), VRF_CONFIGURED)
+ self.assert_equal(self.verify_vrf(vrf_id),
+ VRFState.reset, VRFState)
+ vrf_list_length = len(self.vrf_list)
+ self.assertEqual(
+ vrf_list_length, 0,
+ "List of configured VRFs is not empty: %s != 0" % vrf_list_length)
# Test 4
self.run_verify_test()
# Test 4
self.run_verify_test()
--- /dev/null
+""" VRF Status codes """
+
+from util import NumericConstant
+
+
+class VRFState(NumericConstant):
+ """ VRF State """
+ not_configured = 0
+ configured = 1
+ reset = 2
+
+ desc_dict = {
+ not_configured: "VRF not configured",
+ configured: "VRF configured",
+ reset: "VRF reset",
+ }
+
+ def __init__(self, value):
+ NumericConstant.__init__(self, value)