ice_validator/preload/engine.py

   1 import importlib
   2 import inspect
   3 import os
   4 import pkgutil
   5 import shutil
   6 from itertools import chain
   7 from pathlib import Path
   8 from typing import List, Type
   9
  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
  14
  15
  16 def create_preloads(config, exitstatus):
  17     """
  18     Create preloads in every format that can be discovered by get_generator_plugins
  19     """
  20     if config.getoption("self_test"):
  21         return
  22     print("+===================================================================+")
  23     print("|                      Preload Template Generation                  |")
  24     print("+===================================================================+")
  25
  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
  32     preload_source = None
  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")
  37         )
  38         preload_source = source_class(preload_source_path)
  39
  40     heat_templates = get_heat_templates(config)
  41     vnf = None
  42     for plugin_class in plugins.preload_generators:
  43         if plugin_class.format_name() not in selected_formats:
  44             continue
  45         vnf = Vnf(heat_templates)
  46         generator = plugin_class(vnf, preload_dir, preload_source)
  47         generator.generate()
  48     if vnf and vnf.uses_contrail:
  49         print(
  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."
  53         )
  54     if exitstatus != 0:
  55         print(
  56             "\nWARNING: Heat violations detected. Preload templates may be\n"
  57             "incomplete or have errors."
  58         )
  59
  60
  61 def is_implementation_of(class_, base_class):
  62     """
  63     Returns True if the class is an implementation of AbstractPreloadGenerator
  64     """
  65     return (
  66         inspect.isclass(class_)
  67         and not inspect.isabstract(class_)
  68         and issubclass(class_, base_class)
  69     )
  70
  71
  72 def get_implementations_of(class_, modules):
  73     """
  74     Returns all classes that implement ``class_`` from modules
  75     """
  76     members = list(
  77         chain.from_iterable(
  78             inspect.getmembers(mod, lambda c: is_implementation_of(c, class_))
  79             for mod in modules
  80         )
  81     )
  82     return [m[1] for m in members]
  83
  84
  85 class PluginManager:
  86     def __init__(self):
  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"
  91         ]
  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)
  98
  99     def get_source_for_id(self, identifier: str) -> Type[AbstractPreloadDataSource]:
 100         for source in self.preload_sources:
 101             if identifier == source.get_identifier():
 102                 return source
 103         raise RuntimeError(
 104             "Unable to find preload source for identifier {}".format(identifier)
 105         )
 106
 107     def get_source_for_name(self, name: str) -> Type[AbstractPreloadDataSource]:
 108         for source in self.preload_sources:
 109             if name == source.get_name():
 110                 return source
 111         raise RuntimeError("Unable to find preload source for name {}".format(name))
 112
 113
 114 PLUGIN_MGR = PluginManager()