diff options
author | Klement Sekera <klement.sekera@gmail.com> | 2022-04-26 19:02:15 +0200 |
---|---|---|
committer | Ole Tr�an <otroan@employees.org> | 2022-05-10 18:52:08 +0000 |
commit | d9b0c6fbf7aa5bd9af84264105b39c82028a4a29 (patch) | |
tree | 4f786cfd8ebc2443cb11e11b74c8657204068898 /test/remote_test.py | |
parent | f90348bcb4afd0af2611cefc43b17ef3042b511c (diff) |
tests: replace pycodestyle with black
Drop pycodestyle for code style checking in favor of black. Black is
much faster, stable PEP8 compliant code style checker offering also
automatic formatting. It aims to be very stable and produce smallest
diffs. It's used by many small and big projects.
Running checkstyle with black takes a few seconds with a terse output.
Thus, test-checkstyle-diff is no longer necessary.
Expand scope of checkstyle to all python files in the repo, replacing
test-checkstyle with checkstyle-python.
Also, fixstyle-python is now available for automatic style formatting.
Note: python virtualenv has been consolidated in test/Makefile,
test/requirements*.txt which will eventually be moved to a central
location. This is required to simply the automated generation of
docker executor images in the CI.
Type: improvement
Change-Id: I022a326603485f58585e879ac0f697fceefbc9c8
Signed-off-by: Klement Sekera <klement.sekera@gmail.com>
Signed-off-by: Dave Wallace <dwallacelf@gmail.com>
Diffstat (limited to 'test/remote_test.py')
-rw-r--r-- | test/remote_test.py | 143 |
1 files changed, 70 insertions, 73 deletions
diff --git a/test/remote_test.py b/test/remote_test.py index 19bad897cdf..707d61fa9e5 100644 --- a/test/remote_test.py +++ b/test/remote_test.py @@ -16,10 +16,11 @@ 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__ + return "<SerializableClassCopy dict=%s>" % self.__dict__ class RemoteClassAttr: @@ -32,7 +33,7 @@ class RemoteClassAttr: 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()) @@ -44,25 +45,24 @@ class RemoteClassAttr: 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) + 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('__')): + 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) + 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) + return self._remote._remote_exec( + RemoteClass.CALL, self.path_to_str(), *args, **kwargs + ) class RemoteClass(Process): @@ -98,12 +98,12 @@ class RemoteClass(Process): 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 @@ -128,16 +128,16 @@ class RemoteClass(Process): 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__'): + 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) + 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('__')): + 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) @@ -149,13 +149,11 @@ class RemoteClass(Process): # 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 kwargs.items(): - if isinstance(val, RemoteClass) or \ - isinstance(val, RemoteClassAttr): + if isinstance(val, RemoteClass) or isinstance(val, RemoteClassAttr): kwargs[key] = val.get_remote_value() # send request args = self._make_serializable(args) @@ -163,11 +161,11 @@ class RemoteClass(Process): 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 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: @@ -222,7 +220,7 @@ class RemoteClass(Process): 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 @@ -243,7 +241,7 @@ class RemoteClass(Process): 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() @@ -255,12 +253,12 @@ class RemoteClass(Process): # 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__']: + if name[0] == "_": + if name in ["__weakref__"]: continue - if name in ['__dict__']: + if name in ["__dict__"]: continue - if not (name.startswith('__') and name.endswith('__')): + if not (name.startswith("__") and name.endswith("__")): continue if callable(member) and not isinstance(member, property): continue @@ -281,13 +279,13 @@ class RemoteClass(Process): if type(obj) is tuple: rv = tuple(rv) return rv - elif (isinstance(obj, IntEnum) or isinstance(obj, IntFlag)): + 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): @@ -307,19 +305,19 @@ class RemoteClass(Process): return self._deserialize_obj(obj) def start_remote(self): - """ Start remote execution """ + """Start remote execution""" self.start() def quit_remote(self): - """ Quit remote execution """ + """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): @@ -332,17 +330,16 @@ class RemoteClass(Process): 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: @@ -362,34 +359,34 @@ class RemoteClass(Process): @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 client connects to VPP - cls.remote_test = RemoteClass(RemoteVppTestCase) + @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() + # start remote process + cls.remote_test.start_remote() - # set up your test case - super(MyTestCase, cls).setUpClass() + # set up your test case + super(MyTestCase, cls).setUpClass() - # set up remote test - cls.remote_test.setUpClass(cls.tempdir) + # set up remote test + cls.remote_test.setUpClass(cls.tempdir) - @classmethod - def tearDownClass(cls): - # tear down remote test - cls.remote_test.tearDownClass() + @classmethod + def tearDownClass(cls): + # tear down remote test + cls.remote_test.tearDownClass() - # stop remote process - cls.remote_test.quit_remote() + # stop remote process + cls.remote_test.quit_remote() - # tear down your test case - super(MyTestCase, cls).tearDownClass() + # tear down your test case + super(MyTestCase, cls).tearDownClass() """ def __init__(self): @@ -408,10 +405,10 @@ class RemoteVppTestCase(VppTestCase): 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 @@ -422,7 +419,7 @@ class RemoteVppTestCase(VppTestCase): @unittest.skip("Empty test") def emptyTest(self): - """ Do nothing """ + """Do nothing""" pass def setTestFunctionInfo(self, name, doc): |