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 report_formats(self):
 230         return ["CSV", "Excel", "HTML"]
 231
 232     @property
 233     def preload_formats(self):
 234         excluded = self._config.get("excluded-preloads", [])
 235         formats = (cls.format_name() for cls in get_generator_plugins())
 236         return [f for f in formats if f not in excluded]
 237
 238     @property
 239     def default_preload_format(self):
 240         default = self._user_settings.get("preload_format")
 241         if default and default in self.preload_formats:
 242             return default
 243         else:
 244             return self.preload_formats[0]
 245
 246     @staticmethod
 247     def get_subdir_for_preload(preload_format):
 248         for gen in get_generator_plugins():
 249             if gen.format_name() == preload_format:
 250                 return gen.output_sub_dir()
 251         return ""
 252
 253     @property
 254     def default_input_format(self):
 255         requested_default = self._user_settings.get("input_format") or self._config[
 256             "settings"
 257         ].get("default-input-format")
 258         if requested_default in self.input_formats:
 259             return requested_default
 260         else:
 261             return self.input_formats[0]
 262
 263     @property
 264     def input_formats(self):
 265         return ["Directory (Uncompressed)", "ZIP File"]
 266
 267     @property
 268     def default_halt_on_failure(self):
 269         setting = self._user_settings.get("halt_on_failure", "True")
 270         return setting.lower() == "true"
 271
 272     @property
 273     def env_specs(self):
 274         env_specs = self._config["settings"].get("env-specs")
 275         specs = []
 276         if not env_specs:
 277             return [ENV_PARAMETER_SPEC]
 278         for mod_path, attr in (s.rsplit(".", 1) for s in env_specs):
 279             module = importlib.import_module(mod_path)
 280             specs.append(getattr(module, attr))
 281         return specs
 282
 283     def _validate(self):
 284         """Ensures the config file is properly formatted"""
 285         categories = self._config["categories"]
 286
 287         # All profiles have required keys
 288         expected_keys = {"name", "description"}
 289         for category in categories:
 290             actual_keys = set(category.keys())
 291             missing_keys = expected_keys.difference(actual_keys)
 292             if missing_keys:
 293                 raise RuntimeError(
 294                     "Error in vvp-config.yaml file: "
 295                     "Required field missing in category. "
 296                     "Missing: {} "
 297                     "Categories: {}".format(",".join(missing_keys), category)
 298                 )
 299
 300
 301 class QueueWriter:
 302     """``stdout`` and ``stderr`` will be written to this queue by pytest, and
 303     pulled into the main GUI application"""
 304
 305     def __init__(self, log_queue: queue.Queue):
 306         """Writes data to the provided queue.
 307
 308         :param log_queue: the queue instance to write to.
 309         """
 310         self.queue = log_queue
 311
 312     def write(self, data: str):
 313         """Writes ``data`` to the queue """
 314         self.queue.put(data)
 315
 316     # noinspection PyMethodMayBeStatic
 317     def isatty(self) -> bool:
 318         """Always returns ``False``"""
 319         return False
 320
 321     def flush(self):
 322         """No operation method to satisfy file-like behavior"""
 323         pass
 324
 325
 326 def is_preload_generator(class_):
 327     """
 328     Returns True if the class is an implementation of AbstractPreloadGenerator
 329     """
 330     return (
 331         inspect.isclass(class_)
 332         and not inspect.isabstract(class_)
 333         and issubclass(class_, AbstractPreloadGenerator)
 334     )
 335
 336
 337 def get_generator_plugins():
 338     """
 339     Scan the system path for modules that are preload plugins and discover
 340     and return the classes that implement AbstractPreloadGenerator in those
 341     modules
 342     """
 343     preload_plugins = (
 344         importlib.import_module(name)
 345         for finder, name, ispkg in pkgutil.iter_modules()
 346         if name.startswith("preload_")
 347     )
 348     members = chain.from_iterable(
 349         inspect.getmembers(mod, is_preload_generator) for mod in preload_plugins
 350     )
 351     return [m[1] for m in members]
 352
 353
 354 def get_generator_plugin_names():
 355     return [g.format_name() for g in get_generator_plugins()]