VersionableClass provides class hierarchy which automatically select right version of class. Class manipulation is used for this reason. Closer description is in autotest/client/shared/base_utils.py class VersionableClass Working example is in autotest/client/shared/base_utils_unittest.py pull-request: https://github.com/autotest/autotest/pull/519 Signed-off-by: Jiří Župka <jzupka@xxxxxxxxxx> --- client/shared/base_utils.py | 263 ++++++++++++++++++++++++++++++++++ client/shared/base_utils_unittest.py | 233 ++++++++++++++++++++++++++++++ 2 files changed, 496 insertions(+), 0 deletions(-) diff --git a/client/shared/base_utils.py b/client/shared/base_utils.py index b2064e1..e38baf6 100644 --- a/client/shared/base_utils.py +++ b/client/shared/base_utils.py @@ -2145,3 +2145,266 @@ def generate_random_string(length, ignore_str=string.punctuation, str += tmp length -= 1 return str + + +class VersionableClass(object): + """ + VersionableClass provides class hierarchy which automatically select right + version of class. Class manipulation is used for this reason. + By this reason is: + Advantage) Only one version is working in one process. Class is changed in + whole process. + Disadvantage) Only one version is working in one process. + + Example of usage (in base_utils_unittest): + + class FooC(object): + pass + + #Not implemented get_version -> not used for versioning. + class VCP(FooC, VersionableClass): + def __new__(cls, *args, **kargs): + VCP.master_class = VCP + return super(VCP, cls).__new__(cls, *args, **kargs) + + def foo(self): + pass + + class VC2(VCP, VersionableClass): + @staticmethod + def get_version(): + return "get_version_from_system" + + @classmethod + def is_right_version(cls, version): + if version is not None: + if "version is satisfied": + return True + return False + + def func1(self): + print "func1" + + def func2(self): + print "func2" + + # get_version could be inherited. + class VC3(VC2, VersionableClass): + @classmethod + def is_right_version(cls, version): + if version is not None: + if "version+1 is satisfied": + return True + return False + + def func2(self): + print "func2_2" + + class M(VCP): + pass + + m = M() # <- When class is constructed the right version is + # automatically selected. In this case VC3 is selected. + m.func2() # call VC3.func2(m) + m.func1() # call VC2.func1(m) + m.foo() # call VC1.foo(m) + + # When controlled "program" version is changed then is necessary call + check_repair_versions or recreate object. + + m.check_repair_versions() + + # priority of class. (change place where is method searched first in group + # of verisonable class.) + + class PP(VersionableClass): + def __new__(cls, *args, **kargs): + PP.master_class = PP + return super(PP, cls).__new__(cls, *args, **kargs) + + class PP2(PP, VersionableClass): + @staticmethod + def get_version(): + return "get_version_from_system" + + @classmethod + def is_right_version(cls, version): + if version is not None: + if "version is satisfied": + return True + return False + + def func1(self): + print "PP func1" + + class N(VCP, PP): + pass + + n = N() + + n.func1() # -> "func2" + + n.set_priority_class(PP, [VCP, PP]) + + n.func1() # -> "PP func1" + + Necessary for using: + 1) Subclass of versionable class must have implemented class methods + get_version and is_right_version. These two methods are necessary + for correct version section. Class without this method will be never + chosen like suitable class. + + 2) Every class derived from master_class have to add to class definition + inheritance from VersionableClass. Direct inheritance from Versionable + Class is use like a mark for manipulation with VersionableClass. + + 3) Master of VersionableClass have to defined class variable + cls.master_class. + """ + def __new__(cls, *args, **kargs): + cls.check_repair_versions() + return super(VersionableClass, cls).__new__(cls, *args, **kargs) + + #VersionableClass class management class. + + @classmethod + def check_repair_versions(cls, master_classes=None): + """ + Check version of versionable class and if version not + match repair version to correct version. + + @param master_classes: Check and repair only master_class. + @type master_classes: list. + """ + if master_classes is None: + master_classes = cls._find_versionable_baseclass() + for base in master_classes: + cls._check_repair_version_class(base) + + + @classmethod + def set_priority_class(cls, prioritized_class, group_classes): + """ + Set class priority. Limited only for change bases class priority inside + one subclass.__bases__ after that continue to another class. + """ + def change_position(ccls): + if not VersionableClass in ccls.__bases__: + bases = list(ccls.__bases__) + + index = None + remove_variant = None + for i, item in enumerate(ccls.__bases__): + if (VersionableClass in item.__bases__ and + item.master_class in group_classes): + if index is None: + index = i + if item.master_class is prioritized_class: + remove_variant = item + bases.remove(item) + break + else: + return + + bases.insert(index, remove_variant) + ccls.__bases__ = tuple(bases) + + def find_cls(ccls): + change_position(ccls) + for base in ccls.__bases__: + find_cls(base) + + find_cls(cls) + + + @classmethod + def _check_repair_version_class(cls, master_class): + version = None + for class_version in master_class._find_versionable_subclass(): + try: + version = class_version.get_version() + if class_version.is_right_version(version): + cls._switch_by_class(class_version) + break + except NotImplementedError: + continue + else: + cls._switch_by_class(class_version) + + + @classmethod + def _find_versionable_baseclass(cls): + """ + Find versionable class in master class. + """ + ver_class = [] + for superclass in cls.mro(): + if VersionableClass in superclass.__bases__: + ver_class.append(superclass.master_class) + + return set(ver_class) + + + @classmethod + def _find_versionable_subclass(cls): + """ + Find versionable subclasses which subclass.master_class == cls + """ + subclasses = [cls] + for sub in cls.__subclasses__(): + if VersionableClass in list(sub.__bases__): + subclasses.extend(sub._find_versionable_subclass()) + return subclasses + + + @classmethod + def _switch_by_class(cls, new_class): + """ + Finds all class with same master_class as new_class in class tree + and replaces them by new_class. + + @param new_class: Class for replacing. + """ + def find_replace_class(bases): + for base in bases: + if (VersionableClass in base.__bases__ and + base.master_class == new_class.master_class): + bnew = list(bases) + bnew[bnew.index(base)] = new_class + return tuple(bnew) + else: + bnew = find_replace_class(base.__bases__) + if bnew: + base.__bases__ = bnew + + bnew = find_replace_class(cls.__bases__) + if bnew: + cls.__bases__ = bnew + + + # Method defined in part below must be defined in + # verisonable class subclass. + + @classmethod + def get_version(cls): + """ + Get version of installed OpenVSwtich. + Must be re-implemented for in child class. + + @return: Version or None when get_version is unsuccessful. + """ + raise NotImplementedError("Method 'get_verison' must be" + " implemented in child class") + + + @classmethod + def is_right_version(cls, version): + """ + Check condition for select control class. + Function must be re-implemented in new OpenVSwitchControl class. + Must be re-implemented for in child class. + + @param version: version of OpenVSwtich + """ + raise NotImplementedError("Method 'is_right_version' must be" + " implemented in child class") diff --git a/client/shared/base_utils_unittest.py b/client/shared/base_utils_unittest.py index 604c441..8cfbb5e 100755 --- a/client/shared/base_utils_unittest.py +++ b/client/shared/base_utils_unittest.py @@ -834,5 +834,238 @@ class test_get_random_port(unittest.TestCase): self.assert_(s.getsockname()) +class test_VersionableClass(unittest.TestCase): + def setUp(self): + self.god = mock.mock_god(ut=self) + self.god.stub_function(base_utils.logging, 'warn') + self.god.stub_function(base_utils.logging, 'debug') + self.version = 1 + + + def tearDown(self): + self.god.unstub_all() + + + class FooC(object): + pass + + #Not implemented get_version -> not used for versioning. + class VCP(FooC, base_utils.VersionableClass): + def __new__(cls, *args, **kargs): + test_VersionableClass.VCP.version = 1 # Only for unittesting. + test_VersionableClass.VCP.master_class = test_VersionableClass.VCP + return (super(test_VersionableClass.VCP, cls) + .__new__(cls, *args, **kargs)) + + + def foo(self): + pass + + class VC2(VCP, base_utils.VersionableClass): + @classmethod + def get_version(cls): + return cls.version + + @classmethod + def is_right_version(cls, version): + if version is not None: + if version == 1: + return True + return False + + def func1(self): + logging.info("func1") + + def func2(self): + logging.info("func2") + + # get_version could be inherited. + class VC3(VC2, base_utils.VersionableClass): + @classmethod + def is_right_version(cls, version): + if version is not None: + if version == 2: + return True + return False + + def func2(self): + logging.info("func2_2") + + class PP(base_utils.VersionableClass): + def __new__(cls, *args, **kargs): + test_VersionableClass.PP.version = 1 # Only for unittesting. + test_VersionableClass.PP.master_class = test_VersionableClass.PP + return (super(test_VersionableClass.PP, cls) + .__new__(cls, *args, **kargs)) + + class PP2(PP, base_utils.VersionableClass): + @classmethod + def get_version(cls): + return cls.version + + @classmethod + def is_right_version(cls, version): + if version is not None: + if cls.version == 1: + return True + return False + + def func1(self): + print "PP func1" + + + class WP(base_utils.VersionableClass): + def __new__(cls, *args, **kargs): + test_VersionableClass.WP.version = 1 # Only for unittesting. + test_VersionableClass.WP.master_class = test_VersionableClass.WP + return (super(test_VersionableClass.WP, cls) + .__new__(cls, *args, **kargs)) + + class WP2(WP, base_utils.VersionableClass): + @classmethod + def get_version(cls): + return cls.version + + def func1(self): + print "WP func1" + + + class N(VCP, PP): + pass + + class NN(N): + pass + + class M(VCP): + pass + + class MM(M): + pass + + class W(WP): + pass + + + def test_simple_versioning(self): + self.god.stub_function(test_VersionableClass.VCP, "foo") + self.god.stub_function(test_VersionableClass.VC2, "func1") + self.god.stub_function(test_VersionableClass.VC2, "func2") + self.god.stub_function(test_VersionableClass.VC3, "func2") + + test_VersionableClass.VC2.func2.expect_call() + test_VersionableClass.VC2.func1.expect_call() + test_VersionableClass.VCP.foo.expect_call() + test_VersionableClass.VC3.func2.expect_call() + + test_VersionableClass.VC2.func2.expect_call() + test_VersionableClass.VC2.func1.expect_call() + test_VersionableClass.VCP.foo.expect_call() + test_VersionableClass.VC3.func2.expect_call() + + m = test_VersionableClass.M() + m.func2() # call VC3.func2(m) + m.func1() # call VC2.func1(m) + m.foo() # call VC1.foo(m) + m.version = 2 + m.check_repair_versions() + m.func2() + + #m.version = 1 + #m.check_repair_versions() + + mm = test_VersionableClass.MM() + mm.func2() # call VC3.func2(m) + mm.func1() # call VC2.func1(m) + mm.foo() # call VC1.foo(m) + mm.version = 2 + mm.check_repair_versions() + mm.func2() + + self.god.check_playback() + + def test_set_class_priority(self): + self.god.stub_function(test_VersionableClass.VC2, "func1") + self.god.stub_function(test_VersionableClass.VC2, "func2") + self.god.stub_function(test_VersionableClass.VC3, "func2") + self.god.stub_function(test_VersionableClass.PP2, "func1") + + test_VersionableClass.VC2.func1.expect_call() + test_VersionableClass.PP2.func1.expect_call() + test_VersionableClass.VC3.func2.expect_call() + test_VersionableClass.PP2.func1.expect_call() + test_VersionableClass.VC2.func1.expect_call() + test_VersionableClass.VC2.func2.expect_call() + + m = test_VersionableClass.N() + m.func1() + m.set_priority_class(test_VersionableClass.PP, + [test_VersionableClass.PP, + test_VersionableClass.VCP]) + m.func1() + + m.version = 2 + m.check_repair_versions() + m.func2() + m.func1() + + m.set_priority_class(test_VersionableClass.VCP, + [test_VersionableClass.PP, + test_VersionableClass.VCP]) + + m.func1() + + m.version = 1 + m.check_repair_versions() + m.func2() + + self.god.check_playback() + + + def test_set_class_priority_deep(self): + self.god.stub_function(test_VersionableClass.VC2, "func1") + self.god.stub_function(test_VersionableClass.VC2, "func2") + self.god.stub_function(test_VersionableClass.VC3, "func2") + self.god.stub_function(test_VersionableClass.PP2, "func1") + + test_VersionableClass.VC2.func1.expect_call() + test_VersionableClass.PP2.func1.expect_call() + test_VersionableClass.VC3.func2.expect_call() + test_VersionableClass.PP2.func1.expect_call() + test_VersionableClass.VC2.func1.expect_call() + test_VersionableClass.VC2.func2.expect_call() + + m = test_VersionableClass.NN() + m.func1() + m.set_priority_class(test_VersionableClass.PP, + [test_VersionableClass.PP, + test_VersionableClass.VCP]) + m.func1() + + m.version = 2 + m.check_repair_versions() + m.func2() + m.func1() + + m.set_priority_class(test_VersionableClass.VCP, + [test_VersionableClass.PP, + test_VersionableClass.VCP]) + + m.func1() + + m.version = 1 + m.check_repair_versions() + m.func2() + + self.god.check_playback() + + + def test_check_not_implemented(self): + m = test_VersionableClass.W() + self.assertEqual(m.__class__.__bases__, + tuple([test_VersionableClass.WP2]), + "Class should be WP2 (last defined class in class" + " hierarchy).") + + if __name__ == "__main__": unittest.main() -- 1.7.7.6 -- To unsubscribe from this list: send the line "unsubscribe kvm" in the body of a message to majordomo@xxxxxxxxxxxxxxx More majordomo info at http://vger.kernel.org/majordomo-info.html
