diff options
Diffstat (limited to 'test/asf/remote_test.py')
| -rw-r--r-- | test/asf/remote_test.py | 431 |
1 files changed, 0 insertions, 431 deletions
diff --git a/test/asf/remote_test.py b/test/asf/remote_test.py deleted file mode 100644 index 7743c7782e4..00000000000 --- a/test/asf/remote_test.py +++ /dev/null @@ -1,431 +0,0 @@ -#!/usr/bin/env python3 - -import inspect -import os -import reprlib -import unittest -from asfframework import VppTestCase -from multiprocessing import Process, Pipe -from pickle import dumps -import sys - -from enum import IntEnum, IntFlag - - -class SerializableClassCopy: - """ - Empty class used as a basis for a serializable copy of another class. - """ - - pass - - def __repr__(self): - return "<SerializableClassCopy dict=%s>" % self.__dict__ - - -class RemoteClassAttr: - """ - Wrapper around attribute of a remotely executed class. - """ - - def __init__(self, remote, attr): - self._path = [attr] if attr else [] - self._remote = remote - - def path_to_str(self): - return ".".join(self._path) - - def get_remote_value(self): - return self._remote._remote_exec(RemoteClass.GET, self.path_to_str()) - - def __repr__(self): - return self._remote._remote_exec(RemoteClass.REPR, self.path_to_str()) - - def __str__(self): - return self._remote._remote_exec(RemoteClass.STR, self.path_to_str()) - - def __getattr__(self, attr): - if attr[0] == "_": - if not (attr.startswith("__") and attr.endswith("__")): - raise AttributeError("tried to get private attribute: %s ", attr) - self._path.append(attr) - return self - - def __setattr__(self, attr, val): - if attr[0] == "_": - if not (attr.startswith("__") and attr.endswith("__")): - super(RemoteClassAttr, self).__setattr__(attr, val) - return - self._path.append(attr) - self._remote._remote_exec(RemoteClass.SETATTR, self.path_to_str(), value=val) - - def __call__(self, *args, **kwargs): - return self._remote._remote_exec( - RemoteClass.CALL, self.path_to_str(), *args, **kwargs - ) - - -class RemoteClass(Process): - """ - This class can wrap around and adapt the interface of another class, - and then delegate its execution to a newly forked child process. - - Usage: - - #. Create a remotely executed instance of MyClass. :: - - object = RemoteClass(MyClass, arg1='foo', arg2='bar') - object.start_remote() - - #. Access the object normally as if it was an instance of your - class. :: - - object.my_attribute = 20 - print object.my_attribute - print object.my_method(object.my_attribute) - object.my_attribute.nested_attribute = 'test' - - #. If you need the value of a remote attribute, use .get_remote_value - method. This method is automatically called when needed in the - context of a remotely executed class. E.g. :: - - if (object.my_attribute.get_remote_value() > 20): - object.my_attribute2 = object.my_attribute - - #. Destroy the instance. :: - - object.quit_remote() - object.terminate() - """ - - GET = 0 # Get attribute remotely - CALL = 1 # Call method remotely - SETATTR = 2 # Set attribute remotely - REPR = 3 # Get representation of a remote object - STR = 4 # Get string representation of a remote object - QUIT = 5 # Quit remote execution - - PIPE_PARENT = 0 # Parent end of the pipe - PIPE_CHILD = 1 # Child end of the pipe - - DEFAULT_TIMEOUT = 2 # default timeout for an operation to execute - - def __init__(self, cls, *args, **kwargs): - super(RemoteClass, self).__init__() - self._cls = cls - self._args = args - self._kwargs = kwargs - self._timeout = RemoteClass.DEFAULT_TIMEOUT - self._pipe = Pipe() # pipe for input/output arguments - - def __repr__(self): - return reprlib.repr(RemoteClassAttr(self, None)) - - def __str__(self): - return str(RemoteClassAttr(self, None)) - - def __call__(self, *args, **kwargs): - return self.RemoteClassAttr(self, None)() - - def __getattr__(self, attr): - if attr[0] == "_" or not self.is_alive(): - if not (attr.startswith("__") and attr.endswith("__")): - if hasattr(super(RemoteClass, self), "__getattr__"): - return super(RemoteClass, self).__getattr__(attr) - raise AttributeError("missing: %s", attr) - return RemoteClassAttr(self, attr) - - def __setattr__(self, attr, val): - if attr[0] == "_" or not self.is_alive(): - if not (attr.startswith("__") and attr.endswith("__")): - super(RemoteClass, self).__setattr__(attr, val) - return - setattr(RemoteClassAttr(self, None), attr, val) - - def _remote_exec(self, op, path=None, *args, **kwargs): - """ - Execute given operation on a given, possibly nested, member remotely. - """ - # automatically resolve remote objects in the arguments - mutable_args = list(args) - for i, val in enumerate(mutable_args): - if isinstance(val, RemoteClass) or isinstance(val, RemoteClassAttr): - mutable_args[i] = val.get_remote_value() - args = tuple(mutable_args) - for key, val in kwargs.items(): - if isinstance(val, RemoteClass) or isinstance(val, RemoteClassAttr): - kwargs[key] = val.get_remote_value() - # send request - args = self._make_serializable(args) - kwargs = self._make_serializable(kwargs) - self._pipe[RemoteClass.PIPE_PARENT].send((op, path, args, kwargs)) - timeout = self._timeout - # adjust timeout specifically for the .sleep method - if path is not None and path.split(".")[-1] == "sleep": - if args and isinstance(args[0], (long, int)): - timeout += args[0] - elif "timeout" in kwargs: - timeout += kwargs["timeout"] - if not self._pipe[RemoteClass.PIPE_PARENT].poll(timeout): - return None - try: - rv = self._pipe[RemoteClass.PIPE_PARENT].recv() - rv = self._deserialize(rv) - return rv - except EOFError: - return None - - def _get_local_object(self, path): - """ - Follow the path to obtain a reference on the addressed nested attribute - """ - obj = self._instance - for attr in path: - obj = getattr(obj, attr) - return obj - - def _get_local_value(self, path): - try: - return self._get_local_object(path) - except AttributeError: - return None - - def _call_local_method(self, path, *args, **kwargs): - try: - method = self._get_local_object(path) - return method(*args, **kwargs) - except AttributeError: - return None - - def _set_local_attr(self, path, value): - try: - obj = self._get_local_object(path[:-1]) - setattr(obj, path[-1], value) - except AttributeError: - pass - return None - - def _get_local_repr(self, path): - try: - obj = self._get_local_object(path) - return reprlib.repr(obj) - except AttributeError: - return None - - def _get_local_str(self, path): - try: - obj = self._get_local_object(path) - return str(obj) - except AttributeError: - return None - - def _serializable(self, obj): - """Test if the given object is serializable""" - try: - dumps(obj) - return True - except: - return False - - def _make_obj_serializable(self, obj): - """ - Make a serializable copy of an object. - Members which are difficult/impossible to serialize are stripped. - """ - if self._serializable(obj): - return obj # already serializable - - copy = SerializableClassCopy() - - """ - Dictionaries can hold complex values, so we split keys and values into - separate lists and serialize them individually. - """ - if type(obj) is dict: - copy.type = type(obj) - copy.k_list = list() - copy.v_list = list() - for k, v in obj.items(): - copy.k_list.append(self._make_serializable(k)) - copy.v_list.append(self._make_serializable(v)) - return copy - - # copy at least serializable attributes and properties - for name, member in inspect.getmembers(obj): - # skip private members and non-writable dunder methods. - if name[0] == "_": - if name in ["__weakref__"]: - continue - if name in ["__dict__"]: - continue - if not (name.startswith("__") and name.endswith("__")): - continue - if callable(member) and not isinstance(member, property): - continue - if not self._serializable(member): - member = self._make_serializable(member) - setattr(copy, name, member) - return copy - - def _make_serializable(self, obj): - """ - Make a serializable copy of an object or a list/tuple of objects. - Members which are difficult/impossible to serialize are stripped. - """ - if (type(obj) is list) or (type(obj) is tuple): - rv = [] - for item in obj: - rv.append(self._make_serializable(item)) - if type(obj) is tuple: - rv = tuple(rv) - return rv - elif isinstance(obj, IntEnum) or isinstance(obj, IntFlag): - return obj.value - else: - return self._make_obj_serializable(obj) - - def _deserialize_obj(self, obj): - if hasattr(obj, "type"): - if obj.type is dict: - _obj = dict() - for k, v in zip(obj.k_list, obj.v_list): - _obj[self._deserialize(k)] = self._deserialize(v) - return _obj - return obj - - def _deserialize(self, obj): - if (type(obj) is list) or (type(obj) is tuple): - rv = [] - for item in obj: - rv.append(self._deserialize(item)) - if type(obj) is tuple: - rv = tuple(rv) - return rv - else: - return self._deserialize_obj(obj) - - def start_remote(self): - """Start remote execution""" - self.start() - - def quit_remote(self): - """Quit remote execution""" - self._remote_exec(RemoteClass.QUIT, None) - - def get_remote_value(self): - """Get value of a remotely held object""" - return RemoteClassAttr(self, None).get_remote_value() - - def set_request_timeout(self, timeout): - """Change request timeout""" - self._timeout = timeout - - def run(self): - """ - Create instance of the wrapped class and execute operations - on it as requested by the parent process. - """ - self._instance = self._cls(*self._args, **self._kwargs) - while True: - try: - rv = None - # get request from the parent process - (op, path, args, kwargs) = self._pipe[RemoteClass.PIPE_CHILD].recv() - args = self._deserialize(args) - kwargs = self._deserialize(kwargs) - path = path.split(".") if path else [] - if op == RemoteClass.GET: - rv = self._get_local_value(path) - elif op == RemoteClass.CALL: - rv = self._call_local_method(path, *args, **kwargs) - elif op == RemoteClass.SETATTR and "value" in kwargs: - self._set_local_attr(path, kwargs["value"]) - elif op == RemoteClass.REPR: - rv = self._get_local_repr(path) - elif op == RemoteClass.STR: - rv = self._get_local_str(path) - elif op == RemoteClass.QUIT: - break - else: - continue - # send return value - if not self._serializable(rv): - rv = self._make_serializable(rv) - self._pipe[RemoteClass.PIPE_CHILD].send(rv) - except EOFError: - break - self._instance = None # destroy the instance - - -@unittest.skip("Remote Vpp Test Case Class") -class RemoteVppTestCase(VppTestCase): - """Re-use VppTestCase to create remote VPP segment - - In your test case:: - - @classmethod - def setUpClass(cls): - # fork new process before client connects to VPP - cls.remote_test = RemoteClass(RemoteVppTestCase) - - # start remote process - cls.remote_test.start_remote() - - # set up your test case - super(MyTestCase, cls).setUpClass() - - # set up remote test - cls.remote_test.setUpClass(cls.tempdir) - - @classmethod - def tearDownClass(cls): - # tear down remote test - cls.remote_test.tearDownClass() - - # stop remote process - cls.remote_test.quit_remote() - - # tear down your test case - super(MyTestCase, cls).tearDownClass() - """ - - def __init__(self): - super(RemoteVppTestCase, self).__init__("emptyTest") - - # Note: __del__ is a 'Finalizer" not a 'Destructor'. - # https://docs.python.org/3/reference/datamodel.html#object.__del__ - def __del__(self): - if hasattr(self, "vpp"): - self.vpp.poll() - if self.vpp.returncode is None: - self.vpp.terminate() - self.vpp.communicate() - - @classmethod - def setUpClass(cls, tempdir): - # disable features unsupported in remote VPP - orig_env = dict(os.environ) - if "STEP" in os.environ: - del os.environ["STEP"] - if "DEBUG" in os.environ: - del os.environ["DEBUG"] - cls.tempdir_prefix = os.path.basename(tempdir) + "/" - super(RemoteVppTestCase, cls).setUpClass() - os.environ = orig_env - - @classmethod - def tearDownClass(cls): - super(RemoteVppTestCase, cls).tearDownClass() - - @unittest.skip("Empty test") - def emptyTest(self): - """Do nothing""" - pass - - def setTestFunctionInfo(self, name, doc): - """ - Store the name and documentation string of currently executed test - in the main VPP for logging purposes. - """ - self._testMethodName = name - self._testMethodDoc = doc |