| |
|
|
| import importlib |
| import sys |
| import types |
|
|
|
|
| def get_func_name(func): |
| if isinstance(func, str): |
| return func |
| return '.'.join((func.__module__, func.__qualname__)) |
|
|
|
|
| def dummy_function_wrapper(func_name): |
| def dummy_function(*args, **kwargs): |
| raise RuntimeError('function {} no exist'.format(func_name)) |
|
|
| return dummy_function |
|
|
|
|
| class Patch: |
| def __init__(self, orig_func_name, new_func, create_dummy): |
| split_name = orig_func_name.rsplit('.', 1) |
| if len(split_name) == 1: |
| self.orig_module_name, self.orig_func_name = orig_func_name, None |
| else: |
| self.orig_module_name, self.orig_func_name = split_name |
| self.orig_module = None |
| self.orig_func = None |
|
|
| self.patch_func = None |
| self.wrappers = [] |
| if new_func is None: |
| new_func = dummy_function_wrapper(orig_func_name) |
| self.set_patch_func(new_func) |
| self.is_applied = False |
| self.create_dummy = create_dummy |
|
|
| @property |
| def orig_func_id(self): |
| return id(self.orig_func) |
|
|
| @property |
| def patch_func_id(self): |
| return id(self.patch_func) |
|
|
| def set_patch_func(self, new_func, force_patch=False): |
| if hasattr(new_func, '__name__') and new_func.__name__.endswith(('wrapper', 'decorator')): |
| self.wrappers.append(new_func) |
| else: |
| if self.patch_func and not force_patch: |
| raise RuntimeError('the patch of {} exist !'.format(self.orig_func_name)) |
| self.patch_func = new_func |
| self.is_applied = False |
|
|
| def apply_patch(self): |
| if self.is_applied: |
| return |
|
|
| self.orig_module, self.orig_func = Patch.parse_path(self.orig_module_name, self.orig_func_name, self.create_dummy) |
|
|
| final_patch_func = self.orig_func |
| if self.patch_func is not None: |
| final_patch_func = self.patch_func |
|
|
| for wrapper in self.wrappers: |
| final_patch_func = wrapper(final_patch_func) |
|
|
| if self.orig_func_name is not None: |
| setattr(self.orig_module, self.orig_func_name, final_patch_func) |
| for key, value in sys.modules.copy().items(): |
| if self.orig_func_name is not None and hasattr(value, self.orig_func_name) \ |
| and id(getattr(value, self.orig_func_name)) == self.orig_func_id: |
| setattr(value, self.orig_func_name, final_patch_func) |
| self.is_applied = True |
|
|
| @staticmethod |
| def parse_path(module_path, function_name, create_dummy): |
| from importlib.machinery import ModuleSpec |
| modules = module_path.split('.') |
| for i in range(1, len(modules) + 1): |
| parent = '.'.join(modules[:i - 1]) |
| path = '.'.join(modules[:i]) |
| try: |
| importlib.import_module(path) |
| except ModuleNotFoundError as e: |
| if not parent or not hasattr(importlib.import_module(parent), modules[i - 1]): |
| if not create_dummy: |
| raise ModuleNotFoundError(e) from e |
| sys.modules[path] = types.ModuleType(path) |
| sys.modules[path].__file__ = 'mindspeed.dummy_module.py' |
| sys.modules[path].__spec__ = ModuleSpec(path, None) |
| if parent: |
| setattr(importlib.import_module(parent), modules[i - 1], sys.modules[path]) |
| else: |
| module = getattr(importlib.import_module(parent), modules[i - 1]) |
| if hasattr(module, function_name): |
| return module, getattr(module, function_name) |
| elif create_dummy: |
| return module, dummy_function_wrapper(function_name) |
| else: |
| raise RuntimeError('no exist {} of {}'.format(function_name, module)) |
|
|
| if function_name is not None and not hasattr(sys.modules[module_path], function_name): |
| setattr(sys.modules[module_path], function_name, None) |
| return sys.modules[module_path], getattr(sys.modules[module_path], function_name) if function_name is not None else None |
|
|
|
|
| class MindSpeedPatchesManager: |
| patches_info = {} |
|
|
| @staticmethod |
| def register_patch(orig_func_name, new_func=None, force_patch=False, create_dummy=False): |
| if orig_func_name not in MindSpeedPatchesManager.patches_info: |
| MindSpeedPatchesManager.patches_info[orig_func_name] = Patch(orig_func_name, new_func, create_dummy) |
| else: |
| MindSpeedPatchesManager.patches_info.get(orig_func_name).set_patch_func(new_func, force_patch) |
|
|
| @staticmethod |
| def apply_patches(): |
| for patch in MindSpeedPatchesManager.patches_info.values(): |
| patch.apply_patch() |
|
|
