ice_validator/config.py

   1 import importlib
   2 import inspect
   3 import multiprocessing
   4 import os
   5 import pkgutil
   6 import queue
   7 from configparser import ConfigParser
   8 from itertools import chain
   9 from pathlib import Path
  10 from typing import MutableMapping, Iterator, List, Optional, Dict
  11
  12 import appdirs
  13 import yaml
  14 from cached_property import cached_property
  15
  16 from version import VERSION
  17 from preload.generator import AbstractPreloadGenerator
  18 from tests.test_environment_file_parameters import ENV_PARAMETER_SPEC
  19
  20 PATH = os.path.dirname(os.path.realpath(__file__))
  21 PROTOCOLS = ("http:", "https:", "file:")
  22
  23
  24 def to_uri(path):
  25     if any(path.startswith(p) for p in PROTOCOLS):
  26         return path
  27     return Path(path).absolute().as_uri()
  28
  29
  30 class UserSettings(MutableMapping):
  31     FILE_NAME = "UserSettings.ini"
  32
  33     def __init__(self, namespace, owner):
  34         user_config_dir = appdirs.AppDirs(namespace, owner).user_config_dir
  35         if not os.path.exists(user_config_dir):
  36             os.makedirs(user_config_dir, exist_ok=True)
  37         self._settings_path = os.path.join(user_config_dir, self.FILE_NAME)
  38         self._config = ConfigParser()
  39         self._config.read(self._settings_path)
  40
  41     def __getitem__(self, k):
  42         return self._config["DEFAULT"][k]
  43
  44     def __setitem__(self, k, v) -> None:
  45         self._config["DEFAULT"][k] = v
  46
  47     def __delitem__(self, v) -> None:
  48         del self._config["DEFAULT"][v]
  49
  50     def __len__(self) -> int:
  51         return len(self._config["DEFAULT"])
  52
  53     def __iter__(self) -> Iterator:
  54         return iter(self._config["DEFAULT"])
  55
  56     def save(self):
  57         with open(self._settings_path, "w") as f:
  58             self._config.write(f)
  59
  60
  61 class Config:
  62     """
  63     Configuration for the Validation GUI Application
  64
  65     Attributes
  66     ----------
  67     ``log_queue``       Queue for the ``stdout`` and ``stderr` of
  68                         the background job
  69     ``log_file``        File-like object (write only!) that writes to
  70                         the ``log_queue``
  71     ``status_queue``    Job completion status of the background job is
  72                         posted here as a tuple of (bool, Exception).
  73                         The first parameter is True if the job completed
  74                         successfully, and False otherwise.  If the job
  75                         failed, then an Exception will be provided as the
  76                         second element.
  77     ``command_queue``   Used to send commands to the GUI.  Currently only
  78                         used to send shutdown commands in tests.
  79     """
  80
  81     DEFAULT_FILENAME = "vvp-config.yaml"
  82     DEFAULT_POLLING_FREQUENCY = "1000"
  83
  84     def __init__(self, config: dict = None):
  85         """Creates instance of application configuration.
  86
  87         :param config: override default configuration if provided."""
  88         if config:
  89             self._config = config
  90         else:
  91             with open(self.DEFAULT_FILENAME, "r") as f:
  92                 self._config = yaml.safe_load(f)
  93         self._user_settings = UserSettings(
  94             self._config["namespace"], self._config["owner"]
  95         )
  96         self._watched_variables = []
  97         self._validate()
  98
  99     @cached_property
 100     def manager(self):
 101         return multiprocessing.Manager()
 102
 103     @cached_property
 104     def log_queue(self):
 105         return self.manager.Queue()
 106
 107     @cached_property
 108     def status_queue(self):
 109         return self.manager.Queue()
 110
 111     @cached_property
 112     def log_file(self):
 113         return QueueWriter(self.log_queue)
 114
 115     @cached_property
 116     def command_queue(self):
 117         return self.manager.Queue()
 118
 119     def watch(self, *variables):
 120         """Traces the variables and saves their settings for the user.  The
 121         last settings will be used where available"""
 122         self._watched_variables = variables
 123         for var in self._watched_variables:
 124             var.trace_add("write", self.save_settings)
 125
 126     # noinspection PyProtectedMember,PyUnusedLocal
 127     def save_settings(self, *args):
 128         """Save the value of all watched variables to user settings"""
 129         for var in self._watched_variables:
 130             self._user_settings[var._name] = str(var.get())
 131         self._user_settings.save()
 132
 133     @property
 134     def app_name(self) -> str:
 135         """Name of the application (displayed in title bar)"""
 136         app_name = self._config["ui"].get("app-name", "VNF Validation Tool")
 137         return "{} - {}".format(app_name, VERSION)
 138
 139     @property
 140     def category_names(self) -> List[str]:
 141         """List of validation profile names for display in the UI"""
 142         return [category["name"] for category in self._config["categories"]]
 143
 144     @property
 145     def polling_frequency(self) -> int:
 146         """Returns the frequency (in ms) the UI polls the queue communicating
 147         with any background job"""
 148         return int(
 149             self._config["settings"].get(
 150                 "polling-frequency", self.DEFAULT_POLLING_FREQUENCY
 151             )
 152         )
 153
 154     @property
 155     def disclaimer_text(self) -> str:
 156         return self._config["ui"].get("disclaimer-text", "")
 157
 158     @property
 159     def requirement_link_text(self) -> str:
 160         return self._config["ui"].get("requirement-link-text", "")
 161
 162     @property
 163     def requirement_link_url(self) -> str:
 164         path = self._config["ui"].get("requirement-link-url", "")
 165         return to_uri(path)
 166
 167     @property
 168     def terms(self) -> dict:
 169         return self._config.get("terms", {})
 170
 171     @property
 172     def terms_link_url(self) -> Optional[str]:
 173         path = self.terms.get("path")
 174         return to_uri(path) if path else None
 175
 176     @property
 177     def terms_link_text(self):
 178         return self.terms.get("popup-link-text")
 179
 180     @property
 181     def terms_version(self) -> Optional[str]:
 182         return self.terms.get("version")
 183
 184     @property
 185     def terms_popup_title(self) -> Optional[str]:
 186         return self.terms.get("popup-title")
 187
 188     @property
 189     def terms_popup_message(self) -> Optional[str]:
 190         return self.terms.get("popup-msg-text")
 191
 192     @property
 193     def are_terms_accepted(self) -> bool:
 194         version = "terms-{}".format(self.terms_version)
 195         return self._user_settings.get(version, "False") == "True"
 196
 197     def set_terms_accepted(self):
 198         version = "terms-{}".format(self.terms_version)
 199         self._user_settings[version] = "True"
 200         self._user_settings.save()
 201
 202     def get_description(self, category_name: str) -> str:
 203         """Returns the description associated with the category name"""
 204         return self._get_category(category_name)["description"]
 205
 206     def get_category(self, category_name: str) -> str:
 207         """Returns the category associated with the category name"""
 208         return self._get_category(category_name).get("category", "")
 209
 210     def get_category_value(self, category_name: str) -> str:
 211         """Returns the saved value for a category name"""
 212         return self._user_settings.get(category_name, 0)
 213
 214     def _get_category(self, category_name: str) -> Dict[str, str]:
 215         """Returns the profile definition"""
 216         for category in self._config["categories"]:
 217             if category["name"] == category_name:
 218                 return category
 219         raise RuntimeError(
 220             "Unexpected error: No category found in vvp-config.yaml "
 221             "with a name of " + category_name
 222         )
 223
 224     @property
 225     def default_report_format(self):
 226         return self._user_settings.get("report_format", "HTML")
 227
 228     @property
 229     def default_create_preloads(self):
 230         return self._user_settings.get("create_preloads", 0)
 231
 232     @property
 233     def report_formats(self):
 234         return ["CSV", "Excel", "HTML"]
 235
 236     @property
 237     def preload_formats(self):
 238         excluded = self._config.get("excluded-preloads", [])
 239         formats = (cls.format_name() for cls in get_generator_plugins())
 240         return [f for f in formats if f not in excluded]
 241
 242     @property
 243     def default_preload_format(self):
 244         default = self._user_settings.get("preload_format")
 245         if default and default in self.preload_formats:
 246             return default
 247         else:
 248             return self.preload_formats[0]
 249
 250     @staticmethod
 251     def get_subdir_for_preload(preload_format):
 252         for gen in get_generator_plugins():
 253             if gen.format_name() == preload_format:
 254                 return gen.output_sub_dir()
 255         return ""
 256
 257     @property
 258     def default_input_format(self):
 259         requested_default = self._user_settings.get("input_format") or self._config[
 260             "settings"
 261         ].get("default-input-format")
 262         if requested_default in self.input_formats:
 263             return requested_default
 264         else:
 265             return self.input_formats[0]
 266
 267     @property
 268     def input_formats(self):
 269         return ["Directory (Uncompressed)", "ZIP File"]
 270
 271     @property
 272     def default_halt_on_failure(self):
 273         setting = self._user_settings.get("halt_on_failure", "True")
 274         return setting.lower() == "true"
 275
 276     @property
 277     def env_specs(self):
 278         env_specs = self._config["settings"].get("env-specs")
 279         specs = []
 280         if not env_specs:
 281             return [ENV_PARAMETER_SPEC]
 282         for mod_path, attr in (s.rsplit(".", 1) for s in env_specs):
 283             module = importlib.import_module(mod_path)
 284             specs.append(getattr(module, attr))
 285         return specs
 286
 287     def _validate(self):
 288         """Ensures the config file is properly formatted"""
 289         categories = self._config["categories"]
 290
 291         # All profiles have required keys
 292         expected_keys = {"name", "description"}
 293         for category in categories:
 294             actual_keys = set(category.keys())
 295             missing_keys = expected_keys.difference(actual_keys)
 296             if missing_keys:
 297                 raise RuntimeError(
 298                     "Error in vvp-config.yaml file: "
 299                     "Required field missing in category. "
 300                     "Missing: {} "
 301                     "Categories: {}".format(",".join(missing_keys), category)
 302                 )
 303
 304
 305 class QueueWriter:
 306     """``stdout`` and ``stderr`` will be written to this queue by pytest, and
 307     pulled into the main GUI application"""
 308
 309     def __init__(self, log_queue: queue.Queue):
 310         """Writes data to the provided queue.
 311
 312         :param log_queue: the queue instance to write to.
 313         """
 314         self.queue = log_queue
 315
 316     def write(self, data: str):
 317         """Writes ``data`` to the queue """
 318         self.queue.put(data)
 319
 320     # noinspection PyMethodMayBeStatic
 321     def isatty(self) -> bool:
 322         """Always returns ``False``"""
 323         return False
 324
 325     def flush(self):
 326         """No operation method to satisfy file-like behavior"""
 327         pass
 328
 329
 330 def is_preload_generator(class_):
 331     """
 332     Returns True if the class is an implementation of AbstractPreloadGenerator
 333     """
 334     return (
 335         inspect.isclass(class_)
 336         and not inspect.isabstract(class_)
 337         and issubclass(class_, AbstractPreloadGenerator)
 338     )
 339
 340
 341 def get_generator_plugins():
 342     """
 343     Scan the system path for modules that are preload plugins and discover
 344     and return the classes that implement AbstractPreloadGenerator in those
 345     modules
 346     """
 347     preload_plugins = (
 348         importlib.import_module(name)
 349         for finder, name, ispkg in pkgutil.iter_modules()
 350         if name.startswith("preload_")
 351     )
 352     members = chain.from_iterable(
 353         inspect.getmembers(mod, is_preload_generator) for mod in preload_plugins
 354     )
 355     return [m[1] for m in members]
 356
 357
 358 def get_generator_plugin_names():
 359     return [g.format_name() for g in get_generator_plugins()]