NOTE: This script does require Python 3.6+
"""
-import appdirs
+
import os
+import traceback
+
import pytest
import version
-import yaml
import contextlib
import multiprocessing
import queue
import platform
import subprocess # nosec
-from collections import MutableMapping
-from configparser import ConfigParser
from multiprocessing import Queue
from pathlib import Path
from shutil import rmtree
NORMAL,
)
from tkinter.scrolledtext import ScrolledText
-from typing import Optional, List, Dict, TextIO, Callable, Iterator
+from typing import Optional, TextIO, Callable
-import preload
+from config import Config
VERSION = version.VERSION
PATH = os.path.dirname(os.path.realpath(__file__))
return
-class QueueWriter:
- """``stdout`` and ``stderr`` will be written to this queue by pytest, and
- pulled into the main GUI application"""
-
- def __init__(self, log_queue: queue.Queue):
- """Writes data to the provided queue.
-
- :param log_queue: the queue instance to write to.
- """
- self.queue = log_queue
-
- def write(self, data: str):
- """Writes ``data`` to the queue """
- self.queue.put(data)
-
- # noinspection PyMethodMayBeStatic
- def isatty(self) -> bool:
- """Always returns ``False``"""
- return False
-
- def flush(self):
- """No operation method to satisfy file-like behavior"""
- pass
-
-
def run_pytest(
template_dir: str,
log: TextIO,
result_queue: Queue,
categories: Optional[list],
- verbosity: str,
report_format: str,
halt_on_failure: bool,
template_source: str,
+ env_dir: str,
+ preload_format: list,
):
"""Runs pytest using the given ``profile`` in a background process. All
``stdout`` and ``stderr`` are redirected to ``log``. The result of the job
will collect and execute all tests that are
decorated with any of the passed categories, as
well as tests not decorated with a category.
- :param verbosity: Flag to be passed to pytest to control verbosity.
- Options are '' (empty string), '-v' (verbose),
- '-vv' (more verbose).
:param report_format: Determines the style of report written. Options are
csv, html, or excel
:param halt_on_failure: Determines if validation will halt when basic failures
prevent a large number of errors from flooding the
report.
:param template_source: The path or name of the template to show on the report
+ :param env_dir: Optional directory of env files that can be used
+ to generate populated preload templates
+ :param preload_format: Selected preload format
"""
out_path = "{}/{}".format(PATH, OUT_DIR)
if os.path.exists(out_path):
args = [
"--ignore=app_tests",
"--capture=sys",
- verbosity,
"--template-directory={}".format(template_dir),
"--report-format={}".format(report_format),
"--template-source={}".format(template_source),
]
+ if env_dir:
+ args.append("--env-directory={}".format(env_dir))
if categories:
for category in categories:
args.extend(("--category", category))
if not halt_on_failure:
args.append("--continue-on-failure")
+ if preload_format:
+ args.append("--preload-format={}".format(preload_format))
pytest.main(args=args)
result_queue.put((True, None))
- except Exception as e:
- result_queue.put((False, e))
-
-
-class UserSettings(MutableMapping):
- FILE_NAME = "UserSettings.ini"
-
- def __init__(self, namespace, owner):
- user_config_dir = appdirs.AppDirs(namespace, owner).user_config_dir
- if not os.path.exists(user_config_dir):
- os.makedirs(user_config_dir, exist_ok=True)
- self._settings_path = os.path.join(user_config_dir, self.FILE_NAME)
- self._config = ConfigParser()
- self._config.read(self._settings_path)
-
- def __getitem__(self, k):
- return self._config["DEFAULT"][k]
-
- def __setitem__(self, k, v) -> None:
- self._config["DEFAULT"][k] = v
-
- def __delitem__(self, v) -> None:
- del self._config["DEFAULT"][v]
-
- def __len__(self) -> int:
- return len(self._config["DEFAULT"])
-
- def __iter__(self) -> Iterator:
- return iter(self._config["DEFAULT"])
-
- def save(self):
- with open(self._settings_path, "w") as f:
- self._config.write(f)
-
-
-class Config:
- """
- Configuration for the Validation GUI Application
-
- Attributes
- ----------
- ``log_queue`` Queue for the ``stdout`` and ``stderr` of
- the background job
- ``log_file`` File-like object (write only!) that writes to
- the ``log_queue``
- ``status_queue`` Job completion status of the background job is
- posted here as a tuple of (bool, Exception).
- The first parameter is True if the job completed
- successfully, and False otherwise. If the job
- failed, then an Exception will be provided as the
- second element.
- ``command_queue`` Used to send commands to the GUI. Currently only
- used to send shutdown commands in tests.
- """
-
- DEFAULT_FILENAME = "vvp-config.yaml"
- DEFAULT_POLLING_FREQUENCY = "1000"
-
- def __init__(self, config: dict = None):
- """Creates instance of application configuration.
-
- :param config: override default configuration if provided."""
- if config:
- self._config = config
- else:
- with open(self.DEFAULT_FILENAME, "r") as f:
- self._config = yaml.safe_load(f)
- self._user_settings = UserSettings(
- self._config["namespace"], self._config["owner"]
- )
- self._watched_variables = []
- self._validate()
- self._manager = multiprocessing.Manager()
- self.log_queue = self._manager.Queue()
- self.status_queue = self._manager.Queue()
- self.log_file = QueueWriter(self.log_queue)
- self.command_queue = self._manager.Queue()
-
- def watch(self, *variables):
- """Traces the variables and saves their settings for the user. The
- last settings will be used where available"""
- self._watched_variables = variables
- for var in self._watched_variables:
- var.trace_add("write", self.save_settings)
-
- # noinspection PyProtectedMember,PyUnusedLocal
- def save_settings(self, *args):
- """Save the value of all watched variables to user settings"""
- for var in self._watched_variables:
- self._user_settings[var._name] = str(var.get())
- self._user_settings.save()
-
- @property
- def app_name(self) -> str:
- """Name of the application (displayed in title bar)"""
- app_name = self._config["ui"].get("app-name", "VNF Validation Tool")
- return "{} - {}".format(app_name, VERSION)
-
- @property
- def category_names(self) -> List[str]:
- """List of validation profile names for display in the UI"""
- return [category["name"] for category in self._config["categories"]]
-
- @property
- def polling_frequency(self) -> int:
- """Returns the frequency (in ms) the UI polls the queue communicating
- with any background job"""
- return int(
- self._config["settings"].get(
- "polling-frequency", self.DEFAULT_POLLING_FREQUENCY
- )
- )
-
- @property
- def disclaimer_text(self) -> str:
- return self._config["ui"].get("disclaimer-text", "")
-
- @property
- def requirement_link_text(self) -> str:
- return self._config["ui"].get("requirement-link-text", "")
-
- @property
- def requirement_link_url(self) -> str:
- path = self._config["ui"].get("requirement-link-url", "")
- if not path.startswith("http"):
- path = "file://{}".format(os.path.join(PATH, path))
- return path
-
- @property
- def terms(self) -> dict:
- return self._config.get("terms", {})
-
- @property
- def terms_link_url(self) -> Optional[str]:
- return self.terms.get("path")
-
- @property
- def terms_link_text(self):
- return self.terms.get("popup-link-text")
-
- @property
- def terms_version(self) -> Optional[str]:
- return self.terms.get("version")
-
- @property
- def terms_popup_title(self) -> Optional[str]:
- return self.terms.get("popup-title")
-
- @property
- def terms_popup_message(self) -> Optional[str]:
- return self.terms.get("popup-msg-text")
-
- @property
- def are_terms_accepted(self) -> bool:
- version = "terms-{}".format(self.terms_version)
- return self._user_settings.get(version, "False") == "True"
-
- def set_terms_accepted(self):
- version = "terms-{}".format(self.terms_version)
- self._user_settings[version] = "True"
- self._user_settings.save()
-
- def default_verbosity(self, levels: Dict[str, str]) -> str:
- requested_level = self._user_settings.get("verbosity") or self._config[
- "settings"
- ].get("default-verbosity", "Standard")
- keys = [key for key in levels]
- for key in levels:
- if key.lower().startswith(requested_level.lower()):
- return key
- raise RuntimeError(
- "Invalid default-verbosity level {}. Valid "
- "values are {}".format(requested_level, ", ".join(keys))
- )
-
- def get_description(self, category_name: str) -> str:
- """Returns the description associated with the category name"""
- return self._get_category(category_name)["description"]
-
- def get_category(self, category_name: str) -> str:
- """Returns the category associated with the category name"""
- return self._get_category(category_name).get("category", "")
-
- def get_category_value(self, category_name: str) -> str:
- """Returns the saved value for a category name"""
- return self._user_settings.get(category_name, 0)
-
- def _get_category(self, category_name: str) -> Dict[str, str]:
- """Returns the profile definition"""
- for category in self._config["categories"]:
- if category["name"] == category_name:
- return category
- raise RuntimeError(
- "Unexpected error: No category found in vvp-config.yaml "
- "with a name of " + category_name
- )
-
- @property
- def default_report_format(self):
- return self._user_settings.get("report_format", "HTML")
-
- @property
- def report_formats(self):
- return ["CSV", "Excel", "HTML"]
-
- @property
- def preload_formats(self):
- excluded = self._config.get("excluded-preloads", [])
- formats = (cls.format_name() for cls in preload.get_generator_plugins())
- return [f for f in formats if f not in excluded]
-
- @property
- def default_preload_format(self):
- default = self._user_settings.get("preload_format")
- if default and default in self.preload_formats:
- return default
- else:
- return self.preload_formats[0]
-
- @staticmethod
- def get_subdir_for_preload(preload_format):
- for gen in preload.get_generator_plugins():
- if gen.format_name() == preload_format:
- return gen.output_sub_dir()
- return ""
-
- @property
- def default_input_format(self):
- requested_default = self._user_settings.get("input_format") or self._config[
- "settings"
- ].get("default-input-format")
- if requested_default in self.input_formats:
- return requested_default
- else:
- return self.input_formats[0]
-
- @property
- def input_formats(self):
- return ["Directory (Uncompressed)", "ZIP File"]
-
- @property
- def default_halt_on_failure(self):
- setting = self._user_settings.get("halt_on_failure", "True")
- return setting.lower() == "true"
-
- def _validate(self):
- """Ensures the config file is properly formatted"""
- categories = self._config["categories"]
-
- # All profiles have required keys
- expected_keys = {"name", "description"}
- for category in categories:
- actual_keys = set(category.keys())
- missing_keys = expected_keys.difference(actual_keys)
- if missing_keys:
- raise RuntimeError(
- "Error in vvp-config.yaml file: "
- "Required field missing in category. "
- "Missing: {} "
- "Categories: {}".format(",".join(missing_keys), category)
- )
-
-
-def validate():
- return True
+ except Exception:
+ result_queue.put((False, traceback.format_exc()))
class Dialog(Toplevel):
# noinspection PyUnusedLocal
def ok(self, event=None):
- if not validate():
- self.initial_focus.focus_set() # put focus back
- return
self.withdraw()
self.update_idletasks()
self.apply()
class ValidatorApp:
- VERBOSITY_LEVELS = {"Less": "", "Standard (-v)": "-v", "More (-vv)": "-vv"}
-
def __init__(self, config: Config = None):
"""Constructs the GUI element of the Validation Tool"""
self.task = None
)
actions = Frame(control_panel)
control_panel.add(actions)
- control_panel.paneconfigure(actions, minsize=250)
+ control_panel.paneconfigure(actions, minsize=350)
if self.config.disclaimer_text or self.config.requirement_link_text:
self.footer = self.create_footer(parent_frame)
settings_frame = LabelFrame(actions, text="Settings")
settings_row = 1
settings_frame.grid(row=3, column=1, columnspan=3, pady=10, sticky="we")
- verbosity_label = Label(settings_frame, text="Verbosity:")
- verbosity_label.grid(row=settings_row, column=1, sticky=W)
- self.verbosity = StringVar(self._root, name="verbosity")
- self.verbosity.set(self.config.default_verbosity(self.VERBOSITY_LEVELS))
- verbosity_menu = OptionMenu(
- settings_frame, self.verbosity, *tuple(self.VERBOSITY_LEVELS.keys())
- )
- verbosity_menu.config(width=25)
- verbosity_menu.grid(row=settings_row, column=2, columnspan=3, sticky=E, pady=5)
- settings_row += 1
if self.config.preload_formats:
preload_format_label = Label(settings_frame, text="Preload Template:")
self.halt_on_failure = BooleanVar(self._root, name="halt_on_failure")
self.halt_on_failure.set(self.config.default_halt_on_failure)
- halt_on_failure_label = Label(settings_frame, text="Halt on Basic Failures:")
- halt_on_failure_label.grid(row=settings_row, column=1, sticky=E, pady=5)
+ halt_on_failure_label = Label(
+ settings_frame, text="Halt on Basic Failures:", anchor=W, justify=LEFT
+ )
+ halt_on_failure_label.grid(row=settings_row, column=1, sticky=W, pady=5)
halt_checkbox = Checkbutton(
settings_frame, offvalue=False, onvalue=True, variable=self.halt_on_failure
)
halt_checkbox.grid(row=settings_row, column=2, columnspan=2, sticky=W, pady=5)
+ settings_row += 1
+
+ self.create_preloads = BooleanVar(self._root, name="create_preloads")
+ self.create_preloads.set(self.config.default_create_preloads)
+ create_preloads_label = Label(
+ settings_frame,
+ text="Create Preload from Env Files:",
+ anchor=W,
+ justify=LEFT,
+ )
+ create_preloads_label.grid(row=settings_row, column=1, sticky=W, pady=5)
+ create_preloads_checkbox = Checkbutton(
+ settings_frame,
+ offvalue=False,
+ onvalue=True,
+ variable=self.create_preloads,
+ command=self.set_env_dir_state,
+ )
+ create_preloads_checkbox.grid(
+ row=settings_row, column=2, columnspan=2, sticky=W, pady=5
+ )
directory_label = Label(actions, text="Template Location:")
directory_label.grid(row=4, column=1, pady=5, sticky=W)
directory_browse = Button(actions, text="...", command=self.ask_template_source)
directory_browse.grid(row=4, column=3, pady=5, sticky=W)
+ env_dir_label = Label(actions, text="Env Files:")
+ env_dir_label.grid(row=5, column=1, pady=5, sticky=W)
+ self.env_dir = StringVar(self._root, name="env_dir")
+ env_dir_state = NORMAL if self.create_preloads.get() else DISABLED
+ self.env_dir_entry = Entry(
+ actions, width=40, textvariable=self.env_dir, state=env_dir_state
+ )
+ self.env_dir_entry.grid(row=5, column=2, pady=5, sticky=W)
+ env_dir_browse = Button(actions, text="...", command=self.ask_env_dir_source)
+ env_dir_browse.grid(row=5, column=3, pady=5, sticky=W)
+
validate_button = Button(
- actions, text="Validate Templates", command=self.validate
+ actions, text="Process Templates", command=self.validate
)
- validate_button.grid(row=5, column=1, columnspan=2, pady=5)
+ validate_button.grid(row=6, column=1, columnspan=2, pady=5)
self.result_panel = Frame(actions)
# We'll add these labels now, and then make them visible when the run completes
self.result_label.bind("<Button-1>", self.open_report)
self.preload_label = Label(
- self.result_panel, text="View Preloads", fg="blue", cursor="hand2"
+ self.result_panel, text="View Preload Templates", fg="blue", cursor="hand2"
)
self.underline(self.preload_label)
self.preload_label.bind("<Button-1>", self.open_preloads)
- self.result_panel.grid(row=6, column=1, columnspan=2)
+ self.result_panel.grid(row=7, column=1, columnspan=2)
control_panel.pack(fill=BOTH, expand=1)
main_window.add(control_panel)
self.config.watch(
*self.categories,
- self.verbosity,
self.input_format,
self.report_format,
self.halt_on_failure,
+ self.preload_format,
+ self.create_preloads,
)
- if self.config.preload_formats:
- self.config.watch(self.preload_format)
self.schedule(self.execute_pollers)
if self.config.terms_link_text and not self.config.are_terms_accepted:
TermsAndConditionsDialog(parent_frame, self.config)
footer.pack(fill=BOTH, expand=True)
return footer
+ def set_env_dir_state(self):
+ state = NORMAL if self.create_preloads.get() else DISABLED
+ self.env_dir_entry.config(state=state)
+
def ask_template_source(self):
if self.input_format.get() == "ZIP File":
template_source = filedialog.askopenfilename(
template_source = filedialog.askdirectory()
self.template_source.set(template_source)
+ def ask_env_dir_source(self):
+ self.env_dir.set(filedialog.askdirectory())
+
def validate(self):
"""Run the pytest validations in a background process"""
if not self.delete_prior_report():
self.config.log_file,
self.config.status_queue,
self.categories_list(),
- self.VERBOSITY_LEVELS[self.verbosity.get()],
self.report_format.get().lower(),
self.halt_on_failure.get(),
self.template_source.get(),
+ self.env_dir.get(),
+ self.preload_format.get(),
),
)
self.task.daemon = True
# noinspection PyUnusedLocal
def open_report(self, event):
"""Open the report in the user's default browser"""
- webbrowser.open_new("file://{}".format(self.report_file_path))
+ path = Path(self.report_file_path).absolute().resolve().as_uri()
+ webbrowser.open_new(path)
def open_preloads(self, event):
"""Open the report in the user's default browser"""
Code Review / vvp / validation-scripts.git / blobdiff