"""VPP-api provider using vpp-papi
@property hook: hook object providing before and after api/cli hooks
-
-
"""
+ _zero, _negative = range(2)
+
def __init__(self, name, shm_prefix, test_class):
self.hook = Hook("vpp-papi-provider")
self.name = name
self.shm_prefix = shm_prefix
self.test_class = test_class
+ self._expect_api_retval = self._zero
+ self._expect_stack = []
jsonfiles = []
install_dir = os.getenv('VPP_TEST_INSTALL_PATH')
self.papi = VPP(jsonfiles)
self._events = deque()
+ def __enter__(self):
+ return self
+
+ def expect_negative_api_retval(self):
+ """ Expect API failure """
+ self._expect_stack.append(self._expect_api_retval)
+ self._expect_api_retval = self._negative
+ return self
+
+ def expect_zero_api_retval(self):
+ """ Expect API success """
+ self._expect_stack.append(self._expect_api_retval)
+ self._expect_api_retval = self._zero
+ return self
+
+ def __exit__(self, exc_type, exc_value, traceback):
+ self._expect_api_retval = self._expect_stack.pop()
+
def register_hook(self, hook):
"""Replace hook registration with new hook
"""
self.hook.before_api(api_fn.__name__, api_args)
reply = api_fn(**api_args)
- if hasattr(reply, 'retval') and reply.retval != expected_retval:
- msg = "API call failed, expected retval == %d, got %s" % (
- expected_retval, repr(reply))
- self.test_class.logger.info(msg)
- raise Exception(msg)
+ if self._expect_api_retval == self._negative:
+ if hasattr(reply, 'retval') and reply.retval >= 0:
+ msg = "API call passed unexpectedly: expected negative "\
+ "return value instead of %d in %s" % \
+ (reply.retval, repr(reply))
+ self.test_class.logger.info(msg)
+ raise Exception(msg)
+ elif self._expect_api_retval == self._zero:
+ if hasattr(reply, 'retval') and reply.retval != expected_retval:
+ msg = "API call failed, expected zero return value instead "\
+ "of %d in %s" % (expected_retval, repr(reply))
+ self.test_class.logger.info(msg)
+ raise Exception(msg)
+ else:
+ raise Exception("Internal error, unexpected value for "
+ "self._expect_api_retval %s" %
+ self._expect_api_retval)
self.hook.after_api(api_fn.__name__, api_args)
return reply
def snat_add_static_mapping(
self,
local_ip,
- external_ip,
+ external_ip=0,
+ external_sw_if_index=0xFFFFFFFF,
local_port=0,
external_port=0,
addr_only=1,
:param local_ip: Local IP address
:param external_ip: External IP address
+ :param external_sw_if_index: External interface instead of IP address
:param local_port: Local port number (Default value = 0)
:param external_port: External port number (Default value = 0)
:param addr_only: 1 if address only mapping, 0 if address and port
'external_ip_address': external_ip,
'local_port': local_port,
'external_port': external_port,
+ 'external_sw_if_index': external_sw_if_index,
'vrf_id': vrf_id})
def snat_add_address_range(
"""
return self.api(self.papi.snat_show_config, {})
+ def snat_add_interface_addr(
+ self,
+ sw_if_index,
+ is_add=1):
+ """Add/del S-NAT address from interface
+
+ :param sw_if_index: Software index of the interface
+ :param is_add: 1 if add, 0 if delete (Default value = 1)
+ """
+ return self.api(self.papi.snat_add_del_interface_addr,
+ {'is_add': is_add, 'sw_if_index': sw_if_index})
+
+ def snat_interface_addr_dump(self):
+ """Dump S-NAT addresses interfaces
+ :return: Dictionary of S-NAT addresses interfaces
+ """
+ return self.api(self.papi.snat_interface_addr_dump, {})
+
+ def snat_ipfix(
+ self,
+ domain_id=1,
+ src_port=4739,
+ enable=1):
+ """Enable/disable S-NAT IPFIX logging
+
+ :param domain_id: Observation domain ID (Default value = 1)
+ :param src_port: Source port number (Default value = 4739)
+ :param enable: 1 if enable, 0 if disable (Default value = 1)
+ """
+ return self.api(
+ self.papi.snat_ipfix_enable_disable,
+ {'domain_id': domain_id,
+ 'src_port': src_port,
+ 'enable': enable})
+
def control_ping(self):
self.api(self.papi.control_ping)