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 report_formats(self):
230 return ["CSV", "Excel", "HTML"]
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]
239 def default_preload_format(self):
240 default = self._user_settings.get("preload_format")
241 if default and default in self.preload_formats:
244 return self.preload_formats[0]
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()
254 def default_input_format(self):
255 requested_default = self._user_settings.get("input_format") or self._config[
257 ].get("default-input-format")
258 if requested_default in self.input_formats:
259 return requested_default
261 return self.input_formats[0]
264 def input_formats(self):
265 return ["Directory (Uncompressed)", "ZIP File"]
268 def default_halt_on_failure(self):
269 setting = self._user_settings.get("halt_on_failure", "True")
270 return setting.lower() == "true"
274 env_specs = self._config["settings"].get("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))
284 """Ensures the config file is properly formatted"""
285 categories = self._config["categories"]
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)
294 "Error in vvp-config.yaml file: "
295 "Required field missing in category. "
297 "Categories: {}".format(",".join(missing_keys), category)
302 """``stdout`` and ``stderr`` will be written to this queue by pytest, and
303 pulled into the main GUI application"""
305 def __init__(self, log_queue: queue.Queue):
306 """Writes data to the provided queue.
308 :param log_queue: the queue instance to write to.
310 self.queue = log_queue
312 def write(self, data: str):
313 """Writes ``data`` to the queue """
316 # noinspection PyMethodMayBeStatic
317 def isatty(self) -> bool:
318 """Always returns ``False``"""
322 """No operation method to satisfy file-like behavior"""
326 def is_preload_generator(class_):
328 Returns True if the class is an implementation of AbstractPreloadGenerator
331 inspect.isclass(class_)
332 and not inspect.isabstract(class_)
333 and issubclass(class_, AbstractPreloadGenerator)
337 def get_generator_plugins():
339 Scan the system path for modules that are preload plugins and discover
340 and return the classes that implement AbstractPreloadGenerator in those
344 importlib.import_module(name)
345 for finder, name, ispkg in pkgutil.iter_modules()
346 if name.startswith("preload_")
348 members = chain.from_iterable(
349 inspect.getmembers(mod, is_preload_generator) for mod in preload_plugins
351 return [m[1] for m in members]
354 def get_generator_plugin_names():
355 return [g.format_name() for g in get_generator_plugins()]