Mercurial > repos > shellac > guppy_basecaller
diff env/lib/python3.7/site-packages/gxformat2/lint.py @ 5:9b1c78e6ba9c draft default tip
"planemo upload commit 6c0a8142489327ece472c84e558c47da711a9142"
author | shellac |
---|---|
date | Mon, 01 Jun 2020 08:59:25 -0400 |
parents | 79f47841a781 |
children |
line wrap: on
line diff
--- a/env/lib/python3.7/site-packages/gxformat2/lint.py Thu May 14 16:47:39 2020 -0400 +++ /dev/null Thu Jan 01 00:00:00 1970 +0000 @@ -1,180 +0,0 @@ -"""Workflow linting entry point - main script.""" -import argparse -import os -import sys - -from gxformat2._scripts import ensure_format2 -from gxformat2._yaml import ordered_load -from gxformat2.linting import LintContext -from gxformat2.markdown_parse import validate_galaxy_markdown - -EXIT_CODE_SUCCESS = 0 -EXIT_CODE_LINT_FAILED = 1 -EXIT_CODE_FORMAT_ERROR = 2 -EXIT_CODE_FILE_PARSE_FAILED = 3 - -LINT_FAILED_NO_OUTPUTS = "Workflow contained no outputs" -LINT_FAILED_OUTPUT_NO_LABEL = "Workflow contained output without a label" - - -def ensure_key(lint_context, has_keys, key, has_class=None, has_value=None): - if key not in has_keys: - lint_context.error("expected to find key [{key}] but absent", key=key) - return None - - value = has_keys[key] - return ensure_key_has_value(lint_context, has_keys, key, value, has_class=has_class, has_value=has_value) - - -def ensure_key_if_present(lint_context, has_keys, key, default=None, has_class=None): - if key not in has_keys: - return default - - value = has_keys[key] - return ensure_key_has_value(lint_context, has_keys, key, value, has_class=has_class, has_value=None) - - -def ensure_key_has_value(lint_context, has_keys, key, value, has_class=None, has_value=None): - if has_class is not None and not isinstance(value, has_class): - lint_context.error("expected value [{value}] with key [{key}] to be of class {clazz}", key=key, value=value, clazz=has_class) - if has_value is not None and value != has_value: - lint_context.error("expected value [{value}] with key [{key}] to be {expected_value}", key=key, value=value, expected_value=has_value) - return value - - -def _lint_step_errors(lint_context, step): - step_errors = step.get("errors") - if step_errors is not None: - lint_context.warn("tool step contains error indicated during Galaxy export - %s" % step_errors) - - -def lint_ga(lint_context, workflow_dict, path=None): - """Lint a native/legacy style Galaxy workflow and populate the corresponding LintContext.""" - ensure_key(lint_context, workflow_dict, "format-version", has_value="0.1") - ensure_key(lint_context, workflow_dict, "a_galaxy_workflow", has_value="true") - - native_steps = ensure_key(lint_context, workflow_dict, "steps", has_class=dict) or {} - - found_outputs = False - found_output_without_label = False - for order_index_str, step in native_steps.items(): - if not order_index_str.isdigit(): - lint_context.error("expected step_key to be integer not [{value}]", value=order_index_str) - - workflow_outputs = ensure_key_if_present(lint_context, step, "workflow_outputs", default=[], has_class=list) - for workflow_output in workflow_outputs: - found_outputs = True - - if not workflow_output.get("label"): - found_output_without_label = True - - step_type = step.get("type") - if step_type == "subworkflow": - subworkflow = ensure_key(lint_context, step, "subworkflow", has_class=dict) - lint_ga(lint_context, subworkflow) - - _lint_step_errors(lint_context, step) - _lint_tool_if_present(lint_context, step) - - _validate_report(lint_context, workflow_dict) - if not found_outputs: - lint_context.warn(LINT_FAILED_NO_OUTPUTS) - - if found_output_without_label: - lint_context.warn(LINT_FAILED_OUTPUT_NO_LABEL) - - _lint_training(lint_context, workflow_dict) - - -def lint_format2(lint_context, workflow_dict, path=None): - """Lint a Format 2 Galaxy workflow and populate the corresponding LintContext.""" - from gxformat2.schema.v19_09 import load_document - from schema_salad.exceptions import SchemaSaladException - try: - load_document("file://" + os.path.normpath(path)) - except SchemaSaladException as e: - lint_context.error("Validation failed " + str(e)) - - steps = ensure_key_if_present(lint_context, workflow_dict, 'steps', default={}, has_class=dict) - for key, step in steps.items(): - _lint_step_errors(lint_context, step) - _lint_tool_if_present(lint_context, step) - - _validate_report(lint_context, workflow_dict) - _lint_training(lint_context, workflow_dict) - - -def _lint_tool_if_present(lint_context, step_dict): - tool_id = step_dict.get('tool_id') - if tool_id and 'testtoolshed' in tool_id: - lint_context.warn('Step references a tool from the test tool shed, this should be replaced with a production tool') - - -def _validate_report(lint_context, workflow_dict): - report_dict = ensure_key_if_present(lint_context, workflow_dict, "report", default=None, has_class=dict) - if report_dict is not None: - markdown = ensure_key(lint_context, report_dict, "markdown", has_class=str) - if isinstance(markdown, str): - try: - validate_galaxy_markdown(markdown) - except ValueError as e: - lint_context.error("Report markdown validation failed [%s]" % e) - - -def _lint_training(lint_context, workflow_dict): - if lint_context.training_topic is None: - return - - if "tags" not in workflow_dict: - lint_context.warn("Missing tag(s).") - else: - tags = workflow_dict["tags"] - if lint_context.training_topic not in tags: - lint_context.warn("Missing expected training topic (%s) as workflow tag." % lint_context.training_topic) - # Move up into individual lints - all workflows should have docs. - format2_dict = ensure_format2(workflow_dict) - if "doc" not in format2_dict: - lint_context.warn("Missing workflow documentation (annotation or doc element)") - elif not format2_dict["doc"]: - lint_context.warn("Empty workflow documentation (annotation or doc element)") - - -def main(argv=None): - """Script entry point for linting workflows.""" - if argv is None: - argv = sys.argv - args = _parser().parse_args(argv[1:]) - path = args.path - with open(path, "r") as f: - try: - workflow_dict = ordered_load(f) - except Exception: - return EXIT_CODE_FILE_PARSE_FAILED - workflow_class = workflow_dict.get("class") - lint_func = lint_format2 if workflow_class == "GalaxyWorkflow" else lint_ga - lint_context = LintContext(training_topic=args.training_topic) - lint_func(lint_context, workflow_dict, path=path) - lint_context.print_messages() - if lint_context.found_errors: - return EXIT_CODE_FORMAT_ERROR - elif lint_context.found_warns: - return EXIT_CODE_LINT_FAILED - else: - return EXIT_CODE_SUCCESS - - -def _parser(): - parser = argparse.ArgumentParser() - parser.add_argument("--training-topic", - required=False, - help='If this is a training workflow, specify a training topic.') - parser.add_argument('path', metavar='PATH', type=str, - help='workflow path') - return parser - - -if __name__ == "__main__": - sys.exit(main()) - - -__all__ = ('main', 'lint_format2', 'lint_ga')