7 from configparser import ConfigParser
8 from itertools import chain
9 from pathlib import Path
10 from typing import MutableMapping, Iterator, List, Optional, Dict
14 from cached_property import cached_property
16 from version import VERSION
17 from preload.generator import AbstractPreloadGenerator
18 from tests.test_environment_file_parameters import ENV_PARAMETER_SPEC
20 PATH = os.path.dirname(os.path.realpath(__file__))
21 PROTOCOLS = ("http:", "https:", "file:")
25 if any(path.startswith(p) for p in PROTOCOLS):
27 return Path(path).absolute().as_uri()
30 class UserSettings(MutableMapping):
31 FILE_NAME = "UserSettings.ini"
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)
41 def __getitem__(self, k):
42 return self._config["DEFAULT"][k]
44 def __setitem__(self, k, v) -> None:
45 self._config["DEFAULT"][k] = v
47 def __delitem__(self, v) -> None:
48 del self._config["DEFAULT"][v]
50 def __len__(self) -> int:
51 return len(self._config["DEFAULT"])
53 def __iter__(self) -> Iterator:
54 return iter(self._config["DEFAULT"])
57 with open(self._settings_path, "w") as f:
63 Configuration for the Validation GUI Application
67 ``log_queue`` Queue for the ``stdout`` and ``stderr` of
69 ``log_file`` File-like object (write only!) that writes to
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
77 ``command_queue`` Used to send commands to the GUI. Currently only
78 used to send shutdown commands in tests.
81 DEFAULT_FILENAME = "vvp-config.yaml"
82 DEFAULT_POLLING_FREQUENCY = "1000"
84 def __init__(self, config: dict = None):
85 """Creates instance of application configuration.
87 :param config: override default configuration if provided."""
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"]
96 self._watched_variables = []
101 return multiprocessing.Manager()
105 return self.manager.Queue()
108 def status_queue(self):
109 return self.manager.Queue()
113 return QueueWriter(self.log_queue)
116 def command_queue(self):
117 return self.manager.Queue()
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)
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()
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)
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"]]
145 def polling_frequency(self) -> int:
146 """Returns the frequency (in ms) the UI polls the queue communicating
147 with any background job"""
149 self._config["settings"].get(
150 "polling-frequency", self.DEFAULT_POLLING_FREQUENCY
155 def disclaimer_text(self) -> str:
156 return self._config["ui"].get("disclaimer-text", "")
159 def requirement_link_text(self) -> str:
160 return self._config["ui"].get("requirement-link-text", "")
163 def requirement_link_url(self) -> str:
164 path = self._config["ui"].get("requirement-link-url", "")
168 def terms(self) -> dict:
169 return self._config.get("terms", {})
172 def terms_link_url(self) -> Optional[str]:
173 path = self.terms.get("path")
174 return to_uri(path) if path else None
177 def terms_link_text(self):
178 return self.terms.get("popup-link-text")
181 def terms_version(self) -> Optional[str]:
182 return self.terms.get("version")
185 def terms_popup_title(self) -> Optional[str]:
186 return self.terms.get("popup-title")
189 def terms_popup_message(self) -> Optional[str]:
190 return self.terms.get("popup-msg-text")
193 def are_terms_accepted(self) -> bool:
194 version = "terms-{}".format(self.terms_version)
195 return self._user_settings.get(version, "False") == "True"
197 def set_terms_accepted(self):
198 version = "terms-{}".format(self.terms_version)
199 self._user_settings[version] = "True"
200 self._user_settings.save()
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"]
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", "")
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)
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:
220 "Unexpected error: No category found in vvp-config.yaml "
221 "with a name of " + category_name
225 def default_report_format(self):
226 return self._user_settings.get("report_format", "HTML")
229 def default_create_preloads(self):
230 return self._user_settings.get("create_preloads", 0)
233 def report_formats(self):
234 return ["CSV", "Excel", "HTML"]
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]
243 def default_preload_format(self):
244 default = self._user_settings.get("preload_format")
245 if default and default in self.preload_formats:
248 return self.preload_formats[0]
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()
258 def default_input_format(self):
259 requested_default = self._user_settings.get("input_format") or self._config[
261 ].get("default-input-format")
262 if requested_default in self.input_formats:
263 return requested_default
265 return self.input_formats[0]
268 def input_formats(self):
269 return ["Directory (Uncompressed)", "ZIP File"]
272 def default_halt_on_failure(self):
273 setting = self._user_settings.get("halt_on_failure", "True")
274 return setting.lower() == "true"
278 env_specs = self._config["settings"].get("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))
288 """Ensures the config file is properly formatted"""
289 categories = self._config["categories"]
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)
298 "Error in vvp-config.yaml file: "
299 "Required field missing in category. "
301 "Categories: {}".format(",".join(missing_keys), category)
306 """``stdout`` and ``stderr`` will be written to this queue by pytest, and
307 pulled into the main GUI application"""
309 def __init__(self, log_queue: queue.Queue):
310 """Writes data to the provided queue.
312 :param log_queue: the queue instance to write to.
314 self.queue = log_queue
316 def write(self, data: str):
317 """Writes ``data`` to the queue """
320 # noinspection PyMethodMayBeStatic
321 def isatty(self) -> bool:
322 """Always returns ``False``"""
326 """No operation method to satisfy file-like behavior"""
330 def is_preload_generator(class_):
332 Returns True if the class is an implementation of AbstractPreloadGenerator
335 inspect.isclass(class_)
336 and not inspect.isabstract(class_)
337 and issubclass(class_, AbstractPreloadGenerator)
341 def get_generator_plugins():
343 Scan the system path for modules that are preload plugins and discover
344 and return the classes that implement AbstractPreloadGenerator in those
348 importlib.import_module(name)
349 for finder, name, ispkg in pkgutil.iter_modules()
350 if name.startswith("preload_")
352 members = chain.from_iterable(
353 inspect.getmembers(mod, is_preload_generator) for mod in preload_plugins
355 return [m[1] for m in members]
358 def get_generator_plugin_names():
359 return [g.format_name() for g in get_generator_plugins()]