1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
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()
|
