Mercurial > repos > shellac > guppy_basecaller

comparison env/lib/python3.7/site-packages/gxformat2/lint.py @ 0:26e78fe6e8c4 draft

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
"planemo upload commit c699937486c35866861690329de38ec1a5d9f783"
author shellac
date Sat, 02 May 2020 07:14:21 -0400
parents
children
comparison
equal deleted inserted replaced
-1:000000000000 0:26e78fe6e8c4
1 """Workflow linting entry point - main script."""
2 import argparse
3 import os
4 import sys
5
6 from gxformat2._scripts import ensure_format2
7 from gxformat2._yaml import ordered_load
8 from gxformat2.linting import LintContext
9 from gxformat2.markdown_parse import validate_galaxy_markdown
10
11 EXIT_CODE_SUCCESS = 0
12 EXIT_CODE_LINT_FAILED = 1
13 EXIT_CODE_FORMAT_ERROR = 2
14 EXIT_CODE_FILE_PARSE_FAILED = 3
15
16 LINT_FAILED_NO_OUTPUTS = "Workflow contained no outputs"
17 LINT_FAILED_OUTPUT_NO_LABEL = "Workflow contained output without a label"
18
19
20 def ensure_key(lint_context, has_keys, key, has_class=None, has_value=None):
21 if key not in has_keys:
22 lint_context.error("expected to find key [{key}] but absent", key=key)
23 return None
24
25 value = has_keys[key]
26 return ensure_key_has_value(lint_context, has_keys, key, value, has_class=has_class, has_value=has_value)
27
28
29 def ensure_key_if_present(lint_context, has_keys, key, default=None, has_class=None):
30 if key not in has_keys:
31 return default
32
33 value = has_keys[key]
34 return ensure_key_has_value(lint_context, has_keys, key, value, has_class=has_class, has_value=None)
35
36
37 def ensure_key_has_value(lint_context, has_keys, key, value, has_class=None, has_value=None):
38 if has_class is not None and not isinstance(value, has_class):
39 lint_context.error("expected value [{value}] with key [{key}] to be of class {clazz}", key=key, value=value, clazz=has_class)
40 if has_value is not None and value != has_value:
41 lint_context.error("expected value [{value}] with key [{key}] to be {expected_value}", key=key, value=value, expected_value=has_value)
42 return value
43
44
45 def _lint_step_errors(lint_context, step):
46 step_errors = step.get("errors")
47 if step_errors is not None:
48 lint_context.warn("tool step contains error indicated during Galaxy export - %s" % step_errors)
49
50
51 def lint_ga(lint_context, workflow_dict, path=None):
52 """Lint a native/legacy style Galaxy workflow and populate the corresponding LintContext."""
53 ensure_key(lint_context, workflow_dict, "format-version", has_value="0.1")
54 ensure_key(lint_context, workflow_dict, "a_galaxy_workflow", has_value="true")
55
56 native_steps = ensure_key(lint_context, workflow_dict, "steps", has_class=dict) or {}
57
58 found_outputs = False
59 found_output_without_label = False
60 for order_index_str, step in native_steps.items():
61 if not order_index_str.isdigit():
62 lint_context.error("expected step_key to be integer not [{value}]", value=order_index_str)
63
64 workflow_outputs = ensure_key_if_present(lint_context, step, "workflow_outputs", default=[], has_class=list)
65 for workflow_output in workflow_outputs:
66 found_outputs = True
67
68 if not workflow_output.get("label"):
69 found_output_without_label = True
70
71 step_type = step.get("type")
72 if step_type == "subworkflow":
73 subworkflow = ensure_key(lint_context, step, "subworkflow", has_class=dict)
74 lint_ga(lint_context, subworkflow)
75
76 _lint_step_errors(lint_context, step)
77 _lint_tool_if_present(lint_context, step)
78
79 _validate_report(lint_context, workflow_dict)
80 if not found_outputs:
81 lint_context.warn(LINT_FAILED_NO_OUTPUTS)
82
83 if found_output_without_label:
84 lint_context.warn(LINT_FAILED_OUTPUT_NO_LABEL)
85
86 _lint_training(lint_context, workflow_dict)
87
88
89 def lint_format2(lint_context, workflow_dict, path=None):
90 """Lint a Format 2 Galaxy workflow and populate the corresponding LintContext."""
91 from gxformat2.schema.v19_09 import load_document
92 from schema_salad.exceptions import SchemaSaladException
93 try:
94 load_document("file://" + os.path.normpath(path))
95 except SchemaSaladException as e:
96 lint_context.error("Validation failed " + str(e))
97
98 steps = ensure_key_if_present(lint_context, workflow_dict, 'steps', default={}, has_class=dict)
99 for key, step in steps.items():
100 _lint_step_errors(lint_context, step)
101 _lint_tool_if_present(lint_context, step)
102
103 _validate_report(lint_context, workflow_dict)
104 _lint_training(lint_context, workflow_dict)
105
106
107 def _lint_tool_if_present(lint_context, step_dict):
108 tool_id = step_dict.get('tool_id')
109 if tool_id and 'testtoolshed' in tool_id:
110 lint_context.warn('Step references a tool from the test tool shed, this should be replaced with a production tool')
111
112
113 def _validate_report(lint_context, workflow_dict):
114 report_dict = ensure_key_if_present(lint_context, workflow_dict, "report", default=None, has_class=dict)
115 if report_dict is not None:
116 markdown = ensure_key(lint_context, report_dict, "markdown", has_class=str)
117 if isinstance(markdown, str):
118 try:
119 validate_galaxy_markdown(markdown)
120 except ValueError as e:
121 lint_context.error("Report markdown validation failed [%s]" % e)
122
123
124 def _lint_training(lint_context, workflow_dict):
125 if lint_context.training_topic is None:
126 return
127
128 if "tags" not in workflow_dict:
129 lint_context.warn("Missing tag(s).")
130 else:
131 tags = workflow_dict["tags"]
132 if lint_context.training_topic not in tags:
133 lint_context.warn("Missing expected training topic (%s) as workflow tag." % lint_context.training_topic)
134 # Move up into individual lints - all workflows should have docs.
135 format2_dict = ensure_format2(workflow_dict)
136 if "doc" not in format2_dict:
137 lint_context.warn("Missing workflow documentation (annotation or doc element)")
138 elif not format2_dict["doc"]:
139 lint_context.warn("Empty workflow documentation (annotation or doc element)")
140
141
142 def main(argv=None):
143 """Script entry point for linting workflows."""
144 if argv is None:
145 argv = sys.argv
146 args = _parser().parse_args(argv[1:])
147 path = args.path
148 with open(path, "r") as f:
149 try:
150 workflow_dict = ordered_load(f)
151 except Exception:
152 return EXIT_CODE_FILE_PARSE_FAILED
153 workflow_class = workflow_dict.get("class")
154 lint_func = lint_format2 if workflow_class == "GalaxyWorkflow" else lint_ga
155 lint_context = LintContext(training_topic=args.training_topic)
156 lint_func(lint_context, workflow_dict, path=path)
157 lint_context.print_messages()
158 if lint_context.found_errors:
159 return EXIT_CODE_FORMAT_ERROR
160 elif lint_context.found_warns:
161 return EXIT_CODE_LINT_FAILED
162 else:
163 return EXIT_CODE_SUCCESS
164
165
166 def _parser():
167 parser = argparse.ArgumentParser()
168 parser.add_argument("--training-topic",
169 required=False,
170 help='If this is a training workflow, specify a training topic.')
171 parser.add_argument('path', metavar='PATH', type=str,
172 help='workflow path')
173 return parser
174
175
176 if __name__ == "__main__":
177 sys.exit(main())
178
179
180 __all__ = ('main', 'lint_format2', 'lint_ga')