"""
import binascii
+import copy
import glob
import json
import shutil
from resources.libraries.python.PapiHistory import PapiHistory
from resources.libraries.python.ssh import (
SSH, SSHTimeout, exec_cmd_no_error, scp_node)
+from resources.libraries.python.VppApiCrc import VppApiCrcChecker
__all__ = ["PapiExecutor", "PapiSocketExecutor"]
ret.__getitem__ = new_get
return ret
+
class PapiSocketExecutor(object):
"""Methods for executing VPP Python API commands on forwarded socket.
# Class cache for reuse between instances.
cached_vpp_instance = None
+ api_json_directory = None
+ crc_checker_instance = None
- def __init__(self, node, remote_vpp_socket="/run/vpp-api.sock"):
+ def __init__(self, node, remote_vpp_socket=Constants.SOCKSVR_PATH):
"""Store the given arguments, declare managed variables.
:param node: Node to connect to and forward unix domain socket from.
self._ssh_control_socket = None
self._local_vpp_socket = None
+ def create_crc_checker(self):
+ """Return the cached instance or create new one from directory.
+
+ It is assumed self.api_json_directory is set, as a class variable.
+
+ :returns: Cached or newly created instance aware of .api.json content.
+ :rtype: VppApiCrc.VppApiCrcChecker
+ """
+ cls = self.__class__
+ if cls.crc_checker_instance is None:
+ cls.crc_checker_instance = VppApiCrcChecker(cls.api_json_directory)
+ return cls.crc_checker_instance
+
@property
def vpp_instance(self):
"""Return VPP instance with bindings to all API calls.
exec_cmd_no_error(node, ["bash", "-c", "'" + inner_cmd + "'"])
scp_node(node, tmp_dir + "/papi.txz", "/tmp/papi.txz", get=True)
run(["tar", "xf", tmp_dir + "/papi.txz", "-C", tmp_dir])
+ cls.api_json_directory = tmp_dir + "/usr/share/vpp/api"
+ # Perform initial checks before .api.json files are gone,
+ # by accessing the property (which also creates its instance).
+ self.create_crc_checker()
# When present locally, we finally can find the installation path.
package_path = glob.glob(tmp_dir + installed_papi_glob)[0]
# Package path has to be one level above the vpp_papi directory.
package_path = package_path.rsplit('/', 1)[0]
sys.path.append(package_path)
from vpp_papi.vpp_papi import VPPApiClient as vpp_class
- vpp_class.apidir = tmp_dir + "/usr/share/vpp/api"
+ vpp_class.apidir = cls.api_json_directory
# We need to create instance before removing from sys.path.
cls.cached_vpp_instance = vpp_class(
use_socket=True, server_address="TBD", async_thread=False,
- read_timeout=6, logger=FilteredLogger(logger, "INFO"))
+ read_timeout=14, logger=FilteredLogger(logger, "INFO"))
# Cannot use loglevel parameter, robot.api.logger lacks support.
# TODO: Stop overriding read_timeout when VPP-1722 is fixed.
finally:
def add(self, csit_papi_command, history=True, **kwargs):
"""Add next command to internal command list; return self.
+ Unless disabled, new entry to papi history is also added at this point.
The argument name 'csit_papi_command' must be unique enough as it cannot
be repeated in kwargs.
- Unless disabled, new entry to papi history is also added at this point.
+ The kwargs dict is deep-copied, so it is safe to use the original
+ with partial modifications for subsequent commands.
+
+ Any pending conflicts from .api.json processing are raised.
+ Then the command name is checked for known CRCs.
+ Unsupported commands raise an exception, as CSIT change
+ should not start using messages without making sure which CRCs
+ are supported.
+ Each CRC issue is raised only once, so subsequent tests
+ can raise other issues.
:param csit_papi_command: VPP API command.
:param history: Enable/disable adding command to PAPI command history.
:type kwargs: dict
:returns: self, so that method chaining is possible.
:rtype: PapiSocketExecutor
+ :raises RuntimeError: If unverified or conflicting CRC is encountered.
"""
if history:
PapiHistory.add_to_papi_history(
self._node, csit_papi_command, **kwargs)
self._api_command_list.append(
- dict(api_name=csit_papi_command, api_args=kwargs))
+ dict(api_name=csit_papi_command, api_args=copy.deepcopy(kwargs)))
return self
def get_replies(self, err_msg="Failed to get replies."):
api_name = command["api_name"]
papi_fn = getattr(vpp_instance.api, api_name)
try:
- reply = papi_fn(**command["api_args"])
- except IOError as err:
- # Ocassionally an error happens, try reconnect.
- logger.warn("Reconnect after error: {err!r}".format(err=err))
- self.vpp_instance.disconnect()
- # Testing showes immediate reconnect fails.
- time.sleep(1)
- self.vpp_instance.connect_sync("csit_socket")
- logger.trace("Reconnected.")
- reply = papi_fn(**command["api_args"])
+ try:
+ reply = papi_fn(**command["api_args"])
+ except IOError as err:
+ # Ocassionally an error happens, try reconnect.
+ logger.warn("Reconnect after error: {err!r}".format(
+ err=err))
+ self.vpp_instance.disconnect()
+ # Testing showes immediate reconnect fails.
+ time.sleep(1)
+ self.vpp_instance.connect_sync("csit_socket")
+ logger.trace("Reconnected.")
+ reply = papi_fn(**command["api_args"])
+ except (AttributeError, IOError) as err:
+ raise_from(AssertionError(err_msg), err, level="INFO")
# *_dump commands return list of objects, convert, ordinary reply.
if not isinstance(reply, list):
reply = [reply]
The argument name 'csit_papi_command' must be unique enough as it cannot
be repeated in kwargs.
+ The kwargs dict is deep-copied, so it is safe to use the original
+ with partial modifications for subsequent commands.
:param csit_papi_command: VPP API command.
:param history: Enable/disable adding command to PAPI command history.
if history:
PapiHistory.add_to_papi_history(
self._node, csit_papi_command, **kwargs)
- self._api_command_list.append(dict(api_name=csit_papi_command,
- api_args=kwargs))
+ self._api_command_list.append(dict(
+ api_name=csit_papi_command, api_args=copy.deepcopy(kwargs)))
return self
def get_stats(self, err_msg="Failed to get statistics.", timeout=120):
Code Review / csit.git / blobdiff