--- /dev/null
+#!/usr/bin/env python3
+
+import inspect
+import os
+import reprlib
+import unittest
+from framework import VppTestCase
+from multiprocessing import Process, Pipe
+from pickle import dumps
+
+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
Code Review / vpp.git / blobdiff