5 from configparser import ConfigParser
6 from pathlib import Path
7 from typing import MutableMapping, Iterator, List, Optional, Dict
11 from cached_property import cached_property
13 from preload.engine import PLUGIN_MGR
14 from version import VERSION
15 from tests.test_environment_file_parameters import ENV_PARAMETER_SPEC
17 PATH = os.path.dirname(os.path.realpath(__file__))
18 PROTOCOLS = ("http:", "https:", "file:")
22 if any(path.startswith(p) for p in PROTOCOLS):
24 return Path(path).absolute().as_uri()
27 class UserSettings(MutableMapping):
28 FILE_NAME = "UserSettings.ini"
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)
38 def __getitem__(self, k):
39 return self._config["DEFAULT"][k]
41 def __setitem__(self, k, v) -> None:
42 self._config["DEFAULT"][k] = v
44 def __delitem__(self, v) -> None:
45 del self._config["DEFAULT"][v]
47 def __len__(self) -> int:
48 return len(self._config["DEFAULT"])
50 def __iter__(self) -> Iterator:
51 return iter(self._config["DEFAULT"])
54 with open(self._settings_path, "w") as f:
60 Configuration for the Validation GUI Application
64 ``log_queue`` Queue for the ``stdout`` and ``stderr` of
66 ``log_file`` File-like object (write only!) that writes to
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
74 ``command_queue`` Used to send commands to the GUI. Currently only
75 used to send shutdown commands in tests.
78 DEFAULT_FILENAME = "vvp-config.yaml"
79 DEFAULT_POLLING_FREQUENCY = "1000"
81 def __init__(self, config: dict = None):
82 """Creates instance of application configuration.
84 :param config: override default configuration if provided."""
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"]
93 self._watched_variables = []
98 return multiprocessing.Manager()
102 return self.manager.Queue()
105 def status_queue(self):
106 return self.manager.Queue()
110 return QueueWriter(self.log_queue)
113 def command_queue(self):
114 return self.manager.Queue()
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)
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()
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)
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"]]
142 def polling_frequency(self) -> int:
143 """Returns the frequency (in ms) the UI polls the queue communicating
144 with any background job"""
146 self._config["settings"].get(
147 "polling-frequency", self.DEFAULT_POLLING_FREQUENCY
152 def disclaimer_text(self) -> str:
153 return self._config["ui"].get("disclaimer-text", "")
156 def requirement_link_text(self) -> str:
157 return self._config["ui"].get("requirement-link-text", "")
160 def requirement_link_url(self) -> str:
161 path = self._config["ui"].get("requirement-link-url", "")
165 def terms(self) -> dict:
166 return self._config.get("terms", {})
169 def terms_link_url(self) -> Optional[str]:
170 path = self.terms.get("path")
171 return to_uri(path) if path else None
174 def terms_link_text(self):
175 return self.terms.get("popup-link-text")
178 def terms_version(self) -> Optional[str]:
179 return self.terms.get("version")
182 def terms_popup_title(self) -> Optional[str]:
183 return self.terms.get("popup-title")
186 def terms_popup_message(self) -> Optional[str]:
187 return self.terms.get("popup-msg-text")
190 def are_terms_accepted(self) -> bool:
191 version = "terms-{}".format(self.terms_version)
192 return self._user_settings.get(version, "False") == "True"
194 def set_terms_accepted(self):
195 version = "terms-{}".format(self.terms_version)
196 self._user_settings[version] = "True"
197 self._user_settings.save()
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"]
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", "")
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)
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:
217 "Unexpected error: No category found in vvp-config.yaml "
218 "with a name of " + category_name
222 def default_report_format(self):
223 return self._user_settings.get("report_format", "HTML")
226 def default_create_preloads(self):
227 return self._user_settings.get("create_preloads", 0)
230 def report_formats(self):
231 return ["CSV", "Excel", "HTML"]
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]
240 def preload_source_types(self):
241 return [s.get_name() for s in PLUGIN_MGR.preload_sources]
244 def default_preload_format(self):
245 default = self._user_settings.get("preload_format")
246 if default and default in self.preload_formats:
249 return self.preload_formats[0]
252 def default_preload_source(self):
253 default = self._user_settings.get("preload_source")
254 if default and default in self.preload_source_types:
257 return self.preload_source_types[0]
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()
267 def default_input_format(self):
268 requested_default = self._user_settings.get("input_format") or self._config[
270 ].get("default-input-format")
271 if requested_default in self.input_formats:
272 return requested_default
274 return self.input_formats[0]
277 def input_formats(self):
278 return ["Directory (Uncompressed)", "ZIP File"]
281 def default_halt_on_failure(self):
282 setting = self._user_settings.get("halt_on_failure", "True")
283 return setting.lower() == "true"
287 env_specs = self._config["settings"].get("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))
297 """Ensures the config file is properly formatted"""
298 categories = self._config["categories"]
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)
307 "Error in vvp-config.yaml file: "
308 "Required field missing in category. "
310 "Categories: {}".format(",".join(missing_keys), category)
315 """``stdout`` and ``stderr`` will be written to this queue by pytest, and
316 pulled into the main GUI application"""
318 def __init__(self, log_queue: queue.Queue):
319 """Writes data to the provided queue.
321 :param log_queue: the queue instance to write to.
323 self.queue = log_queue
325 def write(self, data: str):
326 """Writes ``data`` to the queue """
329 # noinspection PyMethodMayBeStatic
330 def isatty(self) -> bool:
331 """Always returns ``False``"""
335 """No operation method to satisfy file-like behavior"""