"""
# Class cache for reuse between instances.
- cached_vpp_instance = None
- api_json_directory = None
- crc_checker_instance = None
+ vpp_instance = None
+ """Takes long time to create, stores all PAPI functions and types."""
+ crc_checker = None
+ """Accesses .api.json files at creation, caching allows deleting them."""
def __init__(self, node, remote_vpp_socket=Constants.SOCKSVR_PATH):
"""Store the given arguments, declare managed variables.
self._temp_dir = None
self._ssh_control_socket = None
self._local_vpp_socket = None
+ self.initialize_vpp_instance()
- def create_crc_checker(self):
- """Return the cached instance or create new one from directory.
+ def initialize_vpp_instance(self):
+ """Create VPP instance with bindings to API calls, store as class field.
- It is assumed self.api_json_directory is set, as a class variable.
+ No-op if the instance had been stored already.
- :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.
-
- The returned instance is initialized for unix domain socket access,
+ The instance is initialized for unix domain socket access,
it has initialized all the bindings, but it is not connected
- (to local socket) yet.
-
- First invocation downloads .api.json files from self._node
- into a temporary directory.
-
- After first invocation, the result is cached, so other calls are quick.
- Class variable is used as the cache, but this property is defined as
- an instance method, so that _node (for api files) is known.
+ (to a local socket) yet.
- :returns: Initialized but not connected VPP instance.
- :rtype: vpp_papi.VPPApiClient
+ This method downloads .api.json files from self._node
+ into a temporary directory, deletes them finally.
"""
- cls = self.__class__
- if cls.cached_vpp_instance is not None:
- return cls.cached_vpp_instance
+ if self.vpp_instance:
+ return
+ cls = self.__class__ # Shorthand for setting class fields.
tmp_dir = tempfile.mkdtemp(dir="/tmp")
- package_path = "Not set yet."
+ package_path = None
try:
# Pack, copy and unpack Python part of VPP installation from _node.
# TODO: Use rsync or recursive version of ssh.scp_node instead?
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"
+ 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()
+ # by creating the checker instance.
+ cls.crc_checker = VppApiCrcChecker(api_json_directory)
# 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 = cls.api_json_directory
+ vpp_class.apidir = api_json_directory
# We need to create instance before removing from sys.path.
- cls.cached_vpp_instance = vpp_class(
+ cls.vpp_instance = vpp_class(
use_socket=True, server_address="TBD", async_thread=False,
read_timeout=14, logger=FilteredLogger(logger, "INFO"))
# Cannot use loglevel parameter, robot.api.logger lacks support.
shutil.rmtree(tmp_dir)
if sys.path[-1] == package_path:
sys.path.pop()
- return cls.cached_vpp_instance
def __enter__(self):
"""Create a tunnel, connect VPP instance.
run(["ssh", "-S", self._ssh_control_socket, "-O", "exit", "0.0.0.0"],
check=False)
shutil.rmtree(self._temp_dir)
- return
def add(self, csit_papi_command, history=True, **kwargs):
"""Add next command to internal command list; return self.
:rtype: PapiSocketExecutor
:raises RuntimeError: If unverified or conflicting CRC is encountered.
"""
+ self.crc_checker.report_initial_conflicts()
if history:
PapiHistory.add_to_papi_history(
self._node, csit_papi_command, **kwargs)
+ self.crc_checker.check_api_name(csit_papi_command)
self._api_command_list.append(
dict(api_name=csit_papi_command, api_args=copy.deepcopy(kwargs)))
return self
if not isinstance(reply, list):
reply = [reply]
for item in reply:
+ self.crc_checker.check_api_name(item.__class__.__name__)
dict_item = dictize(item)
if "retval" in dict_item.keys():
# *_details messages do not contain retval.
from robot.api import logger
+from resources.libraries.python.Constants import Constants
def _str(text):
"""Convert from possible unicode without interpreting as number.
so make sure the calling libraries have appropriate robot library scope.
For usual testing, it means "GLOBAL" scope."""
- def __init__(self, directory):
+ def __init__(self, directory,
+ fail_on_mismatch=Constants.CRC_MISMATCH_FAILS_TEST):
"""Initialize empty state, then register known collections.
This also scans directory for .api.json files
:type directory: str
"""
+ self.fail_on_mismatch = fail_on_mismatch
+ """If True, mismatch leads to test failure, by raising exception.
+ If False, the mismatch is logged, but the test is allowed to continue.
+ """
+
self._expected = dict()
"""Mapping from collection name to mapping from API name to CRC string.
self._register_all()
self._check_dir(directory)
+ def raise_or_log(self, exception):
+ """If fail_on_mismatch, raise, else log to console the exception.
+
+ :param exception: The exception to raise or log.
+ :type exception: RuntimeError
+ """
+ if self.fail_on_mismatch:
+ raise exception
+ logger.console("{exc!r}".format(exc=exception))
+
def _register_collection(self, collection_name, name_to_crc_mapping):
"""Add a named (copy of) collection of CRCs.
:param report_missing: Whether to raise on missing messages.
:type report_missing: bool
- :raises RuntimeError: If CRC mismatch or missing messages are detected.
+ :raises RuntimeError: If CRC mismatch or missing messages are detected,
+ and fail_on_mismatch is True.
"""
if self._initial_conflicts_reported:
return
self._initial_conflicts_reported = True
if self._reported:
- raise RuntimeError("Dir check found incompatible API CRCs: {rep!r}"\
- .format(rep=self._reported))
+ self.raise_or_log(
+ RuntimeError("Dir check found incompatible API CRCs: {rep!r}"\
+ .format(rep=self._reported)))
if not report_missing:
return
missing = {name: mapp for name, mapp in self._missing.items() if mapp}
if missing:
- raise RuntimeError("Dir check found missing API CRCs: {mis!r}"\
- .format(mis=missing))
+ self.raise_or_log(
+ RuntimeError("Dir check found missing API CRCs: {mis!r}"\
+ .format(mis=missing)))
def check_api_name(self, api_name):
"""Fail if the api_name has no known CRC associated.
return
crc = self._found.get(api_name, None)
self._reported[api_name] = crc
- # Disabled temporarily during CRC mismatch.
- #raise RuntimeError("No active collection has API {api!r}"
- # " CRC found {crc!r}".format(api=api_name, crc=crc))
+ self.raise_or_log(
+ RuntimeError("No active collection has API {api!r}"
+ " CRC found {crc!r}".format(api=api_name, crc=crc)))
Code Review / csit.git / commitdiff