Mercurial > repos > shellac > sam_consensus_v3

comparison env/lib/python3.9/site-packages/gxformat2/lint.py @ 0:4f3585e2f14b draft default tip

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