ice_validator/config.py

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