diff options
Diffstat (limited to 'test/asf/remote_test.py')
| -rw-r--r-- | test/asf/remote_test.py | 431 |
1 files changed, 431 insertions, 0 deletions
diff --git a/test/asf/remote_test.py b/test/asf/remote_test.py new file mode 100644 index 00000000000..7743c7782e4 --- /dev/null +++ b/test/asf/remote_test.py @@ -0,0 +1,431 @@ +#!/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 |