2 # ============LICENSE_START====================================================
3 # org.onap.vvp/validation-scripts
4 # ===================================================================
5 # Copyright © 2019 AT&T Intellectual Property. All rights reserved.
6 # ===================================================================
8 # Unless otherwise specified, all software contained herein is licensed
9 # under the Apache License, Version 2.0 (the "License");
10 # you may not use this software except in compliance with the License.
11 # You may obtain a copy of the License at
13 # http://www.apache.org/licenses/LICENSE-2.0
15 # Unless required by applicable law or agreed to in writing, software
16 # distributed under the License is distributed on an "AS IS" BASIS,
17 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
18 # See the License for the specific language governing permissions and
19 # limitations under the License.
23 # Unless otherwise specified, all documentation contained herein is licensed
24 # under the Creative Commons License, Attribution 4.0 Intl. (the "License");
25 # you may not use this documentation except in compliance with the License.
26 # You may obtain a copy of the License at
28 # https://creativecommons.org/licenses/by/4.0/
30 # Unless required by applicable law or agreed to in writing, documentation
31 # distributed under the License is distributed on an "AS IS" BASIS,
32 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
33 # See the License for the specific language governing permissions and
34 # limitations under the License.
36 # ============LICENSE_END============================================
41 A GUI that wraps the pytest validations scripts.
43 To make an executable for windows execute the ``make_exe.bat`` to generate the
44 .exe and its associated files. The the necessary files will be written to the
45 ``dist/vvp/`` directory. This entire directory must be copied to the target machine.
47 NOTE: This script does require Python 3.6+
56 import multiprocessing
62 from collections import MutableMapping
63 from configparser import ConfigParser
64 from multiprocessing import Queue
65 from pathlib import Path
66 from shutil import rmtree
92 from tkinter.scrolledtext import ScrolledText
93 from typing import Optional, List, Dict, TextIO, Callable, Iterator
95 VERSION = version.VERSION
96 PATH = os.path.dirname(os.path.realpath(__file__))
100 class ToolTip(object):
102 create a tooltip for a given widget
105 def __init__(self, widget, text="widget info"):
106 self.waittime = 750 # miliseconds
107 self.wraplength = 180 # pixels
110 self.widget.bind("<Enter>", self.enter)
111 self.widget.bind("<Leave>", self.leave)
112 self.widget.bind("<ButtonPress>", self.leave)
116 def enter(self, event=None):
119 def leave(self, event=None):
125 self.id = self.widget.after(self.waittime, self.showtip)
127 def unschedule(self):
131 self.widget.after_cancel(id)
133 def showtip(self, event=None):
135 x, y, cx, cy = self.widget.bbox("insert")
136 x += self.widget.winfo_rootx() + 25
137 y += self.widget.winfo_rooty() + 20
138 # creates a toplevel window
139 self.tw = Toplevel(self.widget)
140 # Leaves only the label and removes the app window
141 self.tw.wm_overrideredirect(True)
142 self.tw.wm_geometry("+%d+%d" % (x, y))
147 background="#ffffff",
150 wraplength=self.wraplength,
162 """``stdout`` and ``stderr`` will be written to this queue by pytest, and
163 pulled into the main GUI application"""
165 def __init__(self, log_queue: queue.Queue):
166 """Writes data to the provided queue.
168 :param log_queue: the queue instance to write to.
170 self.queue = log_queue
172 def write(self, data: str):
173 """Writes ``data`` to the queue """
176 # noinspection PyMethodMayBeStatic
177 def isatty(self) -> bool:
178 """Always returns ``False``"""
182 """No operation method to satisfy file-like behavior"""
186 def get_plugins() -> Optional[List]:
187 """When running in a frozen bundle, plugins to be registered
188 explicitly. This method will return the required plugins to register
189 based on the run mode"""
190 if hasattr(sys, "frozen"):
191 import pytest_tap.plugin
193 return [pytest_tap.plugin]
202 categories: Optional[list],
205 halt_on_failure: bool,
206 template_source: str,
208 """Runs pytest using the given ``profile`` in a background process. All
209 ``stdout`` and ``stderr`` are redirected to ``log``. The result of the job
210 will be put on the ``completion_queue``
212 :param template_dir: The directory containing the files to be validated.
213 :param log: ` `stderr`` and ``stdout`` of the pytest job will be
215 :param result_queue: Completion status posted here. See :class:`Config`
216 for more information.
217 :param categories: list of optional categories. When provided, pytest
218 will collect and execute all tests that are
219 decorated with any of the passed categories, as
220 well as tests not decorated with a category.
221 :param verbosity: Flag to be passed to pytest to control verbosity.
222 Options are '' (empty string), '-v' (verbose),
223 '-vv' (more verbose).
224 :param report_format: Determines the style of report written. Options are
226 :param halt_on_failure: Determines if validation will halt when basic failures
227 are encountered in the input files. This can help
228 prevent a large number of errors from flooding the
231 out_path = "{}/{}".format(PATH, OUT_DIR)
232 if os.path.exists(out_path):
233 rmtree(out_path, ignore_errors=True)
234 with contextlib.redirect_stderr(log), contextlib.redirect_stdout(log):
237 "--ignore=app_tests",
240 "--template-directory={}".format(template_dir),
241 "--report-format={}".format(report_format),
242 "--template-source={}".format(template_source),
245 for category in categories:
246 args.extend(("--category", category))
247 if not halt_on_failure:
248 args.append("--continue-on-failure")
249 pytest.main(args=args, plugins=get_plugins())
250 result_queue.put((True, None))
251 except Exception as e:
252 result_queue.put((False, e))
255 class UserSettings(MutableMapping):
256 FILE_NAME = "UserSettings.ini"
259 user_config_dir = appdirs.AppDirs("org.onap.vvp", "ONAP").user_config_dir
260 if not os.path.exists(user_config_dir):
261 os.makedirs(user_config_dir, exist_ok=True)
262 self._settings_path = os.path.join(user_config_dir, self.FILE_NAME)
263 self._config = ConfigParser()
264 self._config.read(self._settings_path)
266 def __getitem__(self, k):
267 return self._config["DEFAULT"][k]
269 def __setitem__(self, k, v) -> None:
270 self._config["DEFAULT"][k] = v
272 def __delitem__(self, v) -> None:
273 del self._config["DEFAULT"][v]
275 def __len__(self) -> int:
276 return len(self._config["DEFAULT"])
278 def __iter__(self) -> Iterator:
279 return iter(self._config["DEFAULT"])
282 with open(self._settings_path, "w") as f:
283 self._config.write(f)
288 Configuration for the Validation GUI Application
292 ``log_queue`` Queue for the ``stdout`` and ``stderr` of
294 ``log_file`` File-like object (write only!) that writes to
296 ``status_queue`` Job completion status of the background job is
297 posted here as a tuple of (bool, Exception).
298 The first parameter is True if the job completed
299 successfully, and False otherwise. If the job
300 failed, then an Exception will be provided as the
302 ``command_queue`` Used to send commands to the GUI. Currently only
303 used to send shutdown commands in tests.
306 DEFAULT_FILENAME = "vvp-config.yaml"
307 DEFAULT_POLLING_FREQUENCY = "1000"
309 def __init__(self, config: dict = None):
310 """Creates instance of application configuration.
312 :param config: override default configuration if provided."""
314 self._config = config
316 with open(self.DEFAULT_FILENAME, "r") as f:
317 self._config = yaml.load(f)
318 self._user_settings = UserSettings()
319 self._watched_variables = []
321 self._manager = multiprocessing.Manager()
322 self.log_queue = self._manager.Queue()
323 self.status_queue = self._manager.Queue()
324 self.log_file = QueueWriter(self.log_queue)
325 self.command_queue = self._manager.Queue()
327 def watch(self, *variables):
328 """Traces the variables and saves their settings for the user. The
329 last settings will be used where available"""
330 self._watched_variables = variables
331 for var in self._watched_variables:
332 var.trace_add("write", self.save_settings)
334 def save_settings(self, *args):
335 """Save the value of all watched variables to user settings"""
336 for var in self._watched_variables:
337 self._user_settings[var._name] = str(var.get())
338 self._user_settings.save()
341 def app_name(self) -> str:
342 """Name of the application (displayed in title bar)"""
343 app_name = self._config["ui"].get("app-name", "VNF Validation Tool")
344 return "{} - {}".format(app_name, VERSION)
347 def category_names(self) -> List[str]:
348 """List of validation profile names for display in the UI"""
349 return [category["name"] for category in self._config["categories"]]
352 def polling_frequency(self) -> int:
353 """Returns the frequency (in ms) the UI polls the queue communicating
354 with any background job"""
356 self._config["settings"].get(
357 "polling-frequency", self.DEFAULT_POLLING_FREQUENCY
361 def default_verbosity(self, levels: Dict[str, str]) -> str:
362 requested_level = self._user_settings.get("verbosity") or self._config[
364 ].get("default-verbosity", "Standard")
365 keys = [key for key in levels]
367 if key.lower().startswith(requested_level.lower()):
370 "Invalid default-verbosity level {}. Valid"
371 "values are {}".format(requested_level, ", ".join(keys))
374 def get_description(self, category_name: str) -> str:
375 """Returns the description associated with the category name"""
376 return self._get_category(category_name)["description"]
378 def get_category(self, category_name: str) -> str:
379 """Returns the category associated with the category name"""
380 return self._get_category(category_name).get("category", "")
382 def get_category_value(self, category_name: str) -> str:
383 """Returns the saved value for a category name"""
384 return self._user_settings.get(category_name, 0)
386 def _get_category(self, category_name: str) -> Dict[str, str]:
387 """Returns the profile definition"""
388 for category in self._config["categories"]:
389 if category["name"] == category_name:
392 "Unexpected error: No category found in vvp-config.yaml "
393 "with a name of " + category_name
397 def default_report_format(self):
398 return self._user_settings.get("report_format", "HTML")
401 def report_formats(self):
402 return ["CSV", "Excel", "HTML"]
405 def default_input_format(self):
406 requested_default = self._user_settings.get("input_format") or self._config[
408 ].get("default-input-format")
409 if requested_default in self.input_formats:
410 return requested_default
412 return self.input_formats[0]
415 def input_formats(self):
416 return ["Directory (Uncompressed)", "ZIP File"]
419 def default_halt_on_failure(self):
420 setting = self._user_settings.get("halt_on_failure", "True")
421 return setting.lower() == "true"
424 """Ensures the config file is properly formatted"""
425 categories = self._config["categories"]
427 # All profiles have required keys
428 expected_keys = {"name", "description"}
429 for category in categories:
430 actual_keys = set(category.keys())
431 missing_keys = expected_keys.difference(actual_keys)
434 "Error in vvp-config.yaml file: "
435 "Required field missing in category. "
437 "Categories: {}".format(",".join(missing_keys), category)
442 VERBOSITY_LEVELS = {"Less": "", "Standard (-v)": "-v", "More (-vv)": "-vv"}
444 def __init__(self, config: Config = None):
445 """Constructs the GUI element of the Validation Tool"""
447 self.config = config or Config()
450 self._root.title(self.config.app_name)
451 self._root.protocol("WM_DELETE_WINDOW", self.shutdown)
453 main_window = PanedWindow(self._root)
454 main_window.pack(fill=BOTH, expand=1)
456 control_panel = PanedWindow(
457 main_window, orient=HORIZONTAL, sashpad=4, sashrelief=RAISED
459 actions = Frame(control_panel)
460 control_panel.add(actions)
461 control_panel.paneconfigure(actions, minsize=250)
464 number_of_categories = len(self.config.category_names)
465 category_frame = LabelFrame(actions, text="Additional Validation Categories:")
466 category_frame.grid(row=1, column=1, columnspan=3, pady=5, sticky="we")
470 for x in range(0, number_of_categories):
471 category_name = self.config.category_names[x]
472 category_value = IntVar(value=0)
473 category_value._name = "category_{}".format(category_name.replace(" ", "_"))
474 category_value.set(self.config.get_category_value(category_value._name))
475 self.categories.append(category_value)
476 category_checkbox = Checkbutton(
477 category_frame, text=category_name, variable=self.categories[x]
479 ToolTip(category_checkbox, self.config.get_description(category_name))
480 category_checkbox.grid(row=x + 1, column=1, columnspan=2, sticky="w")
482 settings_frame = LabelFrame(actions, text="Settings")
483 settings_frame.grid(row=3, column=1, columnspan=3, pady=10, sticky="we")
484 verbosity_label = Label(settings_frame, text="Verbosity:")
485 verbosity_label.grid(row=1, column=1, sticky=W)
486 self.verbosity = StringVar(self._root, name="verbosity")
487 self.verbosity.set(self.config.default_verbosity(self.VERBOSITY_LEVELS))
488 verbosity_menu = OptionMenu(
489 settings_frame, self.verbosity, *tuple(self.VERBOSITY_LEVELS.keys())
491 verbosity_menu.config(width=25)
492 verbosity_menu.grid(row=1, column=2, columnspan=3, sticky=E, pady=5)
494 report_format_label = Label(settings_frame, text="Report Format:")
495 report_format_label.grid(row=2, column=1, sticky=W)
496 self.report_format = StringVar(self._root, name="report_format")
497 self.report_format.set(self.config.default_report_format)
498 report_format_menu = OptionMenu(
499 settings_frame, self.report_format, *self.config.report_formats
501 report_format_menu.config(width=25)
502 report_format_menu.grid(row=2, column=2, columnspan=3, sticky=E, pady=5)
504 input_format_label = Label(settings_frame, text="Input Format:")
505 input_format_label.grid(row=3, column=1, sticky=W)
506 self.input_format = StringVar(self._root, name="input_format")
507 self.input_format.set(self.config.default_input_format)
508 input_format_menu = OptionMenu(
509 settings_frame, self.input_format, *self.config.input_formats
511 input_format_menu.config(width=25)
512 input_format_menu.grid(row=3, column=2, columnspan=3, sticky=E, pady=5)
514 self.halt_on_failure = BooleanVar(self._root, name="halt_on_failure")
515 self.halt_on_failure.set(self.config.default_halt_on_failure)
516 halt_on_failure_label = Label(settings_frame, text="Halt on Basic Failures:")
517 halt_on_failure_label.grid(row=4, column=1, sticky=E, pady=5)
518 halt_checkbox = Checkbutton(
519 settings_frame, offvalue=False, onvalue=True, variable=self.halt_on_failure
521 halt_checkbox.grid(row=4, column=2, columnspan=2, sticky=W, pady=5)
523 directory_label = Label(actions, text="Template Location:")
524 directory_label.grid(row=4, column=1, pady=5, sticky=W)
525 self.template_source = StringVar(self._root, name="template_source")
526 directory_entry = Entry(actions, width=40, textvariable=self.template_source)
527 directory_entry.grid(row=4, column=2, pady=5, sticky=W)
528 directory_browse = Button(actions, text="...", command=self.ask_template_source)
529 directory_browse.grid(row=4, column=3, pady=5, sticky=W)
531 validate = Button(actions, text="Validate Templates", command=self.validate)
532 validate.grid(row=5, column=1, columnspan=2, pady=5)
534 self.result_panel = Frame(actions)
535 # We'll add these labels now, and then make them visible when the run completes
536 self.completion_label = Label(self.result_panel, text="Validation Complete!")
537 self.result_label = Label(
538 self.result_panel, text="View Report", fg="blue", cursor="hand2"
540 self.underline(self.result_label)
541 self.result_label.bind("<Button-1>", self.open_report)
542 self.result_panel.grid(row=6, column=1, columnspan=2)
543 control_panel.pack(fill=BOTH, expand=1)
545 main_window.add(control_panel)
547 self.log_panel = ScrolledText(main_window, wrap=WORD, width=120, height=20)
548 self.log_panel.configure(font=font.Font(family="Courier New", size="11"))
549 self.log_panel.pack(fill=BOTH, expand=1)
551 main_window.add(self.log_panel)
553 # Briefly add the completion and result labels so the window size includes
555 self.completion_label.pack()
556 self.result_label.pack() # Show report link
557 self._root.after_idle(
559 self.completion_label.pack_forget(),
560 self.result_label.pack_forget(),
569 self.halt_on_failure,
571 self.schedule(self.execute_pollers)
573 def ask_template_source(self):
574 if self.input_format.get() == "ZIP File":
575 template_source = filedialog.askopenfilename(
576 title="Select Archive",
577 filetypes=(("ZIP Files", "*.zip"), ("All Files", "*")),
580 template_source = filedialog.askdirectory()
581 self.template_source.set(template_source)
584 """Run the pytest validations in a background process"""
585 if not self.delete_prior_report():
588 if not self.template_source.get():
589 self.ask_template_source()
591 template_dir = self.resolve_template_dir()
594 self.kill_background_task()
596 self.completion_label.pack_forget()
597 self.result_label.pack_forget()
598 self.task = multiprocessing.Process(
602 self.config.log_file,
603 self.config.status_queue,
604 self.categories_list(),
605 self.VERBOSITY_LEVELS[self.verbosity.get()],
606 self.report_format.get().lower(),
607 self.halt_on_failure.get(),
608 self.template_source.get(),
611 self.task.daemon = True
616 """Returns the text displayed in the title bar of the application"""
617 return self._root.title()
619 def execute_pollers(self):
620 """Call all methods that require periodic execution, and re-schedule
621 their execution for the next polling interval"""
624 self.poll_status_queue()
625 self.poll_command_queue()
627 self.schedule(self.execute_pollers)
631 """Yields values from the queue until empty"""
634 yield q.get(block=False)
638 def poll_command_queue(self):
639 """Picks up command strings from the commmand queue, and
640 dispatches it for execution. Only SHUTDOWN is supported
642 for command in self._drain_queue(self.config.command_queue):
643 if command == "SHUTDOWN":
646 def poll_status_queue(self):
647 """Checks for completion of the job, and then displays the View Report link
648 if it was successful or writes the exception to the ``log_panel`` if
650 for is_success, e in self._drain_queue(self.config.status_queue):
652 self.completion_label.pack()
653 self.result_label.pack() # Show report link
655 self.log_panel.insert(END, str(e))
657 def poll_log_file(self):
658 """Reads captured stdout and stderr from the log queue and writes it to the
660 for line in self._drain_queue(self.config.log_queue):
661 self.log_panel.insert(END, line)
662 self.log_panel.see(END)
664 def schedule(self, func: Callable):
665 """Schedule the callable ``func`` to be executed according to
666 the polling_frequency"""
667 self._root.after(self.config.polling_frequency, func)
670 """Removes all log entries from teh log panel"""
671 self.log_panel.delete("1.0", END)
673 def delete_prior_report(self) -> bool:
674 """Attempts to delete the current report, and pops up a warning message
675 to the user if it can't be deleted. This will force the user to
676 close the report before re-running the validation. Returns True if
677 the file was deleted or did not exist, or False otherwise"""
678 if not os.path.exists(self.report_file_path):
682 os.remove(self.report_file_path)
685 messagebox.showerror(
687 "Please close or rename the open report file before re-validating",
692 def report_file_path(self):
693 ext_mapping = {"csv": "csv", "html": "html", "excel": "xlsx"}
694 ext = ext_mapping.get(self.report_format.get().lower())
695 return os.path.join(PATH, OUT_DIR, "report.{}".format(ext))
697 def open_report(self, event):
698 """Open the report in the user's default browser"""
699 webbrowser.open_new("file://{}".format(self.report_file_path))
702 """Start the event loop of the application. This method does not return"""
703 self._root.mainloop()
706 def underline(label):
707 """Apply underline format to an existing label"""
708 f = font.Font(label, label.cget("font"))
709 f.configure(underline=True)
710 label.configure(font=f)
712 def kill_background_task(self):
713 if self.task and self.task.is_alive():
714 self.task.terminate()
715 for _ in self._drain_queue(self.config.log_queue):
719 """Shutdown the application"""
720 self.kill_background_task()
723 def check_template_source_is_valid(self):
724 """Verifies the value of template source exists and of valid type based
726 if not self.template_source.get():
728 template_path = Path(self.template_source.get())
730 if not template_path.exists():
731 messagebox.showerror(
733 "Input does not exist. Please provide a valid file or directory.",
737 if self.input_format.get() == "ZIP File":
738 if zipfile.is_zipfile(template_path):
741 messagebox.showerror(
742 "Error", "Expected ZIP file, but input is not a valid ZIP file"
746 if template_path.is_dir():
749 messagebox.showerror(
750 "Error", "Expected directory, but input is not a directory"
754 def resolve_template_dir(self) -> str:
755 """Extracts the zip file to a temporary directory if needed, otherwise
756 returns the directory supplied to template source. Returns empty string
757 if the template source isn't valid"""
758 if not self.check_template_source_is_valid():
760 if self.input_format.get() == "ZIP File":
761 temp_dir = tempfile.mkdtemp()
762 archive = zipfile.ZipFile(self.template_source.get())
763 archive.extractall(path=temp_dir)
766 return self.template_source.get()
768 def categories_list(self) -> list:
770 selected_categories = self.categories
771 for x in range(0, len(selected_categories)):
772 if selected_categories[x].get():
773 category = self.config.category_names[x]
774 categories.append(self.config.get_category(category))
778 if __name__ == "__main__":
779 multiprocessing.freeze_support() # needed for PyInstaller to work
780 ValidatorApp().start()