from collections import deque
from six import moves
-from vpp_papi import mac_pton
+from vpp_papi import VPP, mac_pton
from hook import Hook
from vpp_l2 import L2_PORT_TYPE
-# Sphinx creates auto-generated documentation by importing the python source
-# files and collecting the docstrings from them. The NO_VPP_PAPI flag allows
-# the vpp_papi_provider.py file to be importable without having to build
-# the whole vpp api if the user only wishes to generate the test documentation.
-
-try:
- from vpp_papi import VPP
-except ImportError:
- if not os.getenv("NO_VPP_PAPI") == 1:
- raise
- pass
-
# from vnet/vnet/mpls/mpls_types.h
MPLS_IETF_MAX_LABEL = 0xfffff
MPLS_LABEL_INVALID = MPLS_IETF_MAX_LABEL + 1
self.test_class = test_class
self._expect_api_retval = self._zero
self._expect_stack = []
- jsonfiles = []
install_dir = os.getenv('VPP_INSTALL_PATH')
- for root, dirnames, filenames in os.walk(install_dir):
- for filename in fnmatch.filter(filenames, '*.api.json'):
- jsonfiles.append(os.path.join(root, filename))
- self.vpp = VPP(jsonfiles, logger=test_class.logger,
+ # Vapi requires 'VPP_API_DIR', not set when run from Makefile.
+ if 'VPP_API_DIR' not in os.environ:
+ os.environ['VPP_API_DIR'] = os.getenv('VPP_INSTALL_PATH')
+
+ self.vpp = VPP(logger=test_class.logger,
read_timeout=read_timeout)
self._events = deque()
'reverse': reverse,
'is_ipv6': is_ip6})
- def ip6_nd_proxy(self, address, sw_if_index, is_del=0):
+ def ip6_nd_proxy(self, ip, sw_if_index, is_del=0):
return self.api(self.papi.ip6nd_proxy_add_del,
- {'address': address,
+ {'ip': ip,
'sw_if_index': sw_if_index,
'is_del': is_del})
pref_lifetime=0xffffffff):
return self.api(self.papi.sw_interface_ip6nd_ra_prefix,
{'sw_if_index': sw_if_index,
- 'address': address,
- 'address_length': address_length,
+ 'prefix': {
+ 'address': address,
+ 'address_length': address_length,
+ },
'use_default': use_default,
'no_advertise': no_advertise,
'off_link': off_link,
return self.api(self.papi.bd_ip_mac_dump,
{'bd_id': bd_id})
- def want_ip4_arp_events(self, enable_disable=1, address=0):
+ def want_ip4_arp_events(self, enable_disable=1, ip="0.0.0.0"):
return self.api(self.papi.want_ip4_arp_events,
{'enable_disable': enable_disable,
- 'address': address,
+ 'ip': ip,
'pid': os.getpid(), })
- def want_ip6_nd_events(self, enable_disable=1, address=0):
+ def want_ip6_nd_events(self, enable_disable=1, ip="::"):
return self.api(self.papi.want_ip6_nd_events,
{'enable_disable': enable_disable,
- 'address': address,
+ 'ip': ip,
'pid': os.getpid(), })
def want_ip6_ra_events(self, enable_disable=1):
def ip_neighbor_add_del(self,
sw_if_index,
mac_address,
- dst_address,
+ ip_address,
is_add=1,
- is_ipv6=0,
- is_static=0,
- is_no_adj_fib=0,
- ):
+ flags=0):
""" Add neighbor MAC to IPv4 or IPv6 address.
:param sw_if_index:
:param mac_address:
:param dst_address:
:param is_add: (Default value = 1)
- :param is_ipv6: (Default value = 0)
- :param is_static: (Default value = 0)
- :param is_no_adj_fib: (Default value = 0)
+ :param flags: (Default value = 0/NONE)
"""
return self.api(
self.papi.ip_neighbor_add_del,
- {'sw_if_index': sw_if_index,
- 'is_add': is_add,
- 'is_ipv6': is_ipv6,
- 'is_static': is_static,
- 'is_no_adj_fib': is_no_adj_fib,
- 'mac_address': mac_address,
- 'dst_address': dst_address
- }
+ {
+ 'is_add': is_add,
+ 'neighbor': {
+ 'sw_if_index': sw_if_index,
+ 'flags': flags,
+ 'mac_address': mac_address,
+ 'ip_address': ip_address
+ }
+ }
)
def ip_neighbor_dump(self,
)
def proxy_arp_add_del(self,
- low_address,
- hi_address,
- vrf_id=0,
+ low,
+ hi,
+ table_id=0,
is_add=1):
""" Config Proxy Arp Range.
self.papi.proxy_arp_add_del,
{'proxy':
{
- 'vrf_id': vrf_id,
- 'low_address': low_address,
- 'hi_address': hi_address,
+ 'table_id': table_id,
+ 'low': low,
+ 'hi': hi,
},
'is_add': is_add})
self.papi.ipsec_spd_add_del, {
'spd_id': spd_id, 'is_add': is_add})
+ def ipsec_spds_dump(self):
+ return self.api(self.papi.ipsec_spds_dump, {})
+
def ipsec_interface_add_del_spd(self, spd_id, sw_if_index, is_add=1):
""" IPSEC interface SPD add/del - \
Wrapper to associate/disassociate SPD to interface in VPP
self.papi.ipsec_interface_add_del_spd,
{'spd_id': spd_id, 'sw_if_index': sw_if_index, 'is_add': is_add})
+ def ipsec_spd_interface_dump(self, spd_index=None):
+ return self.api(self.papi.ipsec_spd_interface_dump,
+ {'spd_index': spd_index if spd_index else 0,
+ 'spd_index_valid': 1 if spd_index else 0})
+
def ipsec_sad_add_del_entry(self,
sad_id,
spi,
'use_extended_sequence_number': use_extended_sequence_number,
'use_anti_replay': use_anti_replay})
+ def ipsec_sa_dump(self, sa_id=None):
+ return self.api(self.papi.ipsec_sa_dump,
+ {'sa_id': sa_id if sa_id else 0xffffffff})
+
def ipsec_spd_add_del_entry(self,
spd_id,
sa_id,
'is_ipv6': is_ipv6,
'is_ip_any': is_ip_any})
+ def ipsec_spd_dump(self, spd_id, sa_id=0xffffffff):
+ return self.api(self.papi.ipsec_spd_dump,
+ {'spd_id': spd_id,
+ 'sa_id': sa_id})
+
def ipsec_tunnel_if_add_del(self, local_ip, remote_ip, local_spi,
remote_spi, crypto_alg, local_crypto_key,
remote_crypto_key, integ_alg, local_integ_key,
return self.api(self.papi.gbp_endpoint_dump,
{'_no_type_conversion': True})
- def gbp_endpoint_group_add(self, epg, bd,
+ def gbp_endpoint_group_add(self, epg, sclass, bd,
rd, uplink_sw_if_index):
""" GBP endpoint group Add """
return self.api(self.papi.gbp_endpoint_group_add,
'uplink_sw_if_index': uplink_sw_if_index,
'bd_id': bd,
'rd_id': rd,
- 'epg_id': epg
+ 'epg_id': epg,
+ 'sclass': sclass
}})
def gbp_endpoint_group_del(self, epg):
def gbp_bridge_domain_add(self, bd_id, flags,
bvi_sw_if_index,
- uu_fwd_sw_if_index):
+ uu_fwd_sw_if_index,
+ bm_flood_sw_if_index):
""" GBP bridge-domain Add """
return self.api(self.papi.gbp_bridge_domain_add,
{'bd':
'flags': flags,
'bvi_sw_if_index': bvi_sw_if_index,
'uu_fwd_sw_if_index': uu_fwd_sw_if_index,
+ 'bm_flood_sw_if_index': bm_flood_sw_if_index,
'bd_id': bd_id
}})