6 from itertools import chain
7 from pathlib import Path
8 from typing import List, Type
10 from preload.data import AbstractPreloadDataSource
11 from preload.generator import AbstractPreloadGenerator
12 from preload.model import get_heat_templates, Vnf
13 from tests.helpers import get_output_dir
16 def create_preloads(config, exitstatus):
18 Create preloads in every format that can be discovered by get_generator_plugins
20 if config.getoption("self_test"):
22 print("+===================================================================+")
23 print("| Preload Template Generation |")
24 print("+===================================================================+")
26 preload_dir = os.path.join(get_output_dir(config), "preloads")
27 if os.path.exists(preload_dir):
28 shutil.rmtree(preload_dir)
29 plugins = PluginManager()
30 available_formats = [p.format_name() for p in plugins.preload_generators]
31 selected_formats = config.getoption("preload_formats") or available_formats
33 if config.getoption("preload_source"):
34 preload_source_path = Path(config.getoption("preload_source"))
35 source_class = plugins.get_source_for_id(
36 config.getoption("preload_source_type")
38 preload_source = source_class(preload_source_path)
40 heat_templates = get_heat_templates(config)
42 for plugin_class in plugins.preload_generators:
43 if plugin_class.format_name() not in selected_formats:
45 vnf = Vnf(heat_templates)
46 generator = plugin_class(vnf, preload_dir, preload_source)
48 if vnf and vnf.uses_contrail:
50 "\nWARNING: Preload template generation does not support Contrail\n"
51 "at this time, but Contrail resources were detected. The preload \n"
52 "template may be incomplete."
56 "\nWARNING: Heat violations detected. Preload templates may be\n"
57 "incomplete or have errors."
61 def is_implementation_of(class_, base_class):
63 Returns True if the class is an implementation of AbstractPreloadGenerator
66 inspect.isclass(class_)
67 and not inspect.isabstract(class_)
68 and issubclass(class_, base_class)
72 def get_implementations_of(class_, modules):
74 Returns all classes that implement ``class_`` from modules
78 inspect.getmembers(mod, lambda c: is_implementation_of(c, class_))
82 return [m[1] for m in members]
87 self.preload_plugins = [
88 importlib.import_module(name)
89 for finder, name, ispkg in pkgutil.iter_modules()
90 if name.startswith("preload_") or name == "preload"
92 self.preload_generators: List[
93 Type[AbstractPreloadGenerator]
94 ] = get_implementations_of(AbstractPreloadGenerator, self.preload_plugins)
95 self.preload_sources: List[
96 Type[AbstractPreloadDataSource]
97 ] = get_implementations_of(AbstractPreloadDataSource, self.preload_plugins)
99 def get_source_for_id(self, identifier: str) -> Type[AbstractPreloadDataSource]:
100 for source in self.preload_sources:
101 if identifier == source.get_identifier():
104 "Unable to find preload source for identifier {}".format(identifier)
107 def get_source_for_name(self, name: str) -> Type[AbstractPreloadDataSource]:
108 for source in self.preload_sources:
109 if name == source.get_name():
111 raise RuntimeError("Unable to find preload source for name {}".format(name))
114 PLUGIN_MGR = PluginManager()