diff options
author | 2019-12-03 15:18:03 -0600 | |
---|---|---|
committer | 2019-12-04 14:03:02 -0600 | |
commit | 079622e0b69ec1e5c50af08c8312937609929233 (patch) | |
tree | 617cad5ce13fa7f258cb054d5c0cf1cf37eb028c /ice_validator/preload/engine.py | |
parent | 74436fb509d03ea56feb50afa083c9c393fd95c5 (diff) |
[VVP] Support pluggable data sources for preload data
Change-Id: Ia7fcfa25203a93eac93381f472e3ba1e6c11235f
Issue-ID: VVP-339
Signed-off-by: Lovett, Trevor <trevor.lovett@att.com>
Diffstat (limited to 'ice_validator/preload/engine.py')
-rw-r--r-- | ice_validator/preload/engine.py | 114 |
1 files changed, 114 insertions, 0 deletions
diff --git a/ice_validator/preload/engine.py b/ice_validator/preload/engine.py new file mode 100644 index 0000000..488766d --- /dev/null +++ b/ice_validator/preload/engine.py @@ -0,0 +1,114 @@ +import importlib +import inspect +import os +import pkgutil +import shutil +from itertools import chain +from pathlib import Path +from typing import List, Type + +from preload.data import AbstractPreloadDataSource +from preload.generator import AbstractPreloadGenerator +from preload.model import get_heat_templates, Vnf +from tests.helpers import get_output_dir + + +def create_preloads(config, exitstatus): + """ + Create preloads in every format that can be discovered by get_generator_plugins + """ + if config.getoption("self_test"): + return + print("+===================================================================+") + print("| Preload Template Generation |") + print("+===================================================================+") + + preload_dir = os.path.join(get_output_dir(config), "preloads") + if os.path.exists(preload_dir): + shutil.rmtree(preload_dir) + plugins = PluginManager() + available_formats = [p.format_name() for p in plugins.preload_generators] + selected_formats = config.getoption("preload_formats") or available_formats + preload_source = None + if config.getoption("preload_source"): + preload_source_path = Path(config.getoption("preload_source")) + source_class = plugins.get_source_for_id( + config.getoption("preload_source_type") + ) + preload_source = source_class(preload_source_path) + + heat_templates = get_heat_templates(config) + vnf = None + for plugin_class in plugins.preload_generators: + if plugin_class.format_name() not in selected_formats: + continue + vnf = Vnf(heat_templates) + generator = plugin_class(vnf, preload_dir, preload_source) + generator.generate() + if vnf and vnf.uses_contrail: + print( + "\nWARNING: Preload template generation does not support Contrail\n" + "at this time, but Contrail resources were detected. The preload \n" + "template may be incomplete." + ) + if exitstatus != 0: + print( + "\nWARNING: Heat violations detected. Preload templates may be\n" + "incomplete or have errors." + ) + + +def is_implementation_of(class_, base_class): + """ + Returns True if the class is an implementation of AbstractPreloadGenerator + """ + return ( + inspect.isclass(class_) + and not inspect.isabstract(class_) + and issubclass(class_, base_class) + ) + + +def get_implementations_of(class_, modules): + """ + Returns all classes that implement ``class_`` from modules + """ + members = list( + chain.from_iterable( + inspect.getmembers(mod, lambda c: is_implementation_of(c, class_)) + for mod in modules + ) + ) + return [m[1] for m in members] + + +class PluginManager: + def __init__(self): + self.preload_plugins = [ + importlib.import_module(name) + for finder, name, ispkg in pkgutil.iter_modules() + if name.startswith("preload_") or name == "preload" + ] + self.preload_generators: List[ + Type[AbstractPreloadGenerator] + ] = get_implementations_of(AbstractPreloadGenerator, self.preload_plugins) + self.preload_sources: List[ + Type[AbstractPreloadDataSource] + ] = get_implementations_of(AbstractPreloadDataSource, self.preload_plugins) + + def get_source_for_id(self, identifier: str) -> Type[AbstractPreloadDataSource]: + for source in self.preload_sources: + if identifier == source.get_identifier(): + return source + raise RuntimeError( + "Unable to find preload source for identifier {}".format(identifier) + ) + + def get_source_for_name(self, name: str) -> Type[AbstractPreloadDataSource]: + for source in self.preload_sources: + if name == source.get_name(): + return source + raise RuntimeError("Unable to find preload source for name {}".format(name)) + + +PLUGIN_MGR = PluginManager() |