-#!/usr/bin/env python
+#!/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
-import six
-from six import moves
-
-from framework import VppTestCase
-from enum import Enum
+from enum import IntEnum, IntFlag
-class SerializableClassCopy(object):
+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(object):
+class RemoteClassAttr:
"""
Wrapper around attribute of a remotely executed class.
"""
self._remote = remote
def path_to_str(self):
- return '.'.join(self._path)
+ return ".".join(self._path)
def get_remote_value(self):
return self._remote._remote_exec(RemoteClass.GET, self.path_to_str())
return self._remote._remote_exec(RemoteClass.STR, self.path_to_str())
def __getattr__(self, attr):
- if attr[0] == '_':
- raise AttributeError
+ 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] == '_':
- super(RemoteClassAttr, self).__setattr__(attr, val)
- return
+ 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(),
- True, value=val)
+ 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(),
- True, *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()
+
+ #. 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
+ 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
self._pipe = Pipe() # pipe for input/output arguments
def __repr__(self):
- return moves.reprlib.repr(RemoteClassAttr(self, None))
+ return reprlib.repr(RemoteClassAttr(self, None))
def __str__(self):
return str(RemoteClassAttr(self, None))
return self.RemoteClassAttr(self, None)()
def __getattr__(self, attr):
- if attr[0] == '_' or not self.is_alive():
- if hasattr(super(RemoteClass, self), '__getattr__'):
- return super(RemoteClass, self).__getattr__(attr)
- raise AttributeError
+ 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():
- super(RemoteClass, self).__setattr__(attr, val)
- return
+ 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, ret=True, *args, **kwargs):
+ 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):
+ if isinstance(val, RemoteClass) or isinstance(val, RemoteClassAttr):
mutable_args[i] = val.get_remote_value()
args = tuple(mutable_args)
- for key, val in six.iteritems(kwargs):
- if isinstance(val, RemoteClass) or \
- isinstance(val, RemoteClassAttr):
+ 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))
- if not ret:
- # no return value expected
- return None
timeout = self._timeout
# adjust timeout specifically for the .sleep method
- if path.split('.')[-1] == 'sleep':
+ 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']
+ elif "timeout" in kwargs:
+ timeout += kwargs["timeout"]
if not self._pipe[RemoteClass.PIPE_PARENT].poll(timeout):
return None
try:
def _get_local_repr(self, path):
try:
obj = self._get_local_object(path)
- return moves.reprlib.repr(obj)
+ return reprlib.repr(obj)
except AttributeError:
return None
return None
def _serializable(self, obj):
- """ Test if the given object is serializable """
+ """Test if the given object is serializable"""
try:
dumps(obj)
return True
Dictionaries can hold complex values, so we split keys and values into
separate lists and serialize them individually.
"""
- if (type(obj) is dict):
+ if type(obj) is dict:
copy.type = type(obj)
copy.k_list = list()
copy.v_list = list()
# copy at least serializable attributes and properties
for name, member in inspect.getmembers(obj):
- if name[0] == '_': # skip private members
- continue
+ # 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):
- continue
+ member = self._make_serializable(member)
setattr(copy, name, member)
return copy
if type(obj) is tuple:
rv = tuple(rv)
return rv
- elif (isinstance(obj, Enum)):
+ 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 hasattr(obj, "type"):
if obj.type is dict:
_obj = dict()
for k, v in zip(obj.k_list, obj.v_list):
return self._deserialize_obj(obj)
def start_remote(self):
- """ Start remote execution """
+ """Start remote execution"""
self.start()
def quit_remote(self):
- """ Quit remote execution """
- self._remote_exec(RemoteClass.QUIT, None, False)
+ """Quit remote execution"""
+ self._remote_exec(RemoteClass.QUIT, None)
def get_remote_value(self):
- """ Get value of a remotely held object """
+ """Get value of a remotely held object"""
return RemoteClassAttr(self, None).get_remote_value()
def set_request_timeout(self, timeout):
- """ Change request timeout """
+ """Change request timeout"""
self._timeout = timeout
def run(self):
try:
rv = None
# get request from the parent process
- (op, path, args,
- kwargs) = self._pipe[RemoteClass.PIPE_CHILD].recv()
+ (op, path, args, kwargs) = self._pipe[RemoteClass.PIPE_CHILD].recv()
args = self._deserialize(args)
kwargs = self._deserialize(kwargs)
- path = path.split('.') if path else []
+ 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.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:
@unittest.skip("Remote Vpp Test Case Class")
class RemoteVppTestCase(VppTestCase):
- """ Re-use VppTestCase to create remote VPP segment
+ """Re-use VppTestCase to create remote VPP segment
- In your test case:
+ In your test case::
@classmethod
def setUpClass(cls):
- # fork new process before clinet connects to VPP
+ # fork new process before client connects to VPP
cls.remote_test = RemoteClass(RemoteVppTestCase)
# start remote process
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']
+ 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 """
+ """Do nothing"""
pass
def setTestFunctionInfo(self, name, doc):
Code Review / vpp.git / blobdiff