diff --git a/.github/workflows/other.yml b/.github/workflows/other.yml index 17f91991da5..7cf672afd14 100644 --- a/.github/workflows/other.yml +++ b/.github/workflows/other.yml @@ -49,13 +49,10 @@ jobs: if: ${{ matrix.python-version == '2.7' }} run: > pip install enum future mock pytest-coverage pytest-mock && - pytest - --cov=scripts --cov=ocaml/xcp-rrdd - scripts/ ocaml/xcp-rrdd -vv -rA - --junitxml=.git/pytest${{matrix.python-version}}.xml + pytest -vv -rA --cov=ocaml ocaml --cov-report term-missing --cov-report xml:.git/coverage${{matrix.python-version}}.xml - --cov-fail-under 0 + --cov-fail-under 60 env: PYTHONDEVMODE: yes diff --git a/ocaml/xapi-storage/python/xapi/__init__.py b/ocaml/xapi-storage/python/xapi/__init__.py index 57a7c0c9f2d..50eae33fe1a 100644 --- a/ocaml/xapi-storage/python/xapi/__init__.py +++ b/ocaml/xapi-storage/python/xapi/__init__.py @@ -31,11 +31,23 @@ import json import argparse -# pylint: disable=invalid-name,redefined-builtin,undefined-variable -# pyright: reportUndefinedVariable=false + +# is_str(): Shortcut to check if a value is an instance of a string type. +# +# Replace: +# if not isinstance(code, str) and not isinstance(code, unicode): +# with: +# if not is_str(code): +# +# This makes for much cleaner code and suits Python3 well too. if sys.version_info[0] > 2: long = int - unicode = str + def is_str(x): + return isinstance(x, str) # With Python3, all strings are unicode +else: + def is_str(x): # pragma: no cover + return isinstance(x, (str, unicode)) # pylint: disable=undefined-variable + def success(result): return {"Status": "Success", "Value": result} @@ -72,7 +84,7 @@ class XenAPIException(Exception): def __init__(self, code, params): Exception.__init__(self) - if not isinstance(code, str) and not isinstance(code, unicode): + if not is_str(code): raise TypeError("string", repr(code)) if not isinstance(params, list): raise TypeError("list", repr(params)) @@ -124,7 +136,7 @@ def __init__(self, thing, ty, desc): "UnmarshalException thing=%s ty=%s desc=%s" % (thing, ty, desc)) -class TypeError(InternalError): +class TypeError(InternalError): # pylint: disable=redefined-builtin def __init__(self, expected, actual): InternalError.__init__( diff --git a/ocaml/xapi-storage/python/xapi/storage/api/datapath.py b/ocaml/xapi-storage/python/xapi/storage/api/datapath.py index 69b37e5a9e7..957b8c0362f 100644 --- a/ocaml/xapi-storage/python/xapi/storage/api/datapath.py +++ b/ocaml/xapi-storage/python/xapi/storage/api/datapath.py @@ -1,11 +1,23 @@ from __future__ import print_function -from xapi import success, Rpc_light_failure, InternalError, UnmarshalException, TypeError, is_long, UnknownMethod -import xapi -import sys -import json + import argparse -import traceback +import json import logging +import sys +import traceback + +import xapi +# pylint: disable=line-too-long,superfluous-parens,unused-argument +# pylint: disable-next=redefined-builtin # FIXME: TypeError is a custom class in xapi +from xapi import ( + InternalError, + Rpc_light_failure, + TypeError, + UnknownMethod, + UnmarshalException, + is_str, + success, +) # pylint: disable=invalid-name,redefined-builtin,undefined-variable # pyright: reportUndefinedVariable=false @@ -15,7 +27,7 @@ class Unimplemented(Rpc_light_failure): def __init__(self, arg_0): Rpc_light_failure.__init__(self, "Unimplemented", [ arg_0 ]) - if not isinstance(arg_0, str) and not isinstance(arg_0, unicode): + if not is_str(arg_0): raise TypeError("string", repr(arg_0)) self.arg_0 = arg_0 class Datapath_server_dispatcher: @@ -30,12 +42,12 @@ def open(self, args): if not('dbg' in args): raise UnmarshalException('argument missing', 'dbg', '') dbg = args["dbg"] - if not isinstance(dbg, str) and not isinstance(dbg, unicode): + if not is_str(dbg): raise TypeError("string", repr(dbg)) if not('uri' in args): raise UnmarshalException('argument missing', 'uri', '') uri = args["uri"] - if not isinstance(uri, str) and not isinstance(uri, unicode): + if not is_str(uri): raise TypeError("string", repr(uri)) if not('persistent' in args): raise UnmarshalException('argument missing', 'persistent', '') @@ -51,29 +63,29 @@ def attach(self, args): if not('dbg' in args): raise UnmarshalException('argument missing', 'dbg', '') dbg = args["dbg"] - if not isinstance(dbg, str) and not isinstance(dbg, unicode): + if not is_str(dbg): raise TypeError("string", repr(dbg)) if not('uri' in args): raise UnmarshalException('argument missing', 'uri', '') uri = args["uri"] - if not isinstance(uri, str) and not isinstance(uri, unicode): + if not is_str(uri): raise TypeError("string", repr(uri)) if not('domain' in args): raise UnmarshalException('argument missing', 'domain', '') domain = args["domain"] - if not isinstance(domain, str) and not isinstance(domain, unicode): + if not is_str(domain): raise TypeError("string", repr(domain)) results = self._impl.attach(dbg, uri, domain) - if not isinstance(results['domain_uuid'], str) and not isinstance(results['domain_uuid'], unicode): + if not is_str(results['domain_uuid']): raise TypeError("string", repr(results['domain_uuid'])) if results['implementation'][0] == 'Blkback': - if not isinstance(results['implementation'][1], str) and not isinstance(results['implementation'][1], unicode): + if not is_str(results['implementation'][1]): raise TypeError("string", repr(results['implementation'][1])) elif results['implementation'][0] == 'Tapdisk3': - if not isinstance(results['implementation'][1], str) and not isinstance(results['implementation'][1], unicode): + if not is_str(results['implementation'][1]): raise TypeError("string", repr(results['implementation'][1])) elif results['implementation'][0] == 'Qdisk': - if not isinstance(results['implementation'][1], str) and not isinstance(results['implementation'][1], unicode): + if not is_str(results['implementation'][1]): raise TypeError("string", repr(results['implementation'][1])) return results def activate(self, args): @@ -83,17 +95,17 @@ def activate(self, args): if not('dbg' in args): raise UnmarshalException('argument missing', 'dbg', '') dbg = args["dbg"] - if not isinstance(dbg, str) and not isinstance(dbg, unicode): + if not is_str(dbg): raise TypeError("string", repr(dbg)) if not('uri' in args): raise UnmarshalException('argument missing', 'uri', '') uri = args["uri"] - if not isinstance(uri, str) and not isinstance(uri, unicode): + if not is_str(uri): raise TypeError("string", repr(uri)) if not('domain' in args): raise UnmarshalException('argument missing', 'domain', '') domain = args["domain"] - if not isinstance(domain, str) and not isinstance(domain, unicode): + if not is_str(domain): raise TypeError("string", repr(domain)) results = self._impl.activate(dbg, uri, domain) return results @@ -104,17 +116,17 @@ def deactivate(self, args): if not('dbg' in args): raise UnmarshalException('argument missing', 'dbg', '') dbg = args["dbg"] - if not isinstance(dbg, str) and not isinstance(dbg, unicode): + if not is_str(dbg): raise TypeError("string", repr(dbg)) if not('uri' in args): raise UnmarshalException('argument missing', 'uri', '') uri = args["uri"] - if not isinstance(uri, str) and not isinstance(uri, unicode): + if not is_str(uri): raise TypeError("string", repr(uri)) if not('domain' in args): raise UnmarshalException('argument missing', 'domain', '') domain = args["domain"] - if not isinstance(domain, str) and not isinstance(domain, unicode): + if not is_str(domain): raise TypeError("string", repr(domain)) results = self._impl.deactivate(dbg, uri, domain) return results @@ -125,17 +137,17 @@ def detach(self, args): if not('dbg' in args): raise UnmarshalException('argument missing', 'dbg', '') dbg = args["dbg"] - if not isinstance(dbg, str) and not isinstance(dbg, unicode): + if not is_str(dbg): raise TypeError("string", repr(dbg)) if not('uri' in args): raise UnmarshalException('argument missing', 'uri', '') uri = args["uri"] - if not isinstance(uri, str) and not isinstance(uri, unicode): + if not is_str(uri): raise TypeError("string", repr(uri)) if not('domain' in args): raise UnmarshalException('argument missing', 'domain', '') domain = args["domain"] - if not isinstance(domain, str) and not isinstance(domain, unicode): + if not is_str(domain): raise TypeError("string", repr(domain)) results = self._impl.detach(dbg, uri, domain) return results @@ -146,12 +158,12 @@ def close(self, args): if not('dbg' in args): raise UnmarshalException('argument missing', 'dbg', '') dbg = args["dbg"] - if not isinstance(dbg, str) and not isinstance(dbg, unicode): + if not is_str(dbg): raise TypeError("string", repr(dbg)) if not('uri' in args): raise UnmarshalException('argument missing', 'uri', '') uri = args["uri"] - if not isinstance(uri, str) and not isinstance(uri, unicode): + if not is_str(uri): raise TypeError("string", repr(uri)) results = self._impl.close(dbg, uri) return results @@ -193,7 +205,11 @@ def close(self, dbg, uri): """Xapi will call the functions here on VM start/shutdown/suspend/resume/migrate. Every function is idempotent. Every function takes a domain parameter which allows the implementation to track how many domains are currently using the volume.""" raise Unimplemented("Datapath.close") class Datapath_test: - """Xapi will call the functions here on VM start/shutdown/suspend/resume/migrate. Every function is idempotent. Every function takes a domain parameter which allows the implementation to track how many domains are currently using the volume.""" + """ + Xapi will call the functions here on VM start/shutdown/suspend/resume/migrate. + Every function is idempotent. Every function takes a domain parameter which allows + the implementation to track how many domains are currently using the volume. + """ def __init__(self): pass def open(self, dbg, uri, persistent): @@ -201,10 +217,27 @@ def open(self, dbg, uri, persistent): result = {} return result def attach(self, dbg, uri, domain): - """Xapi will call the functions here on VM start/shutdown/suspend/resume/migrate. Every function is idempotent. Every function takes a domain parameter which allows the implementation to track how many domains are currently using the volume.""" - result = {} - result["backend"] = { "domain_uuid": "string", "implementation": None } + # type:(str, str, str) -> dict[str, tuple[str, Any] | str] + """ + Return a valid results dictionary to Datapath_server_dispatcher.attach() + + The returned dict must contain the "domain_uuid" key with a string value. + The returned dict must contain the "implementation" key with two elements: + If the first element is one of "Blkback", "Tapdisk3" or "Qdisk", + the second element must be a string. Else, the dispatcher returns an error. + + See Datapath_server_dispatcher.attach() for the implementation details. + """ + # Fixed to not raise an internal error in Datapath_server_dispatcher.attach(): + result = { "domain_uuid": domain, "implementation": (uri, dbg) } + if not domain: # Provoke an internal error in the dispatcher to cover its code + result.pop("domain_uuid") # by removing the required "domain_uuid" key. + if domain == "5": + result["domain_uuid"] = 5 # Return an integer to provoke a type error. + if dbg == "inject_error" and uri in ["Blkback", "Tapdisk3", "Qdisk"]: + result["implementation"] = (uri, False) return result + def activate(self, dbg, uri, domain): """Xapi will call the functions here on VM start/shutdown/suspend/resume/migrate. Every function is idempotent. Every function takes a domain parameter which allows the implementation to track how many domains are currently using the volume.""" result = {} diff --git a/ocaml/xapi-storage/python/xapi/storage/api/plugin.py b/ocaml/xapi-storage/python/xapi/storage/api/plugin.py index 1b6d37214ca..d9199a98771 100644 --- a/ocaml/xapi-storage/python/xapi/storage/api/plugin.py +++ b/ocaml/xapi-storage/python/xapi/storage/api/plugin.py @@ -1,11 +1,23 @@ from __future__ import print_function -from xapi import success, Rpc_light_failure, InternalError, UnmarshalException, TypeError, is_long, UnknownMethod -import xapi -import sys -import json + import argparse -import traceback +import json import logging +import sys +import traceback + +import xapi +# pylint: disable=line-too-long,superfluous-parens,unused-argument +# pylint: disable-next=redefined-builtin # FIXME: TypeError is a custom class in xapi +from xapi import ( + InternalError, + Rpc_light_failure, + TypeError, + UnknownMethod, + UnmarshalException, + is_str, + success, +) # pylint: disable=invalid-name,redefined-builtin,undefined-variable # pyright: reportUndefinedVariable=false @@ -15,7 +27,7 @@ class Unimplemented(Rpc_light_failure): def __init__(self, arg_0): Rpc_light_failure.__init__(self, "Unimplemented", [ arg_0 ]) - if not isinstance(arg_0, str) and not isinstance(arg_0, unicode): + if not is_str(arg_0): raise TypeError("string", repr(arg_0)) self.arg_0 = arg_0 class Plugin_server_dispatcher: @@ -30,40 +42,40 @@ def query(self, args): if not('dbg' in args): raise UnmarshalException('argument missing', 'dbg', '') dbg = args["dbg"] - if not isinstance(dbg, str) and not isinstance(dbg, unicode): + if not is_str(dbg): raise TypeError("string", repr(dbg)) results = self._impl.query(dbg) - if not isinstance(results['plugin'], str) and not isinstance(results['plugin'], unicode): + if not is_str(results['plugin']): raise TypeError("string", repr(results['plugin'])) - if not isinstance(results['name'], str) and not isinstance(results['name'], unicode): + if not is_str(results['name']): raise TypeError("string", repr(results['name'])) - if not isinstance(results['description'], str) and not isinstance(results['description'], unicode): + if not is_str(results['description']): raise TypeError("string", repr(results['description'])) - if not isinstance(results['vendor'], str) and not isinstance(results['vendor'], unicode): + if not is_str(results['vendor']): raise TypeError("string", repr(results['vendor'])) - if not isinstance(results['copyright'], str) and not isinstance(results['copyright'], unicode): + if not is_str(results['copyright']): raise TypeError("string", repr(results['copyright'])) - if not isinstance(results['version'], str) and not isinstance(results['version'], unicode): + if not is_str(results['version']): raise TypeError("string", repr(results['version'])) - if not isinstance(results['required_api_version'], str) and not isinstance(results['required_api_version'], unicode): + if not is_str(results['required_api_version']): raise TypeError("string", repr(results['required_api_version'])) if not isinstance(results['features'], list): raise TypeError("string list", repr(results['features'])) for tmp_1 in results['features']: - if not isinstance(tmp_1, str) and not isinstance(tmp_1, unicode): + if not is_str(tmp_1): raise TypeError("string", repr(tmp_1)) if not isinstance(results['configuration'], dict): raise TypeError("(string * string) list", repr(results['configuration'])) for tmp_2 in results['configuration'].keys(): - if not isinstance(tmp_2, str) and not isinstance(tmp_2, unicode): + if not is_str(tmp_2): raise TypeError("string", repr(tmp_2)) for tmp_2 in results['configuration'].values(): - if not isinstance(tmp_2, str) and not isinstance(tmp_2, unicode): + if not is_str(tmp_2): raise TypeError("string", repr(tmp_2)) if not isinstance(results['required_cluster_stack'], list): raise TypeError("string list", repr(results['required_cluster_stack'])) for tmp_3 in results['required_cluster_stack']: - if not isinstance(tmp_3, str) and not isinstance(tmp_3, unicode): + if not is_str(tmp_3): raise TypeError("string", repr(tmp_3)) return results def ls(self, args): @@ -73,13 +85,13 @@ def ls(self, args): if not('dbg' in args): raise UnmarshalException('argument missing', 'dbg', '') dbg = args["dbg"] - if not isinstance(dbg, str) and not isinstance(dbg, unicode): + if not is_str(dbg): raise TypeError("string", repr(dbg)) results = self._impl.ls(dbg) if not isinstance(results, list): raise TypeError("string list", repr(results)) for tmp_4 in results: - if not isinstance(tmp_4, str) and not isinstance(tmp_4, unicode): + if not is_str(tmp_4): raise TypeError("string", repr(tmp_4)) return results def diagnostics(self, args): @@ -89,10 +101,10 @@ def diagnostics(self, args): if not('dbg' in args): raise UnmarshalException('argument missing', 'dbg', '') dbg = args["dbg"] - if not isinstance(dbg, str) and not isinstance(dbg, unicode): + if not is_str(dbg): raise TypeError("string", repr(dbg)) results = self._impl.diagnostics(dbg) - if not isinstance(results, str) and not isinstance(results, unicode): + if not is_str(results): raise TypeError("string", repr(results)) return results def _dispatch(self, method, params): diff --git a/ocaml/xapi-storage/python/xapi/storage/api/test_datapath.py b/ocaml/xapi-storage/python/xapi/storage/api/test_datapath.py new file mode 100644 index 00000000000..8b6436657da --- /dev/null +++ b/ocaml/xapi-storage/python/xapi/storage/api/test_datapath.py @@ -0,0 +1,138 @@ +import logging + +import pytest + +import xapi +import xapi.storage.api.datapath + + +def internal_error(error): + """Return a dictionary with an internal error""" + return {"ErrorDescription": ["Internal_error", error], "Status": "Failure"} + + +def assert_error(testee, caplog, method_args, method, error): + """Assert that the result of the testee matches the expected error result""" + args = method_args.copy() + if method != "open": # the persistent arg is only checked for the open method + args["persistent"] = None # pass it, but with a wrong type(not used/checked) + assert testee._dispatch("Datapath." + method, [args]) == internal_error(error) + assert caplog.messages[0] == "caught " + error + caplog.clear() + + +def assert_type_checks(testee, methods, template_args, bad_args, caplog): + """Assert that the result of the testee matches the expected result""" + for arg in bad_args: + # Sigh, if Python would be strongly typed, we wouldn't need this: + # Assert the type checks of the arguments + expected = "bool" if arg == "persistent" else "string" + other_type = False if expected == "string" else "str" + for actual in [None, [], (), {"dict": "val"}, 1, 1.0, str, caplog, other_type]: + bad_args = template_args.copy() + bad_args[arg] = actual + error_msg = "TypeError expected={} actual={}".format(expected, repr(actual)) + for method in methods: + assert_error(testee, caplog, bad_args, method, error_msg) + + # Remove the argument and assert the missing argument checks + bad_args.pop(arg) + error_msg = "UnmarshalException thing=argument missing ty={} desc=".format(arg) + for method in methods: + assert_error(testee, caplog, bad_args, method, error_msg) + + +def assert_attach_type_check(testee, caplog, args, uri): + """Assert that the result of the testee matches the expected result""" + a = args.copy() + a["uri"] = uri + assert testee._dispatch("Datapath.attach", [a]) == { + "Status": "Success", + "Value": {"domain_uuid": a["domain"], "implementation": (uri, a["dbg"])}, + } + if uri == "other": + return + a["dbg"] = "inject_error" + assert_error(testee, caplog, a, "attach", "TypeError expected=string actual=False") + + +def assert_attach_type_checks(testee, caplog, args): + """Assert type checks when attach() returns Blkback, Tapdisk3, Qdisk and others""" + for uri in ["Blkback", "Tapdisk3", "Qdisk", "other"]: + assert_attach_type_check(testee, caplog, args, uri) + + +def test_dispatcher(caplog, capsys): + """ + Test the dispatcher of the Xapi storage API datapath interface + + The dispatcher is a class that routes the calls to the corresponding methods + of a given Datapath implementation class. + """ + # Setup + caplog.set_level(logging.INFO) + + # The testee passes them to the Datapath_test class and its attach method + # is expected to return the values which we use to test the dispatcher: + args = {"dbg": "", "uri": "uri", "domain": "uuid", "persistent": True} + + # Call + + # datapath_server_test() returns an instance of the dispatcher class that + # routes the calls to the corresponding methods of the Datapath_test class: + testee = xapi.storage.api.datapath.datapath_server_test() + + # Test the argument checks of the dispatcher to identify missing arguments: + + # Assert type checks on the dbg and uri arguments + missing = ["dbg", "uri"] + methods = ["attach", "activate", "deactivate", "detach", "open", "close"] + assert_type_checks(testee, methods, args, missing, caplog) + + # Assert type checks on the missing domain argument + missing = ["domain"] + methods = ["attach", "activate", "deactivate", "detach"] + assert_type_checks(testee, methods, args, missing, caplog) + + # Assert type checks on the persistent flag for the open method + missing = ["persistent"] + methods = ["open"] + assert_type_checks(testee, methods, args, missing, caplog) + + # Assert the dispatcher returns the example results of Datapath_test.attach(): + assert_attach_type_checks(testee, caplog, args) + + # Assert the internal error to cover the check by removing the domain argument: + bad = args.copy() + bad["domain"] = "" + assert_error(testee, caplog, bad, "attach", "'domain_uuid'") + # Assert the type check on the domain_uuid return value: + bad["domain"] = "5" + assert_error(testee, caplog, bad, "attach", "TypeError expected=string actual=5") + + # The other methods work as expected. Setup, Call, Assert: + success = {"Status": "Success", "Value": {}} + assert testee._dispatch("Datapath.open", [args]) == success + assert testee._dispatch("Datapath.activate", [args]) == success + assert testee._dispatch("Datapath.deactivate", [args]) == success + assert testee._dispatch("Datapath.detach", [args]) == success + assert testee._dispatch("Datapath.close", [args]) == success + + # Assert that no errors were logged and no output was printed: + assert caplog.messages == [] # No messages were logged + assert capsys.readouterr().out == "" # No output was printed + assert capsys.readouterr().err == "" # No errors were printed + + +def test_exceptions(): + """Cover the code changed by using the is_str() function""" + + with pytest.raises(xapi.TypeError) as exc_info: + _ = xapi.XenAPIException(1, "params") # pylint: disable=pointless-statement + assert str(exc_info.value) == "TypeError expected=string actual=1" + + with pytest.raises(xapi.TypeError) as exc_info: + _ = xapi.storage.api.datapath.Unimplemented( + False + ) # pylint: disable=pointless-statement + assert str(exc_info.value) == "TypeError expected=string actual=False" diff --git a/ocaml/xapi-storage/python/xapi/storage/api/volume.py b/ocaml/xapi-storage/python/xapi/storage/api/volume.py index b89574f9570..0f01ed6fd97 100644 --- a/ocaml/xapi-storage/python/xapi/storage/api/volume.py +++ b/ocaml/xapi-storage/python/xapi/storage/api/volume.py @@ -1,11 +1,24 @@ from __future__ import print_function -from xapi import success, Rpc_light_failure, InternalError, UnmarshalException, TypeError, is_long, UnknownMethod -import xapi -import sys -import json + import argparse -import traceback +import json import logging +import sys +import traceback + +import xapi +# pylint: disable=line-too-long,superfluous-parens,unused-argument +# pylint: disable-next=redefined-builtin # FIXME: TypeError is a custom class in xapi +from xapi import ( + InternalError, + Rpc_light_failure, + TypeError, + UnknownMethod, + UnmarshalException, + is_long, + is_str, + success, +) # pylint: disable=invalid-name,redefined-builtin,undefined-variable # pyright: reportUndefinedVariable=false @@ -17,31 +30,31 @@ class Sr_not_attached(Rpc_light_failure): def __init__(self, arg_0): Rpc_light_failure.__init__(self, "Sr_not_attached", [ arg_0 ]) - if not isinstance(arg_0, str) and not isinstance(arg_0, unicode): + if not is_str(arg_0): raise TypeError("string", repr(arg_0)) self.arg_0 = arg_0 class SR_does_not_exist(Rpc_light_failure): def __init__(self, arg_0): Rpc_light_failure.__init__(self, "SR_does_not_exist", [ arg_0 ]) - if not isinstance(arg_0, str) and not isinstance(arg_0, unicode): + if not is_str(arg_0): raise TypeError("string", repr(arg_0)) self.arg_0 = arg_0 class Volume_does_not_exist(Rpc_light_failure): def __init__(self, arg_0): Rpc_light_failure.__init__(self, "Volume_does_not_exist", [ arg_0 ]) - if not isinstance(arg_0, str) and not isinstance(arg_0, unicode): + if not is_str(arg_0): raise TypeError("string", repr(arg_0)) self.arg_0 = arg_0 class Unimplemented(Rpc_light_failure): def __init__(self, arg_0): Rpc_light_failure.__init__(self, "Unimplemented", [ arg_0 ]) - if not isinstance(arg_0, str) and not isinstance(arg_0, unicode): + if not is_str(arg_0): raise TypeError("string", repr(arg_0)) self.arg_0 = arg_0 class Cancelled(Rpc_light_failure): def __init__(self, arg_0): Rpc_light_failure.__init__(self, "Cancelled", [ arg_0 ]) - if not isinstance(arg_0, str) and not isinstance(arg_0, unicode): + if not is_str(arg_0): raise TypeError("string", repr(arg_0)) self.arg_0 = arg_0 class Volume_server_dispatcher: @@ -56,22 +69,22 @@ def create(self, args): if not('dbg' in args): raise UnmarshalException('argument missing', 'dbg', '') dbg = args["dbg"] - if not isinstance(dbg, str) and not isinstance(dbg, unicode): + if not is_str(dbg): raise TypeError("string", repr(dbg)) if not('sr' in args): raise UnmarshalException('argument missing', 'sr', '') sr = args["sr"] - if not isinstance(sr, str) and not isinstance(sr, unicode): + if not is_str(sr): raise TypeError("string", repr(sr)) if not('name' in args): raise UnmarshalException('argument missing', 'name', '') name = args["name"] - if not isinstance(name, str) and not isinstance(name, unicode): + if not is_str(name): raise TypeError("string", repr(name)) if not('description' in args): raise UnmarshalException('argument missing', 'description', '') description = args["description"] - if not isinstance(description, str) and not isinstance(description, unicode): + if not is_str(description): raise TypeError("string", repr(description)) if not('size' in args): raise UnmarshalException('argument missing', 'size', '') @@ -79,14 +92,14 @@ def create(self, args): if not(is_long(size)): raise TypeError("int64", repr(size)) results = self._impl.create(dbg, sr, name, description, size) - if not isinstance(results['key'], str) and not isinstance(results['key'], unicode): + if not is_str(results['key']): raise TypeError("string", repr(results['key'])) if results['uuid'] is not None: - if not isinstance(results['uuid'], str) and not isinstance(results['uuid'], unicode): + if not is_str(results['uuid']): raise TypeError("string", repr(results['uuid'])) - if not isinstance(results['name'], str) and not isinstance(results['name'], unicode): + if not is_str(results['name']): raise TypeError("string", repr(results['name'])) - if not isinstance(results['description'], str) and not isinstance(results['description'], unicode): + if not is_str(results['description']): raise TypeError("string", repr(results['description'])) if not isinstance(results['read_write'], bool): raise TypeError("bool", repr(results['read_write'])) @@ -97,15 +110,15 @@ def create(self, args): if not isinstance(results['uri'], list): raise TypeError("string list", repr(results['uri'])) for tmp_5 in results['uri']: - if not isinstance(tmp_5, str) and not isinstance(tmp_5, unicode): + if not is_str(tmp_5): raise TypeError("string", repr(tmp_5)) if not isinstance(results['keys'], dict): raise TypeError("(string * string) list", repr(results['keys'])) for tmp_6 in results['keys'].keys(): - if not isinstance(tmp_6, str) and not isinstance(tmp_6, unicode): + if not is_str(tmp_6): raise TypeError("string", repr(tmp_6)) for tmp_6 in results['keys'].values(): - if not isinstance(tmp_6, str) and not isinstance(tmp_6, unicode): + if not is_str(tmp_6): raise TypeError("string", repr(tmp_6)) return results def snapshot(self, args): @@ -115,27 +128,27 @@ def snapshot(self, args): if not('dbg' in args): raise UnmarshalException('argument missing', 'dbg', '') dbg = args["dbg"] - if not isinstance(dbg, str) and not isinstance(dbg, unicode): + if not is_str(dbg): raise TypeError("string", repr(dbg)) if not('sr' in args): raise UnmarshalException('argument missing', 'sr', '') sr = args["sr"] - if not isinstance(sr, str) and not isinstance(sr, unicode): + if not is_str(sr): raise TypeError("string", repr(sr)) if not('key' in args): raise UnmarshalException('argument missing', 'key', '') key = args["key"] - if not isinstance(key, str) and not isinstance(key, unicode): + if not is_str(key): raise TypeError("string", repr(key)) results = self._impl.snapshot(dbg, sr, key) - if not isinstance(results['key'], str) and not isinstance(results['key'], unicode): + if not is_str(results['key']): raise TypeError("string", repr(results['key'])) if results['uuid'] is not None: - if not isinstance(results['uuid'], str) and not isinstance(results['uuid'], unicode): + if not is_str(results['uuid']): raise TypeError("string", repr(results['uuid'])) - if not isinstance(results['name'], str) and not isinstance(results['name'], unicode): + if not is_str(results['name']): raise TypeError("string", repr(results['name'])) - if not isinstance(results['description'], str) and not isinstance(results['description'], unicode): + if not is_str(results['description']): raise TypeError("string", repr(results['description'])) if not isinstance(results['read_write'], bool): raise TypeError("bool", repr(results['read_write'])) @@ -146,15 +159,15 @@ def snapshot(self, args): if not isinstance(results['uri'], list): raise TypeError("string list", repr(results['uri'])) for tmp_7 in results['uri']: - if not isinstance(tmp_7, str) and not isinstance(tmp_7, unicode): + if not is_str(tmp_7): raise TypeError("string", repr(tmp_7)) if not isinstance(results['keys'], dict): raise TypeError("(string * string) list", repr(results['keys'])) for tmp_8 in results['keys'].keys(): - if not isinstance(tmp_8, str) and not isinstance(tmp_8, unicode): + if not is_str(tmp_8): raise TypeError("string", repr(tmp_8)) for tmp_8 in results['keys'].values(): - if not isinstance(tmp_8, str) and not isinstance(tmp_8, unicode): + if not is_str(tmp_8): raise TypeError("string", repr(tmp_8)) return results def clone(self, args): @@ -164,27 +177,27 @@ def clone(self, args): if not('dbg' in args): raise UnmarshalException('argument missing', 'dbg', '') dbg = args["dbg"] - if not isinstance(dbg, str) and not isinstance(dbg, unicode): + if not is_str(dbg): raise TypeError("string", repr(dbg)) if not('sr' in args): raise UnmarshalException('argument missing', 'sr', '') sr = args["sr"] - if not isinstance(sr, str) and not isinstance(sr, unicode): + if not is_str(sr): raise TypeError("string", repr(sr)) if not('key' in args): raise UnmarshalException('argument missing', 'key', '') key = args["key"] - if not isinstance(key, str) and not isinstance(key, unicode): + if not is_str(key): raise TypeError("string", repr(key)) results = self._impl.clone(dbg, sr, key) - if not isinstance(results['key'], str) and not isinstance(results['key'], unicode): + if not is_str(results['key']): raise TypeError("string", repr(results['key'])) if results['uuid'] is not None: - if not isinstance(results['uuid'], str) and not isinstance(results['uuid'], unicode): + if not is_str(results['uuid']): raise TypeError("string", repr(results['uuid'])) - if not isinstance(results['name'], str) and not isinstance(results['name'], unicode): + if not is_str(results['name']): raise TypeError("string", repr(results['name'])) - if not isinstance(results['description'], str) and not isinstance(results['description'], unicode): + if not is_str(results['description']): raise TypeError("string", repr(results['description'])) if not isinstance(results['read_write'], bool): raise TypeError("bool", repr(results['read_write'])) @@ -195,15 +208,15 @@ def clone(self, args): if not isinstance(results['uri'], list): raise TypeError("string list", repr(results['uri'])) for tmp_9 in results['uri']: - if not isinstance(tmp_9, str) and not isinstance(tmp_9, unicode): + if not is_str(tmp_9): raise TypeError("string", repr(tmp_9)) if not isinstance(results['keys'], dict): raise TypeError("(string * string) list", repr(results['keys'])) for tmp_10 in results['keys'].keys(): - if not isinstance(tmp_10, str) and not isinstance(tmp_10, unicode): + if not is_str(tmp_10): raise TypeError("string", repr(tmp_10)) for tmp_10 in results['keys'].values(): - if not isinstance(tmp_10, str) and not isinstance(tmp_10, unicode): + if not is_str(tmp_10): raise TypeError("string", repr(tmp_10)) return results def destroy(self, args): @@ -213,17 +226,17 @@ def destroy(self, args): if not('dbg' in args): raise UnmarshalException('argument missing', 'dbg', '') dbg = args["dbg"] - if not isinstance(dbg, str) and not isinstance(dbg, unicode): + if not is_str(dbg): raise TypeError("string", repr(dbg)) if not('sr' in args): raise UnmarshalException('argument missing', 'sr', '') sr = args["sr"] - if not isinstance(sr, str) and not isinstance(sr, unicode): + if not is_str(sr): raise TypeError("string", repr(sr)) if not('key' in args): raise UnmarshalException('argument missing', 'key', '') key = args["key"] - if not isinstance(key, str) and not isinstance(key, unicode): + if not is_str(key): raise TypeError("string", repr(key)) results = self._impl.destroy(dbg, sr, key) return results @@ -234,22 +247,22 @@ def set_name(self, args): if not('dbg' in args): raise UnmarshalException('argument missing', 'dbg', '') dbg = args["dbg"] - if not isinstance(dbg, str) and not isinstance(dbg, unicode): + if not is_str(dbg): raise TypeError("string", repr(dbg)) if not('sr' in args): raise UnmarshalException('argument missing', 'sr', '') sr = args["sr"] - if not isinstance(sr, str) and not isinstance(sr, unicode): + if not is_str(sr): raise TypeError("string", repr(sr)) if not('key' in args): raise UnmarshalException('argument missing', 'key', '') key = args["key"] - if not isinstance(key, str) and not isinstance(key, unicode): + if not is_str(key): raise TypeError("string", repr(key)) if not('new_name' in args): raise UnmarshalException('argument missing', 'new_name', '') new_name = args["new_name"] - if not isinstance(new_name, str) and not isinstance(new_name, unicode): + if not is_str(new_name): raise TypeError("string", repr(new_name)) results = self._impl.set_name(dbg, sr, key, new_name) return results @@ -260,22 +273,22 @@ def set_description(self, args): if not('dbg' in args): raise UnmarshalException('argument missing', 'dbg', '') dbg = args["dbg"] - if not isinstance(dbg, str) and not isinstance(dbg, unicode): + if not is_str(dbg): raise TypeError("string", repr(dbg)) if not('sr' in args): raise UnmarshalException('argument missing', 'sr', '') sr = args["sr"] - if not isinstance(sr, str) and not isinstance(sr, unicode): + if not is_str(sr): raise TypeError("string", repr(sr)) if not('key' in args): raise UnmarshalException('argument missing', 'key', '') key = args["key"] - if not isinstance(key, str) and not isinstance(key, unicode): + if not is_str(key): raise TypeError("string", repr(key)) if not('new_description' in args): raise UnmarshalException('argument missing', 'new_description', '') new_description = args["new_description"] - if not isinstance(new_description, str) and not isinstance(new_description, unicode): + if not is_str(new_description): raise TypeError("string", repr(new_description)) results = self._impl.set_description(dbg, sr, key, new_description) return results @@ -286,27 +299,27 @@ def set(self, args): if not('dbg' in args): raise UnmarshalException('argument missing', 'dbg', '') dbg = args["dbg"] - if not isinstance(dbg, str) and not isinstance(dbg, unicode): + if not is_str(dbg): raise TypeError("string", repr(dbg)) if not('sr' in args): raise UnmarshalException('argument missing', 'sr', '') sr = args["sr"] - if not isinstance(sr, str) and not isinstance(sr, unicode): + if not is_str(sr): raise TypeError("string", repr(sr)) if not('key' in args): raise UnmarshalException('argument missing', 'key', '') key = args["key"] - if not isinstance(key, str) and not isinstance(key, unicode): + if not is_str(key): raise TypeError("string", repr(key)) if not('k' in args): raise UnmarshalException('argument missing', 'k', '') k = args["k"] - if not isinstance(k, str) and not isinstance(k, unicode): + if not is_str(k): raise TypeError("string", repr(k)) if not('v' in args): raise UnmarshalException('argument missing', 'v', '') v = args["v"] - if not isinstance(v, str) and not isinstance(v, unicode): + if not is_str(v): raise TypeError("string", repr(v)) results = self._impl.set(dbg, sr, key, k, v) return results @@ -317,22 +330,22 @@ def unset(self, args): if not('dbg' in args): raise UnmarshalException('argument missing', 'dbg', '') dbg = args["dbg"] - if not isinstance(dbg, str) and not isinstance(dbg, unicode): + if not is_str(dbg): raise TypeError("string", repr(dbg)) if not('sr' in args): raise UnmarshalException('argument missing', 'sr', '') sr = args["sr"] - if not isinstance(sr, str) and not isinstance(sr, unicode): + if not is_str(sr): raise TypeError("string", repr(sr)) if not('key' in args): raise UnmarshalException('argument missing', 'key', '') key = args["key"] - if not isinstance(key, str) and not isinstance(key, unicode): + if not is_str(key): raise TypeError("string", repr(key)) if not('k' in args): raise UnmarshalException('argument missing', 'k', '') k = args["k"] - if not isinstance(k, str) and not isinstance(k, unicode): + if not is_str(k): raise TypeError("string", repr(k)) results = self._impl.unset(dbg, sr, key, k) return results @@ -343,17 +356,17 @@ def resize(self, args): if not('dbg' in args): raise UnmarshalException('argument missing', 'dbg', '') dbg = args["dbg"] - if not isinstance(dbg, str) and not isinstance(dbg, unicode): + if not is_str(dbg): raise TypeError("string", repr(dbg)) if not('sr' in args): raise UnmarshalException('argument missing', 'sr', '') sr = args["sr"] - if not isinstance(sr, str) and not isinstance(sr, unicode): + if not is_str(sr): raise TypeError("string", repr(sr)) if not('key' in args): raise UnmarshalException('argument missing', 'key', '') key = args["key"] - if not isinstance(key, str) and not isinstance(key, unicode): + if not is_str(key): raise TypeError("string", repr(key)) if not('new_size' in args): raise UnmarshalException('argument missing', 'new_size', '') @@ -369,27 +382,27 @@ def stat(self, args): if not('dbg' in args): raise UnmarshalException('argument missing', 'dbg', '') dbg = args["dbg"] - if not isinstance(dbg, str) and not isinstance(dbg, unicode): + if not is_str(dbg): raise TypeError("string", repr(dbg)) if not('sr' in args): raise UnmarshalException('argument missing', 'sr', '') sr = args["sr"] - if not isinstance(sr, str) and not isinstance(sr, unicode): + if not is_str(sr): raise TypeError("string", repr(sr)) if not('key' in args): raise UnmarshalException('argument missing', 'key', '') key = args["key"] - if not isinstance(key, str) and not isinstance(key, unicode): + if not is_str(key): raise TypeError("string", repr(key)) results = self._impl.stat(dbg, sr, key) - if not isinstance(results['key'], str) and not isinstance(results['key'], unicode): + if not is_str(results['key']): raise TypeError("string", repr(results['key'])) if results['uuid'] is not None: - if not isinstance(results['uuid'], str) and not isinstance(results['uuid'], unicode): + if not is_str(results['uuid']): raise TypeError("string", repr(results['uuid'])) - if not isinstance(results['name'], str) and not isinstance(results['name'], unicode): + if not is_str(results['name']): raise TypeError("string", repr(results['name'])) - if not isinstance(results['description'], str) and not isinstance(results['description'], unicode): + if not is_str(results['description']): raise TypeError("string", repr(results['description'])) if not isinstance(results['read_write'], bool): raise TypeError("bool", repr(results['read_write'])) @@ -400,15 +413,15 @@ def stat(self, args): if not isinstance(results['uri'], list): raise TypeError("string list", repr(results['uri'])) for tmp_11 in results['uri']: - if not isinstance(tmp_11, str) and not isinstance(tmp_11, unicode): + if not is_str(tmp_11): raise TypeError("string", repr(tmp_11)) if not isinstance(results['keys'], dict): raise TypeError("(string * string) list", repr(results['keys'])) for tmp_12 in results['keys'].keys(): - if not isinstance(tmp_12, str) and not isinstance(tmp_12, unicode): + if not is_str(tmp_12): raise TypeError("string", repr(tmp_12)) for tmp_12 in results['keys'].values(): - if not isinstance(tmp_12, str) and not isinstance(tmp_12, unicode): + if not is_str(tmp_12): raise TypeError("string", repr(tmp_12)) return results def _dispatch(self, method, params): @@ -801,22 +814,22 @@ def probe(self, args): if not('dbg' in args): raise UnmarshalException('argument missing', 'dbg', '') dbg = args["dbg"] - if not isinstance(dbg, str) and not isinstance(dbg, unicode): + if not is_str(dbg): raise TypeError("string", repr(dbg)) if not('uri' in args): raise UnmarshalException('argument missing', 'uri', '') uri = args["uri"] - if not isinstance(uri, str) and not isinstance(uri, unicode): + if not is_str(uri): raise TypeError("string", repr(uri)) results = self._impl.probe(dbg, uri) if not isinstance(results['srs'], list): raise TypeError("7 list", repr(results['srs'])) for tmp_13 in results['srs']: - if not isinstance(tmp_13['sr'], str) and not isinstance(tmp_13['sr'], unicode): + if not is_str(tmp_13['sr']): raise TypeError("string", repr(tmp_13['sr'])) - if not isinstance(tmp_13['name'], str) and not isinstance(tmp_13['name'], unicode): + if not is_str(tmp_13['name']): raise TypeError("string", repr(tmp_13['name'])) - if not isinstance(tmp_13['description'], str) and not isinstance(tmp_13['description'], unicode): + if not is_str(tmp_13['description']): raise TypeError("string", repr(tmp_13['description'])) if not(is_long(tmp_13['free_space'])): raise TypeError("int64", repr(tmp_13['free_space'])) @@ -825,20 +838,20 @@ def probe(self, args): if not isinstance(tmp_13['datasources'], list): raise TypeError("string list", repr(tmp_13['datasources'])) for tmp_14 in tmp_13['datasources']: - if not isinstance(tmp_14, str) and not isinstance(tmp_14, unicode): + if not is_str(tmp_14): raise TypeError("string", repr(tmp_14)) if not isinstance(tmp_13['clustered'], bool): raise TypeError("bool", repr(tmp_13['clustered'])) if tmp_13['health'][0] == 'Healthy': - if not isinstance(tmp_13['health'][1], str) and not isinstance(tmp_13['health'][1], unicode): + if not is_str(tmp_13['health'][1]): raise TypeError("string", repr(tmp_13['health'][1])) elif tmp_13['health'][0] == 'Recovering': - if not isinstance(tmp_13['health'][1], str) and not isinstance(tmp_13['health'][1], unicode): + if not is_str(tmp_13['health'][1]): raise TypeError("string", repr(tmp_13['health'][1])) if not isinstance(results['uris'], list): raise TypeError("string list", repr(results['uris'])) for tmp_15 in results['uris']: - if not isinstance(tmp_15, str) and not isinstance(tmp_15, unicode): + if not is_str(tmp_15): raise TypeError("string", repr(tmp_15)) return results def create(self, args): @@ -848,22 +861,22 @@ def create(self, args): if not('dbg' in args): raise UnmarshalException('argument missing', 'dbg', '') dbg = args["dbg"] - if not isinstance(dbg, str) and not isinstance(dbg, unicode): + if not is_str(dbg): raise TypeError("string", repr(dbg)) if not('uri' in args): raise UnmarshalException('argument missing', 'uri', '') uri = args["uri"] - if not isinstance(uri, str) and not isinstance(uri, unicode): + if not is_str(uri): raise TypeError("string", repr(uri)) if not('name' in args): raise UnmarshalException('argument missing', 'name', '') name = args["name"] - if not isinstance(name, str) and not isinstance(name, unicode): + if not is_str(name): raise TypeError("string", repr(name)) if not('description' in args): raise UnmarshalException('argument missing', 'description', '') description = args["description"] - if not isinstance(description, str) and not isinstance(description, unicode): + if not is_str(description): raise TypeError("string", repr(description)) if not('configuration' in args): raise UnmarshalException('argument missing', 'configuration', '') @@ -871,10 +884,10 @@ def create(self, args): if not isinstance(configuration, dict): raise TypeError("(string * string) list", repr(configuration)) for tmp_16 in configuration.keys(): - if not isinstance(tmp_16, str) and not isinstance(tmp_16, unicode): + if not is_str(tmp_16): raise TypeError("string", repr(tmp_16)) for tmp_16 in configuration.values(): - if not isinstance(tmp_16, str) and not isinstance(tmp_16, unicode): + if not is_str(tmp_16): raise TypeError("string", repr(tmp_16)) results = self._impl.create(dbg, uri, name, description, configuration) return results @@ -885,15 +898,15 @@ def attach(self, args): if not('dbg' in args): raise UnmarshalException('argument missing', 'dbg', '') dbg = args["dbg"] - if not isinstance(dbg, str) and not isinstance(dbg, unicode): + if not is_str(dbg): raise TypeError("string", repr(dbg)) if not('uri' in args): raise UnmarshalException('argument missing', 'uri', '') uri = args["uri"] - if not isinstance(uri, str) and not isinstance(uri, unicode): + if not is_str(uri): raise TypeError("string", repr(uri)) results = self._impl.attach(dbg, uri) - if not isinstance(results, str) and not isinstance(results, unicode): + if not is_str(results): raise TypeError("string", repr(results)) return results def detach(self, args): @@ -903,12 +916,12 @@ def detach(self, args): if not('dbg' in args): raise UnmarshalException('argument missing', 'dbg', '') dbg = args["dbg"] - if not isinstance(dbg, str) and not isinstance(dbg, unicode): + if not is_str(dbg): raise TypeError("string", repr(dbg)) if not('sr' in args): raise UnmarshalException('argument missing', 'sr', '') sr = args["sr"] - if not isinstance(sr, str) and not isinstance(sr, unicode): + if not is_str(sr): raise TypeError("string", repr(sr)) results = self._impl.detach(dbg, sr) return results @@ -919,12 +932,12 @@ def destroy(self, args): if not('dbg' in args): raise UnmarshalException('argument missing', 'dbg', '') dbg = args["dbg"] - if not isinstance(dbg, str) and not isinstance(dbg, unicode): + if not is_str(dbg): raise TypeError("string", repr(dbg)) if not('sr' in args): raise UnmarshalException('argument missing', 'sr', '') sr = args["sr"] - if not isinstance(sr, str) and not isinstance(sr, unicode): + if not is_str(sr): raise TypeError("string", repr(sr)) results = self._impl.destroy(dbg, sr) return results @@ -935,19 +948,19 @@ def stat(self, args): if not('dbg' in args): raise UnmarshalException('argument missing', 'dbg', '') dbg = args["dbg"] - if not isinstance(dbg, str) and not isinstance(dbg, unicode): + if not is_str(dbg): raise TypeError("string", repr(dbg)) if not('sr' in args): raise UnmarshalException('argument missing', 'sr', '') sr = args["sr"] - if not isinstance(sr, str) and not isinstance(sr, unicode): + if not is_str(sr): raise TypeError("string", repr(sr)) results = self._impl.stat(dbg, sr) - if not isinstance(results['sr'], str) and not isinstance(results['sr'], unicode): + if not is_str(results['sr']): raise TypeError("string", repr(results['sr'])) - if not isinstance(results['name'], str) and not isinstance(results['name'], unicode): + if not is_str(results['name']): raise TypeError("string", repr(results['name'])) - if not isinstance(results['description'], str) and not isinstance(results['description'], unicode): + if not is_str(results['description']): raise TypeError("string", repr(results['description'])) if not(is_long(results['free_space'])): raise TypeError("int64", repr(results['free_space'])) @@ -956,15 +969,15 @@ def stat(self, args): if not isinstance(results['datasources'], list): raise TypeError("string list", repr(results['datasources'])) for tmp_17 in results['datasources']: - if not isinstance(tmp_17, str) and not isinstance(tmp_17, unicode): + if not is_str(tmp_17): raise TypeError("string", repr(tmp_17)) if not isinstance(results['clustered'], bool): raise TypeError("bool", repr(results['clustered'])) if results['health'][0] == 'Healthy': - if not isinstance(results['health'][1], str) and not isinstance(results['health'][1], unicode): + if not is_str(results['health'][1]): raise TypeError("string", repr(results['health'][1])) elif results['health'][0] == 'Recovering': - if not isinstance(results['health'][1], str) and not isinstance(results['health'][1], unicode): + if not is_str(results['health'][1]): raise TypeError("string", repr(results['health'][1])) return results def set_name(self, args): @@ -974,17 +987,17 @@ def set_name(self, args): if not('dbg' in args): raise UnmarshalException('argument missing', 'dbg', '') dbg = args["dbg"] - if not isinstance(dbg, str) and not isinstance(dbg, unicode): + if not is_str(dbg): raise TypeError("string", repr(dbg)) if not('sr' in args): raise UnmarshalException('argument missing', 'sr', '') sr = args["sr"] - if not isinstance(sr, str) and not isinstance(sr, unicode): + if not is_str(sr): raise TypeError("string", repr(sr)) if not('new_name' in args): raise UnmarshalException('argument missing', 'new_name', '') new_name = args["new_name"] - if not isinstance(new_name, str) and not isinstance(new_name, unicode): + if not is_str(new_name): raise TypeError("string", repr(new_name)) results = self._impl.set_name(dbg, sr, new_name) return results @@ -995,17 +1008,17 @@ def set_description(self, args): if not('dbg' in args): raise UnmarshalException('argument missing', 'dbg', '') dbg = args["dbg"] - if not isinstance(dbg, str) and not isinstance(dbg, unicode): + if not is_str(dbg): raise TypeError("string", repr(dbg)) if not('sr' in args): raise UnmarshalException('argument missing', 'sr', '') sr = args["sr"] - if not isinstance(sr, str) and not isinstance(sr, unicode): + if not is_str(sr): raise TypeError("string", repr(sr)) if not('new_description' in args): raise UnmarshalException('argument missing', 'new_description', '') new_description = args["new_description"] - if not isinstance(new_description, str) and not isinstance(new_description, unicode): + if not is_str(new_description): raise TypeError("string", repr(new_description)) results = self._impl.set_description(dbg, sr, new_description) return results @@ -1016,25 +1029,25 @@ def ls(self, args): if not('dbg' in args): raise UnmarshalException('argument missing', 'dbg', '') dbg = args["dbg"] - if not isinstance(dbg, str) and not isinstance(dbg, unicode): + if not is_str(dbg): raise TypeError("string", repr(dbg)) if not('sr' in args): raise UnmarshalException('argument missing', 'sr', '') sr = args["sr"] - if not isinstance(sr, str) and not isinstance(sr, unicode): + if not is_str(sr): raise TypeError("string", repr(sr)) results = self._impl.ls(dbg, sr) if not isinstance(results, list): raise TypeError("8 list", repr(results)) for tmp_18 in results: - if not isinstance(tmp_18['key'], str) and not isinstance(tmp_18['key'], unicode): + if not is_str(tmp_18['key']): raise TypeError("string", repr(tmp_18['key'])) if tmp_18['uuid'] is not None: - if not isinstance(tmp_18['uuid'], str) and not isinstance(tmp_18['uuid'], unicode): + if not is_str(tmp_18['uuid']): raise TypeError("string", repr(tmp_18['uuid'])) - if not isinstance(tmp_18['name'], str) and not isinstance(tmp_18['name'], unicode): + if not is_str(tmp_18['name']): raise TypeError("string", repr(tmp_18['name'])) - if not isinstance(tmp_18['description'], str) and not isinstance(tmp_18['description'], unicode): + if not is_str(tmp_18['description']): raise TypeError("string", repr(tmp_18['description'])) if not isinstance(tmp_18['read_write'], bool): raise TypeError("bool", repr(tmp_18['read_write'])) @@ -1045,15 +1058,15 @@ def ls(self, args): if not isinstance(tmp_18['uri'], list): raise TypeError("string list", repr(tmp_18['uri'])) for tmp_19 in tmp_18['uri']: - if not isinstance(tmp_19, str) and not isinstance(tmp_19, unicode): + if not is_str(tmp_19): raise TypeError("string", repr(tmp_19)) if not isinstance(tmp_18['keys'], dict): raise TypeError("(string * string) list", repr(tmp_18['keys'])) for tmp_20 in tmp_18['keys'].keys(): - if not isinstance(tmp_20, str) and not isinstance(tmp_20, unicode): + if not is_str(tmp_20): raise TypeError("string", repr(tmp_20)) for tmp_20 in tmp_18['keys'].values(): - if not isinstance(tmp_20, str) and not isinstance(tmp_20, unicode): + if not is_str(tmp_20): raise TypeError("string", repr(tmp_20)) return results def _dispatch(self, method, params): diff --git a/pyproject.toml b/pyproject.toml index efdcd13494e..8f1b5931255 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -195,8 +195,10 @@ disable = [ "no-else-break", # else clause following a break statement "protected-access", # Best done during the code cleanup phase "super-with-arguments", # Consider using Python 3 style super(no args) calls + "too-few-public-methods", # Some classes only overload private methods, is fine "too-many-branches", # Existing code breaches this, not part of porting "too-many-arguments", # Likewise, not part of porting + "too-many-lines", # Likewise, not part of porting "too-many-locals", # Likewise, not part of porting "too-many-statements", # Likewise, not part of porting "unnecessary-pass", # Cosmetic, best done during the code cleanup phase @@ -258,7 +260,7 @@ addopts = "-v -ra" # xfail_strict: require to remove pytext.xfail marker when test is fixed # required_plugins: require that these plugins are installed before testing # ----------------------------------------------------------------------------- -testpaths = ["python3", "scripts", "ocaml/xcp-rrdd"] +testpaths = ["python3", "ocaml/xcp-rrdd", "ocaml/xapi-storage"] required_plugins = ["pytest-cov", "pytest-mock"] log_cli_level = "INFO" log_cli = true